[Python] How to Validate and Improve a Data Preprocessing Pipeline for an E-commerce Search...

I'm new to data engineering and was recently tasked with building a production-level data preprocessing pipeline with python and MySQL for a search engine in an e-commerce platform. The pipeline includes data quality and validation checks, but since my team and I are relatively new to this domain, we're unsure about the following:

1. How can we verify if our current pipeline is correctly implemented and meets production standards?
2. What additional improvements or best practices should we consider for better performance, scalability, and robustness?
3. Should we integrate scikit-learn into this pipeline for preprocessing, or is our current approach sufficient?

I have attached the code for the pipeline below. I would really appreciate any constructive feedback, critiques, or suggestions for improvement. Thanks in advance for your help!

# Database connection details
db_config = {
'user': 'user', # Replace with your database username
'password': 'password', # Replace with your database password
'host': 'host', # Replace with your database host
'port': 3306, # Replace with your database port
'database': 'mysql'
}

# Create connection string and engine
connection_string = f"mysql+pymysql://{db_config['user']}:{db_config['password']}@{db_config['host']}:{db_config['port']}/{db_config['database']}"
engine = create_engine(connection_string)

# SQL query
query = """
insert the SQL query here
"""

# Function: Fetch Data from Database with Chunking
def fetch_data(query, engine, chunksize=10000):
"""Fetches data from the database using the given SQL query in chunks."""
print("Connecting to database and fetching data...")
chunks = [] # List to store individual chunks
with engine.connect() as connection:
# Use pd.read_sql with chunksize to fetch data in smaller chunks
for chunk in pd.read_sql(query, connection, chunksize=chunksize):
chunks.append(chunk) # Append each chunk to the list
print("Data fetched successfully.")
# Concatenate all chunks into a single DataFrame
data = pd.concat(chunks, ignore_index=True)
return data

# Function: Preprocess Data
# optimize clean_text to regex pass for better performance -- 2
# and add check for outliers, statistical summary and basic data summaries, data type checks, null/missing value check and many others \
# which then should processed into a report before proceeding with preprocess changes -- 3
def analyze_data(df: pd.DataFrame) -> Dict:
"""Generate statistical summary and data quality report."""
stats = {
'missing_values': df.isnull().sum().to_dict(),
'dtypes': df.dtypes.to_dict(),
'outliers': {},
'unique_counts': df.nunique().to_dict(),
'memory_usage': df.memory_usage(deep=True).sum() / 1024**2 # MB
}

# Detect outliers in numeric columns using IQR method
for col in df.select_dtypes(include=['float64', 'int64']):
Q1 = df[col].quantile(0.25)
Q3 = df[col].quantile(0.75)
IQR = Q3 - Q1
outliers = df[(df[col] < Q1 - 1.5 * IQR) | (df[col] > Q3 + 1.5 * IQR)]
stats['outliers'][col] = len(outliers)

return stats

def clean_text(text: str) -> str:
"""Optimized text cleaning using compiled regex patterns."""
if pd.isnull(text):
return ""
patterns = {
'html': re.compile(r'<[^>]+>'),
'special_chars': re.compile(r'[^a-zA-Z0-9\s]'),
'whitespace': re.compile(r'\s+')
}
text = patterns['html'].sub('', text)
text = patterns['special_chars'].sub('', text)
text = patterns['whitespace'].sub(' ', text).strip()
return text.lower()

def preprocess_data(df: pd.DataFrame) -> pd.DataFrame:
"""Enhanced data preprocessing with quality checks and optimized cleaning."""
# Create working copy
df = df.copy()

# Generate initial data quality report
initial_stats = analyze_data(df)

# Clean text columns in one operation
text_columns = ['description', 'variant', 'store', 'store_description']
df[text_columns] = df[text_columns].apply(lambda x: x.apply(clean_text))

# Handle numeric nulls with column-specific strategies
numeric_fills = {
'price': df['price'].fillna(0)
}
# Handle categorical nulls with empty strings
var_fills = {
'country':df['country'].fillna("0"),
'state':df['state'].fillna("0"),
'city':df['city'].fillna("0")
}
df.fillna(numeric_fills, inplace=True)
df.fillna(var_fills, inplace=True)

