Upload 4 files

app.py

@@ -1,28 +1,23 @@
 import gradio as gr
-from pyspark.sql import SparkSession
-import os
 import pandas as pd
+import os
 from datetime import datetime
 from clean import clean_data, get_numeric_columns
 from report import create_full_report, REPORT_DIR
 
-
 def clean_and_visualize(file, primary_key_column, progress=gr.Progress()):
- # Create a Spark session
- spark = SparkSession.builder.appName("DataCleaner").getOrCreate()
-
  # Read the CSV file
  progress(0.05, desc="Reading CSV file")
- df = spark.read.csv(file.name, header=True, inferSchema=True)
+ df = pd.read_csv(file.name)
 
  # Clean the data
  progress(0.1, desc="Starting data cleaning")
- cleaned_df, nonconforming_cells_before, process_times = clean_data(spark, df, primary_key_column, progress)
+ cleaned_df, nonconforming_cells_before, process_times = clean_data(df, primary_key_column, progress)
  progress(0.8, desc="Data cleaning completed")
 
  # Calculate removed columns and rows
  removed_columns = len(df.columns) - len(cleaned_df.columns)
- removed_rows = df.count() - cleaned_df.count()
+ removed_rows = len(df) - len(cleaned_df)
 
  # Generate full visualization report
  progress(0.9, desc="Generating report")
@@ -36,26 +31,18 @@ def clean_and_visualize(file, primary_key_column, progress=gr.Progress()):
  primary_key_column
  )
 
- # Convert PySpark DataFrame to Pandas DataFrame and save as CSV
+ # Save cleaned data as CSV
  progress(0.95, desc="Saving cleaned data")
- pandas_df = cleaned_df.toPandas()
-
- # Generate cleaned CSV file name with current date and time
  current_time = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
  cleaned_csv_path = os.path.join(f"cleaned_data_{current_time}.csv")
-
- pandas_df.to_csv(cleaned_csv_path, index=False)
+ cleaned_df.to_csv(cleaned_csv_path, index=False)
 
  # Collect all generated images
  image_files = [os.path.join(REPORT_DIR, f) for f in os.listdir(REPORT_DIR) if f.endswith('.png')]
 
- # Stop the Spark session
- spark.stop()
-
  progress(1.0, desc="Process completed")
  return cleaned_csv_path, image_files
 
-
 def launch_app():
  with gr.Blocks() as app:
  gr.Markdown("# Data Cleaner")
@@ -91,10 +78,8 @@ def launch_app():
  if file is None:
  return gr.Dropdown(choices=[])
 
- spark = SparkSession.builder.appName("DataCleaner").getOrCreate()
- df = spark.read.csv(file.name, header=True, inferSchema=True)
+ df = pd.read_csv(file.name)
  numeric_columns = get_numeric_columns(df)
- spark.stop()
 
  return gr.Dropdown(choices=numeric_columns)
 
@@ -119,6 +104,5 @@ def launch_app():
 
  app.launch()
 
-
 if __name__ == "__main__":
  launch_app()

clean.py

@@ -1,12 +1,8 @@
 import re
-
-from pyspark.sql import SparkSession
-from pyspark.sql.functions import col, isnan, when, count, lower, regexp_replace, to_date, to_timestamp, udf, \
- levenshtein, array, lit, trim, size, coalesce
-from pyspark.sql.types import DoubleType, IntegerType, StringType, DateType, TimestampType, ArrayType
-from pyspark.sql.utils import AnalysisException
-import time
+import pandas as pd
+import numpy as np
 from time import perf_counter
+import time
 
 # Constants
 EMPTY_THRESHOLD = 0.5
@@ -15,261 +11,183 @@ VALID_DATA_THRESHOLD = 0.5
 
 def print_dataframe_info(df, step=""):
  num_columns = len(df.columns)
- num_rows = df.count()
+ num_rows = len(df)
  num_cells = num_columns * num_rows
  print(f"{step}Dataframe info:")
  print(f" Number of columns: {num_columns}")
  print(f" Number of rows: {num_rows}")
  print(f" Total number of cells: {num_cells}")
 
-
 def check_and_normalize_column_headers(df):
  print("Checking and normalizing column headers...")