# Check for invalid values
df = df[df['price'] > 0] # Ensure prices are greater than 0

# Convert text columns to lowercase
object_columns = df.select_dtypes(include=['object']).columns
df[object_columns] = df[object_columns].apply(lambda x: x.str.lower())

# Generate final data quality report
final_stats = analyze_data(df)

# Add processing metadata
df.attrs['preprocessing_stats'] = {
'initial': initial_stats,
'final': final_stats,
'rows_processed': len(df),
'columns_processed': len(df.columns)
}

return df

def get_preprocessing_summary(df: pd.DataFrame) -> Dict:
"""Return preprocessing statistics and data quality metrics."""
if 'preprocessing_stats' not in df.attrs:
return {'error': 'No preprocessing statistics available'}
return df.attrs['preprocessing_stats']

# Function: Separate and Save Products and Services
def preprocess_and_separate(df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
"""
Optimized preprocessing and separation of data into products, services, and shops.
Returns tuple of (products_df, services_df, shops_df)
"""
# Configuration dictionaries
config = {
'products': {
'mask': df['digital'] == 0,
'id_col': 'product_id',
'name_col': 'product_name'
},
'services': {
'mask': df['digital'] == 1,
'id_col': 'service_id',
'name_col': 'service_name'
}
}

# Common columns to drop
drop_cols = ['digital', 'user_id', 'discount', 'class', 'store_id', 'store_description']

# Common column renames
base_rename = {
'unique_id': None, # Will be updated dynamically
'display_name': None, # Will be updated dynamically
'category_names': 'category',
'subcategory_names': 'subcategory',
'subsubcategory_name': 'subsubcategory'
}

# Process products and services in a single loop
processed_dfs = {}
for df_type, params in config.items():
# Update dynamic rename mappings
rename_map = base_rename.copy()
rename_map['unique_id'] = params['id_col']
rename_map['display_name'] = params['name_col']

# Process dataframe
processed_dfs[df_type] = (
df[params['mask']]
.drop(columns=drop_cols)
.rename(columns=rename_map)
.reset_index(drop=True)
)

# Process shops
shop_columns = {
'store': 'first',
'store_id': 'first',
'variant': lambda x: ', '.join(filter(bool, x)),
'store_description': 'first',
'supplier': 'first',
'country': 'first',
'state': 'first',
'city': 'first',
'business_source': 'first'
}

# Create product/service name columns efficiently
name_df = pd.DataFrame({
'product_name': df.apply(lambda x: x['display_name'] if x['digital'] == 0 else "", axis=1),
'service_name': df.apply(lambda x: x['display_name'] if x['digital'] == 1 else "", axis=1)
})

# Add name columns to aggregation config
shop_columns.update({
'product_name': lambda x: ', '.join(filter(bool, x)),
'service_name': lambda x: ', '.join(filter(bool, x))
})

# Process shops dataframe
shops_df = (pd.concat([df[list(shop_columns.keys())], name_df], axis=1)
.groupby('store', as_index=False)
.agg(shop_columns))

# Add metadata to all dataframes
dfs = {
'products': processed_dfs['products'],
'services': processed_dfs['services'],
'shops': shops_df
}

for name, df_obj in dfs.items():
df_obj.attrs['preprocessing_info'] = {
'rows': len(df_obj),
'columns': list(df_obj.columns),
'memory_usage_mb': df_obj.memory_usage(deep=True).sum() / 1024**2
}
# Save each dataframe
filename = f"{'shop' if name == 'shops' else name + '_data'}.csv"
df_obj.to_csv(filename, index=False)
print(f"Saved {len(df_obj)} {name} to '{filename}'")

return tuple(dfs.values())

def get_processing_summary(dfs: Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]) -> Dict:
"""Returns summary statistics for the processed dataframes."""
return {
name: df.attrs.get('preprocessing_info', {})
for name, df in zip(['products', 'services', 'shops'], dfs)
}

# data quality checks
def data_quality_checks(df):
"""Checks for missing values, invalid data, and consistency in the dataframe."""
issues = []
critical_columns = ['display_name', 'price']
for col in critical_columns:
if df[col].isnull().any():
issues.append(f"Column '{col}' contains missing values.")

# Check for missing critical values
if df['display_name'].isnull().any():
issues.append("Missing values found in 'display_name' (product/service name).")
if df['price'].isnull().any() or (df['price'] <= 0).any():
issues.append("Invalid or missing values found in 'price'. Prices must be greater than 0.")

# Check for invalid data types
if not pd.api.types.is_numeric_dtype(df['price']):
issues.append("'price' column has invalid data type. Must be numeric.")

# Ensure no duplicate unique IDs
if df['unique_id'].duplicated().any():
issues.append("Duplicate values found in 'unique_id'. IDs must be unique.")

# Check if critical columns are empty after preprocessing
critical_columns = ['display_name', 'price']
for col in critical_columns:
if df[col].isnull().any():
issues.append(f"Column '{col}' contains missing values.")

return issues


def validation_checks(products_df, services_df, shop_df):
"""Performs domain-specific validation checks on the processed data."""
issues = []

# Validate products data
if products_df.empty:
issues.append("Products dataframe is empty.")
if (products_df['product_name'] == "").any():
issues.append("Products dataframe contains empty 'product_name'.")
if (products_df['price'] <= 0).any():
issues.append("Products dataframe contains invalid prices.")

# Validate services data
if services_df.empty:
issues.append("Services dataframe is empty.")
if (services_df['service_name'] == "").any():
issues.append("Services dataframe contains empty 'service_name'.")

# Validate shop data
if shop_df.empty:
issues.append("Shop dataframe is empty.")
if shop_df['store'].isnull().any():
issues.append("Shop dataframe contains missing 'store' values.")
if shop_df['store_id'].duplicated().any():
issues.append("Duplicate store IDs found in shop dataframe.")

return issues

# Main Workflow
if __name__ == "__main__":
# Step 1: Fetch data
raw_data = fetch_data(query, engine)

# Step 2: Preprocess data
print("\nStep 2: Preprocessing Data...")
clean_data = preprocess_data(raw_data)

# Step 3: Data Quality Checks on Preprocessed Data
print("\nStep 3: Running Data Quality Checks on Preprocessed Data...")
quality_issues = data_quality_checks(clean_data)
if quality_issues:
print("Data Quality Issues Identified:")
for issue in quality_issues:
print(f" - {issue}")
raise ValueError("Data quality issues must be resolved before proceeding.")

# Step 4: Separate and save products and services
print("\nStep 4: Separating and Saving Products and Services...")
products_data, services_data = separate_and_save(clean_data)

# Step 5: Add type column and save combined shop data
print("\nStep 5: Adding 'Type' Column and Saving Combined Shop Data...")
shop_data = add_type_and_save_shop(clean_data)

# Step 6: Validation Checks on Processed Data
print("\nStep 6: Running Validation Checks on Processed Data...")
validation_issues = validation_checks(products_data, services_data, shop_data)
if validation_issues:
print("Validation Issues Identified:")
for issue in validation_issues:
print(f" - {issue}")
raise ValueError("Validation issues must be resolved before finalizing output.")

print("\nPipeline completed successfully!")



I have implemented a basic data preprocessing pipeline that includes steps like data cleaning, feature transformation, and validation checks. Specifically, I tried:

Handling missing values using replacement with zeros. Normalizing categorical features to prepare data for the search engine. Applying validation checks such as checking for duplicates and incorrect data types. I expected the pipeline to be efficient, scalable, and suitable for a production environment, but I am unsure whether:

It follows best practices for a production-level system. The validation checks are sufficient, or if there are better alternatives. The current approach (without scikit-learn) is adequate, or if integrating it would improve preprocessing. Looking forward to feedback on whether this implementation is sound and how it can be optimized.

Continue reading...