-
- for old_name in df.columns:
- # Create the new name using string manipulation
- new_name = old_name.lower().replace(' ', '_')
-
- # Remove any non-alphanumeric characters (excluding underscores)
- new_name = re.sub(r'[^0-9a-zA-Z_]', '', new_name)
-
- # Rename the column
- df = df.withColumnRenamed(old_name, new_name)
-
+ df.columns = df.columns.str.lower().str.replace(' ', '_')
+ df.columns = [re.sub(r'[^0-9a-zA-Z_]', '', col) for col in df.columns]
  print("Column names have been normalized.")
  return df
 
-
 def remove_empty_columns(df, threshold=EMPTY_THRESHOLD):
  print(f"Removing columns with less than {threshold * 100}% valid data...")
-
- # Calculate the percentage of non-null values for each column
- df_stats = df.select(
- [((count(when(col(c).isNotNull(), c)) / count('*')) >= threshold).alias(c) for c in df.columns])
- valid_columns = [c for c in df_stats.columns if df_stats.select(c).first()[0]]
-
- return df.select(valid_columns)
-
+ return df.dropna(axis=1, thresh=int(threshold * len(df)))
 
 def remove_empty_rows(df, threshold=EMPTY_THRESHOLD):
  print(f"Removing rows with less than {threshold * 100}% valid data...")
-
- # Count the number of non-null values for each row
- expr = sum([when(col(c).isNotNull(), lit(1)).otherwise(lit(0)) for c in df.columns])
- df_valid_count = df.withColumn('valid_count', expr)
-
- # Filter rows based on the threshold
- total_columns = len(df.columns)
- df_filtered = df_valid_count.filter(col('valid_count') >= threshold * total_columns)
-
- print('count of valid rows:', df_filtered.count())
-
- return df_filtered.drop('valid_count')
-
+ return df.dropna(thresh=int(threshold * len(df.columns)))
 
 def drop_rows_with_nas(df, threshold=VALID_DATA_THRESHOLD):
  print(f"Dropping rows with NAs for columns with more than {threshold * 100}% valid data...")
-
- # Calculate the percentage of non-null values for each column
- df_stats = df.select([((count(when(col(c).isNotNull(), c)) / count('*'))).alias(c) for c in df.columns])
-
- # Get columns with more than threshold valid data
- valid_columns = [c for c in df_stats.columns if df_stats.select(c).first()[0] > threshold]
-
- # Drop rows with NAs only for the valid columns
- for column in valid_columns:
- df = df.filter(col(column).isNotNull())
-
- return df
+ valid_columns = df.columns[df.notna().mean() > threshold]
+ return df.dropna(subset=valid_columns)
 
 def check_typos(df, column_name, threshold=2, top_n=100):
- # Check if the column is of StringType
- if not isinstance(df.schema[column_name].dataType, StringType):
+ if df[column_name].dtype != 'object':
  print(f"Skipping typo check for column {column_name} as it is not a string type.")
  return None
 
  print(f"Checking for typos in column: {column_name}")
 
  try:
- # Get value counts for the specific column
- value_counts = df.groupBy(column_name).count().orderBy("count", ascending=False)
-
- # Take top N most frequent values
- top_values = [row[column_name] for row in value_counts.limit(top_n).collect()]
+ value_counts = df[column_name].value_counts()
+ top_values = value_counts.head(top_n).index.tolist()
 
- # Broadcast the top values to all nodes
- broadcast_top_values = df.sparkSession.sparkContext.broadcast(top_values)
-
- # Define UDF to find similar strings
- @udf(returnType=ArrayType(StringType()))
  def find_similar_strings(value):
- if value is None:
+ if pd.isna(value):
  return []
- similar = []
- for top_value in broadcast_top_values.value:
- if value != top_value and levenshtein(value, top_value) <= threshold:
- similar.append(top_value)
- return similar
-
- # Apply the UDF to the column
- df_with_typos = df.withColumn("possible_typos", find_similar_strings(col(column_name)))
+ return [tv for tv in top_values if value != tv and levenshtein_distance(value, tv) <= threshold]
 
- # Filter rows with possible typos and select only the relevant columns
- typos_df = df_with_typos.filter(size("possible_typos") > 0).select(column_name, "possible_typos")
+ df['possible_typos'] = df[column_name].apply(find_similar_strings)
+ typos_df = df[df['possible_typos'].apply(len) > 0][[column_name, 'possible_typos']]
 
- # Check if there are any potential typos
- typo_count = typos_df.count()
+ typo_count = len(typos_df)
  if typo_count > 0:
  print(f"Potential typos found in column {column_name}: {typo_count}")
- typos_df.show(10, truncate=False)
+ print(typos_df.head(10))
  return typos_df
  else:
  print(f"No potential typos found in column {column_name}")
  return None
 
- except AnalysisException as e:
- print(f"Error analyzing column {column_name}: {str(e)}")
- return None
  except Exception as e:
  print(f"Unexpected error in check_typos for column {column_name}: {str(e)}")
  return None
 
+def levenshtein_distance(s1, s2):
+ if len(s1) < len(s2):
+ return levenshtein_distance(s2, s1)
+ if len(s2) == 0:
+ return len(s1)
+ previous_row = range(len(s2) + 1)
+ for i, c1 in enumerate(s1):
+ current_row = [i + 1]
+ for j, c2 in enumerate(s2):
+ insertions = previous_row[j + 1] + 1
+ deletions = current_row[j] + 1
+ substitutions = previous_row[j] + (c1 != c2)
+ current_row.append(min(insertions, deletions, substitutions))
+ previous_row = current_row
+ return previous_row[-1]
 
 def transform_string_column(df, column_name):
  print(f"Transforming string column: {column_name}")
- # Lower case transformation (if applicable)
- df = df.withColumn(column_name, lower(col(column_name)))
- # Remove leading and trailing spaces
- df = df.withColumn(column_name, trim(col(column_name)))
- # Replace multiple spaces with a single space
- df = df.withColumn(column_name, regexp_replace(col(column_name), "\\s+", " "))
- # Remove special characters except those used in dates and times
- df = df.withColumn(column_name, regexp_replace(col(column_name), "[^a-zA-Z0-9\\s/:.-]", ""))
+ df[column_name] = df[column_name].str.lower()
+ df[column_name] = df[column_name].str.strip()
+ df[column_name] = df[column_name].str.replace(r'\s+', ' ', regex=True)
+ df[column_name] = df[column_name].str.replace(r'[^a-zA-Z0-9\s/:.-]', '', regex=True)
  return df
 
-
 def clean_column(df, column_name):
  print(f"Cleaning column: {column_name}")
  start_time = perf_counter()
- # Get the data type of the current column
- column_type = df.schema[column_name].dataType
 
- if isinstance(column_type, StringType):
- # Skip date detection and directly process as string
- # For string columns, check for typos and transform
+ if df[column_name].dtype == 'object':
  typos_df = check_typos(df, column_name)
- if typos_df is not None and typos_df.count() > 0:
+ if typos_df is not None and len(typos_df) > 0:
  print(f"Detailed typos for column {column_name}:")
- typos_df.show(truncate=False)
+ print(typos_df)
  df = transform_string_column(df, column_name)
-
- elif isinstance(column_type, (DoubleType, IntegerType)):
- # For numeric columns, we'll do a simple null check
- df = df.withColumn(column_name, when(col(column_name).isNull(), lit(None)).otherwise(col(column_name)))
+ elif pd.api.types.is_numeric_dtype(df[column_name]):
+ df[column_name] = pd.to_numeric(df[column_name], errors='coerce')
 
  end_time = perf_counter()
  print(f"Time taken to clean {column_name}: {end_time - start_time:.6f} seconds")
  return df
 
-
-
-
-# Update the remove_outliers function to work on a single column
 def remove_outliers(df, column):
  print(f"Removing outliers from column: {column}")
-
- stats = df.select(column).summary("25%", "75%").collect()
- q1 = float(stats[0][1])
- q3 = float(stats[1][1])
+ q1 = df[column].quantile(0.25)
+ q3 = df[column].quantile(0.75)
  iqr = q3 - q1
  lower_bound = q1 - 1.5 * iqr
  upper_bound = q3 + 1.5 * iqr
- df = df.filter((col(column) >= lower_bound) & (col(column) <= upper_bound))
-
- return df
-
+ return df[(df[column] >= lower_bound) & (df[column] <= upper_bound)]
 
 def calculate_nonconforming_cells(df):
- nonconforming_cells = {}
- for column in df.columns:
- nonconforming_count = df.filter(col(column).isNull() | isnan(column)).count()
- nonconforming_cells[column] = nonconforming_count
- return nonconforming_cells
-
+ return df.isna().sum().to_dict()
 
 def get_numeric_columns(df):
- return [field.name for field in df.schema.fields if isinstance(field.dataType, (IntegerType, DoubleType))]
+ return df.select_dtypes(include=[np.number]).columns.tolist()
 
 def remove_duplicates_from_primary_key(df, primary_key_column):
  print(f"Removing duplicates based on primary key column: {primary_key_column}")
- return df.dropDuplicates([primary_key_column])
+ return df.drop_duplicates(subset=[primary_key_column])
 
-def clean_data(spark, df, primary_key_column, progress):
+def clean_data(df, primary_key_column, progress):
  start_time = time.time()
  process_times = {}
 
  print("Starting data validation and cleaning...")
  print_dataframe_info(df, "Initial - ")
 
- # Calculate nonconforming cells before cleaning
  nonconforming_cells_before = calculate_nonconforming_cells(df)
 
- # Step 1: Normalize column headers
  progress(0.1, desc="Normalizing column headers")
  step_start_time = time.time()
  df = check_and_normalize_column_headers(df)
  process_times['Normalize headers'] = time.time() - step_start_time
 
- # Step 2: Remove empty columns
  progress(0.2, desc="Removing empty columns")
  step_start_time = time.time()
  df = remove_empty_columns(df)
- print('2) count of valid rows:', df.count())
+ print('2) count of valid rows:', len(df))
  process_times['Remove empty columns'] = time.time() - step_start_time
 
- # Step 3: Remove empty rows
  progress(0.3, desc="Removing empty rows")
  step_start_time = time.time()
  df = remove_empty_rows(df)
- print('3) count of valid rows:', df.count())
+ print('3) count of valid rows:', len(df))
  process_times['Remove empty rows'] = time.time() - step_start_time
 
- # Step 4: Drop rows with NAs for columns with more than 50% valid data
  progress(0.4, desc="Dropping rows with NAs")
  step_start_time = time.time()
  df = drop_rows_with_nas(df)
- print('4) count of valid rows:', df.count())
+ print('4) count of valid rows:', len(df))
  process_times['Drop rows with NAs'] = time.time() - step_start_time
 
- # Step 5: Clean columns (including typo checking and string transformation)
  column_cleaning_times = {}
  total_columns = len(df.columns)
  for index, column in enumerate(df.columns):
  progress(0.5 + (0.2 * (index / total_columns)), desc=f"Cleaning column: {column}")
  column_start_time = time.time()
  df = clean_column(df, column)
- print('5) count of valid rows:', df.count())
+ print('5) count of valid rows:', len(df))
  column_cleaning_times[f"Clean column: {column}"] = time.time() - column_start_time
  process_times.update(column_cleaning_times)
 
- # Step 6: Remove outliers from numeric columns (excluding primary key)
  progress(0.7, desc="Removing outliers")
  step_start_time = time.time()
  numeric_columns = get_numeric_columns(df)
  numeric_columns = [col for col in numeric_columns if col != primary_key_column]
  for column in numeric_columns:
  df = remove_outliers(df, column)
- print('6) count of valid rows:', df.count())
+ print('6) count of valid rows:', len(df))
  process_times['Remove outliers'] = time.time() - step_start_time
 
- # Step 7: Remove duplicates from primary key column
  progress(0.8, desc="Removing duplicates from primary key")
  step_start_time = time.time()
  df = remove_duplicates_from_primary_key(df, primary_key_column)
- print('7) count of valid rows:', df.count())
+ print('7) count of valid rows:', len(df))
  process_times['Remove duplicates from primary key'] = time.time() - step_start_time
 
  print("Cleaning process completed.")
  print_dataframe_info(df, "Final - ")
 
- return df, nonconforming_cells_before, process_times
+ return df, nonconforming_cells_before, process_times

report.py

@@ -6,43 +6,28 @@ import seaborn as sns
 import matplotlib.pyplot as plt
 from datetime import datetime
 
-from pyspark.ml.feature import VectorAssembler
-from pyspark.ml.stat import Correlation
-from pyspark.sql.functions import col, count, when, lit, isnan
-from pyspark.sql.types import DoubleType, IntegerType, LongType, FloatType, StringType, DateType, TimestampType
-
 REPORT_DIR = f"cleaning_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
 os.makedirs(REPORT_DIR, exist_ok=True)
 
-
 def save_plot(fig, filename):
  fig.savefig(os.path.join(REPORT_DIR, filename), dpi=400, bbox_inches='tight')
  plt.close(fig)
 
-
 def plot_heatmap(df, title):
  # Calculate the percentage of null values for each column
- null_percentages = df.select([
- (100 * count(when(col(c).isNull() | isnan(c), c)) / count('*')).alias(c)
- for c in df.columns
- ]).toPandas()
+ null_percentages = df.isnull().mean() * 100
 
  plt.figure(figsize=(12, 8))
- sns.heatmap(null_percentages, cbar=True, cmap='Reds', annot=True, fmt='.1f')
+ sns.heatmap(null_percentages.to_frame().T, cbar=True, cmap='Reds', annot=True, fmt='.1f')
  plt.title(title)
  plt.ylabel('Percentage of Missing Values')
  plt.tight_layout()
  save_plot(plt.gcf(), f'{title.lower().replace(" ", "_")}.png')
 
-
 def plot_column_schemas(df):
  # Get the data types of all columns
- schema = df.schema
- data_types = []
- for field in schema.fields:
- dtype_name = field.dataType.typeName()
- print(f"Column '{field.name}' has data type '{dtype_name}'")
- data_types.append(dtype_name.capitalize())
+ data_types = df.dtypes.astype(str).tolist()
+ data_types = [dtype.capitalize() for dtype in data_types]
 
  # Count the occurrences of each data type
  type_counts = Counter(data_types)
@@ -70,7 +55,6 @@ def plot_column_schemas(df):
  plt.tight_layout()
  save_plot(fig, 'column_schemas.png')
 
-
 def plot_nonconforming_cells(nonconforming_cells):
  # Ensure that nonconforming_cells is a dictionary
  if isinstance(nonconforming_cells, dict):
@@ -99,16 +83,11 @@ def plot_nonconforming_cells(nonconforming_cells):
  else:
  print(f"Expected nonconforming_cells to be a dictionary, but got {type(nonconforming_cells)}.")
 
-
 def plot_column_distributions(cleaned_df, primary_key_column):
  print("Plotting distribution charts for numeric columns in the cleaned DataFrame...")
 
- def get_numeric_columns(df):
- return [field.name for field in df.schema.fields
- if isinstance(field.dataType, (IntegerType, LongType, FloatType, DoubleType))
- and field.name != primary_key_column]
-
- numeric_columns = get_numeric_columns(cleaned_df)
+ numeric_columns = cleaned_df.select_dtypes(include=[np.number]).columns.tolist()
+ numeric_columns = [col for col in numeric_columns if col != primary_key_column]
  num_columns = len(numeric_columns)
 
  if num_columns == 0:
@@ -122,8 +101,7 @@ def plot_column_distributions(cleaned_df, primary_key_column):
  axes = axes.flatten() if num_columns > 1 else [axes]
 
  for i, column in enumerate(numeric_columns):
- # Convert to pandas for plotting
- cleaned_data = cleaned_df.select(column).toPandas()[column].dropna()
+ cleaned_data = cleaned_df[column].dropna()
 
  sns.histplot(cleaned_data, ax=axes[i], kde=True, color='orange', label='After Cleaning', alpha=0.7)
  axes[i].set_title(f'{column} - Distribution After Cleaning')
@@ -136,16 +114,11 @@ def plot_column_distributions(cleaned_df, primary_key_column):
  plt.tight_layout()
  save_plot(fig, 'distributions_after_cleaning.png')
 
-
 def plot_boxplot_with_outliers(original_df, primary_key_column):
  print("Plotting boxplots for numeric columns in the original DataFrame...")
 
- def get_numeric_columns(df):
- return [field.name for field in df.schema.fields
- if isinstance(field.dataType, (IntegerType, LongType, FloatType, DoubleType))
- and field.name != primary_key_column]
-
- numeric_columns = get_numeric_columns(original_df)
+ numeric_columns = original_df.select_dtypes(include=[np.number]).columns.tolist()
+ numeric_columns = [col for col in numeric_columns if col != primary_key_column]
  num_columns = len(numeric_columns)
 
  if num_columns == 0:
@@ -159,8 +132,7 @@ def plot_boxplot_with_outliers(original_df, primary_key_column):
  axes = axes.flatten() if num_columns > 1 else [axes]
 
  for i, column in enumerate(numeric_columns):
- # Convert data to pandas for plotting
- data = original_df.select(column).toPandas()[column].dropna()
+ data = original_df[column].dropna()
 
  sns.boxplot(x=data, ax=axes[i], color='blue', orient='h')
  axes[i].set_title(f'Boxplot of {column} (Before Cleaning)')
@@ -172,36 +144,22 @@ def plot_boxplot_with_outliers(original_df, primary_key_column):
  plt.tight_layout()
  save_plot(fig, 'boxplots_before_cleaning.png')
 
-
 def plot_correlation_heatmap(df, primary_key_column):
- # Select only numeric columns
- numeric_columns = [field.name for field in df.schema.fields
- if isinstance(field.dataType, (IntegerType, LongType, FloatType, DoubleType))
- and field.name != primary_key_column]
+ numeric_columns = df.select_dtypes(include=[np.number]).columns.tolist()
+ numeric_columns = [col for col in numeric_columns if col != primary_key_column]
 
  if not numeric_columns:
  print("No numeric columns found for correlation heatmap.")
  return
 
- # Create a vector column of numeric columns
- assembler = VectorAssembler(inputCols=numeric_columns, outputCol="features")
- df_vector = assembler.transform(df).select("features")
-
- # Compute correlation matrix
- matrix = Correlation.corr(df_vector, "features").collect()[0][0]
- corr_matrix = matrix.toArray().tolist()
-
- # Convert to pandas DataFrame for plotting
- corr_df = pd.DataFrame(corr_matrix, columns=numeric_columns, index=numeric_columns)
+ corr_matrix = df[numeric_columns].corr()
 
- # Plot the heatmap
  plt.figure(figsize=(15, 10))
- sns.heatmap(corr_df, annot=True, fmt=".2f", cmap='coolwarm', cbar_kws={'label': 'Correlation'})
+ sns.heatmap(corr_matrix, annot=True, fmt=".2f", cmap='coolwarm', cbar_kws={'label': 'Correlation'})
  plt.title('Correlation Heatmap')
  plt.tight_layout()
  save_plot(plt.gcf(), 'correlation_heatmap.png')
 
-
 def plot_process_times(process_times):
  # Convert seconds to minutes
  process_times_minutes = {k: v / 60 for k, v in process_times.items()}
@@ -225,21 +183,20 @@ def plot_process_times(process_times):
  ax2.set_ylabel('Time (minutes)')
  ax2.tick_params(axis='x', rotation=90)
 
- # Add value labels on top of each bar
+ # Add value labels on top of each bar with 3 decimal places
  for ax, bars in zip([ax1, ax2], [bars1, bars2]):
  for bar in bars:
  height = bar.get_height()
  ax.text(bar.get_x() + bar.get_width() / 2., height,
- f'{height:.2f}', ha='center', va='bottom')
+ f'{height:.4f}', ha='center', va='bottom')
 
- # Add total time to the plot
+ # Add total time to the plot with 3 decimal places
  total_time = sum(process_times_minutes.values())
- fig.suptitle(f'Process Times (Total: {total_time:.2f} minutes)', fontsize=16)
+ fig.suptitle(f'Process Times (Total: {total_time:.3f} minutes)', fontsize=16)
 
  plt.tight_layout()
  save_plot(fig, 'process_times.png')
 
-
 def create_full_report(original_df, cleaned_df, nonconforming_cells_before, process_times, removed_columns,
  removed_rows, primary_key_column):
  os.makedirs(REPORT_DIR, exist_ok=True)

requirements.txt

@@ -2,5 +2,4 @@ numpy
 pandas
 seaborn
 matplotlib
-pyspark
 gradio