Data Cleaning & Transformation

← Previous: Data Ingestion & Sources | Back to Index | Next: Data Modeling & Warehousing →

Introduction

In my experience, data cleaning takes 60-80% of pipeline development time. Real-world data is messy: missing values, inconsistent formats, duplicates, and outliers. This article covers the cleaning patterns I use daily.

Common Data Quality Issues

Missing Data

# Python 3.12 - Handling missing data
import pandas as pd
import numpy as np
from typing import Optional

class MissingDataHandler:
    """
    Strategies for handling missing data.
    Choice depends on business context.
    """
    
    @staticmethod
    def analyze_missing(df: pd.DataFrame) -> pd.DataFrame:
        """
        Analyze missing data patterns.
        First step before deciding how to handle.
        """
        missing_stats = pd.DataFrame({
            'column': df.columns,
            'missing_count': df.isnull().sum(),
            'missing_pct': (df.isnull().sum() / len(df) * 100).round(2),
            'dtype': df.dtypes
        })
        
        return missing_stats[missing_stats['missing_count'] > 0].sort_values(
            'missing_pct',
            ascending=False
        )
    
    @staticmethod
    def drop_missing(
        df: pd.DataFrame,
        threshold: float = 0.5
    ) -> pd.DataFrame:
        """
        Drop columns with too many missing values.
        Drop rows with any missing critical fields.
        """
        # Drop columns with >50% missing
        df = df.loc[:, df.isnull().mean() < threshold]
        
        # Drop rows missing critical fields
        critical_cols = ['user_id', 'transaction_id', 'amount']
        df = df.dropna(subset=critical_cols)
        
        return df
    
    @staticmethod
    def fill_missing(df: pd.DataFrame) -> pd.DataFrame:
        """
        Fill missing values with appropriate strategies.
        """
        df = df.copy()
        
        # Numeric: fill with median (robust to outliers)
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        for col in numeric_cols:
            df[col].fillna(df[col].median(), inplace=True)
        
        # Categorical: fill with mode or 'unknown'
        categorical_cols = df.select_dtypes(include=['object']).columns
        for col in categorical_cols:
            mode_value = df[col].mode()[0] if not df[col].mode().empty else 'unknown'
            df[col].fillna(mode_value, inplace=True)
        
        # Boolean: fill with False
        bool_cols = df.select_dtypes(include=['bool']).columns
        for col in bool_cols:
            df[col].fillna(False, inplace=True)
        
        return df
    
    @staticmethod
    def forward_fill_time_series(
        df: pd.DataFrame,
        timestamp_col: str,
        value_col: str
    ) -> pd.DataFrame:
        """
        Forward fill for time series data.
        Use previous value when data is missing.
        """
        df = df.sort_values(timestamp_col)
        df[value_col] = df[value_col].fillna(method='ffill')
        return df

# Usage
handler = MissingDataHandler()
missing_report = handler.analyze_missing(df)
df_cleaned = handler.fill_missing(df)

Duplicate Records

# Python 3.12 - Handling duplicates
class DuplicateHandler:
    """
    Identify and remove duplicates.
    Duplicates are common in data from multiple sources.
    """
    
    @staticmethod
    def find_duplicates(
        df: pd.DataFrame,
        subset: list[str]
    ) -> pd.DataFrame:
        """Find duplicate records for investigation."""
        duplicates = df[df.duplicated(subset=subset, keep=False)]
        return duplicates.sort_values(subset)
    
    @staticmethod
    def remove_exact_duplicates(df: pd.DataFrame) -> pd.DataFrame:
        """Remove completely identical rows."""
        initial_count = len(df)
        df = df.drop_duplicates()
        removed = initial_count - len(df)
        
        logging.info(f"Removed {removed} exact duplicate rows")
        return df
    
    @staticmethod
    def remove_duplicates_keep_latest(
        df: pd.DataFrame,
        subset: list[str],
        timestamp_col: str
    ) -> pd.DataFrame:
        """
        Keep most recent record for each unique key.
        Common pattern for slowly changing dimensions.
        """
        df = df.sort_values(timestamp_col, ascending=False)
        df = df.drop_duplicates(subset=subset, keep='first')
        return df.sort_values(timestamp_col)
    
    @staticmethod
    def merge_duplicates(
        df: pd.DataFrame,
        key_cols: list[str],
        agg_dict: dict
    ) -> pd.DataFrame:
        """
        Merge duplicates by aggregating values.
        Example: sum amounts, take max timestamp.
        """
        return df.groupby(key_cols).agg(agg_dict).reset_index()

# Usage
# Find duplicates for review
dupes = DuplicateHandler.find_duplicates(df, subset=['user_id', 'date'])

# Remove keeping latest
df = DuplicateHandler.remove_duplicates_keep_latest(
    df,
    subset=['user_id', 'date'],
    timestamp_col='created_at'
)

Data Type Conversions

# Python 3.12 - Type conversions
class TypeConverter:
    """
    Safe type conversions with error handling.
    Prevents pipeline failures from bad data.
    """
    
    @staticmethod
    def convert_to_numeric(
        df: pd.DataFrame,
        columns: list[str],
        errors: str = 'coerce'
    ) -> pd.DataFrame:
        """
        Convert columns to numeric.
        errors='coerce' converts invalid values to NaN.
        """
        for col in columns:
            df[col] = pd.to_numeric(df[col], errors=errors)
        
        return df
    
    @staticmethod
    def convert_to_datetime(
        df: pd.DataFrame,
        columns: list[str],
        format: Optional[str] = None
    ) -> pd.DataFrame:
        """
        Convert to datetime with multiple format attempts.
        """
        for col in columns:
            try:
                df[col] = pd.to_datetime(df[col], format=format)
            except Exception:
                # Try without format (slower but more flexible)
                df[col] = pd.to_datetime(df[col], errors='coerce')
        
        return df
    
    @staticmethod
    def standardize_strings(
        df: pd.DataFrame,
        columns: list[str]
    ) -> pd.DataFrame:
        """
        Standardize string columns.
        Strip whitespace, lowercase, remove special chars.
        """
        for col in columns:
            df[col] = (
                df[col]
                .astype(str)
                .str.strip()
                .str.lower()
                .str.replace(r'[^\w\s]', '', regex=True)
            )
        
        return df
    
    @staticmethod
    def convert_boolean(
        df: pd.DataFrame,
        column: str,
        true_values: list = None,
        false_values: list = None
    ) -> pd.DataFrame:
        """
        Convert various formats to boolean.
        """
        if true_values is None:
            true_values = ['true', 't', 'yes', 'y', '1', 1]
        if false_values is None:
            false_values = ['false', 'f', 'no', 'n', '0', 0]
        
        df[column] = df[column].astype(str).str.lower()
        df[column] = df[column].apply(
            lambda x: True if x in map(str, true_values)
            else False if x in map(str, false_values)
            else None
        )
        
        return df

# Usage
df = TypeConverter.convert_to_numeric(df, ['amount', 'quantity'])
df = TypeConverter.convert_to_datetime(df, ['created_at', 'updated_at'])
df = TypeConverter.standardize_strings(df, ['email', 'country'])

Data Validation

# Python 3.12 - Comprehensive validation
from dataclasses import dataclass
from typing import Callable
from enum import Enum

class ValidationSeverity(Enum):
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

@dataclass
class ValidationRule:
    """Single validation rule."""
    name: str
    check_function: Callable
    severity: ValidationSeverity
    error_message: str

class DataValidator:
    """
    Production data validation framework.
    Catches data quality issues before they cause problems.
    """
    
    def __init__(self, df: pd.DataFrame):
        self.df = df
        self.violations = []
    
    def validate_not_null(
        self,
        columns: list[str],
        severity: ValidationSeverity = ValidationSeverity.ERROR
    ):
        """Validate no null values in critical columns."""
        for col in columns:
            null_count = self.df[col].isnull().sum()
            if null_count > 0:
                self.violations.append({
                    'rule': f'not_null_{col}',
                    'severity': severity.value,
                    'message': f'{null_count} null values in {col}',
                    'failed_count': null_count
                })
    
    def validate_unique(
        self,
        columns: list[str],
        severity: ValidationSeverity = ValidationSeverity.ERROR
    ):
        """Validate unique constraint."""
        duplicate_count = self.df.duplicated(subset=columns).sum()
        if duplicate_count > 0:
            self.violations.append({
                'rule': f'unique_{\"_\".join(columns)}',
                'severity': severity.value,
                'message': f'{duplicate_count} duplicate records',
                'failed_count': duplicate_count
            })
    
    def validate_range(
        self,
        column: str,
        min_value: float,
        max_value: float,
        severity: ValidationSeverity = ValidationSeverity.WARNING
    ):
        """Validate values within expected range."""
        out_of_range = (
            (self.df[column] < min_value) |
            (self.df[column] > max_value)
        ).sum()
        
        if out_of_range > 0:
            self.violations.append({
                'rule': f'range_{column}',
                'severity': severity.value,
                'message': f'{out_of_range} values outside [{min_value}, {max_value}]',
                'failed_count': out_of_range
            })
    
    def validate_format(
        self,
        column: str,
        pattern: str,
        severity: ValidationSeverity = ValidationSeverity.ERROR
    ):
        """Validate string format with regex."""
        import re
        invalid = ~self.df[column].astype(str).str.match(pattern)
        invalid_count = invalid.sum()
        
        if invalid_count > 0:
            self.violations.append({
                'rule': f'format_{column}',
                'severity': severity.value,
                'message': f'{invalid_count} invalid formats in {column}',
                'failed_count': invalid_count
            })
    
    def validate_referential_integrity(
        self,
        column: str,
        reference_values: set,
        severity: ValidationSeverity = ValidationSeverity.CRITICAL
    ):
        """Validate foreign key integrity."""
        invalid = ~self.df[column].isin(reference_values)
        invalid_count = invalid.sum()
        
        if invalid_count > 0:
            self.violations.append({
                'rule': f'referential_{column}',
                'severity': severity.value,
                'message': f'{invalid_count} invalid references in {column}',
                'failed_count': invalid_count
            })
    
    def get_report(self) -> pd.DataFrame:
        """Get validation report."""
        if not self.violations:
            logging.info("All validations passed!")
            return pd.DataFrame()
        
        return pd.DataFrame(self.violations)
    
    def has_critical_failures(self) -> bool:
        """Check if any critical validations failed."""
        return any(
            v['severity'] == ValidationSeverity.CRITICAL.value
            for v in self.violations
        )

# Usage
validator = DataValidator(df)

# Define validation rules
validator.validate_not_null(['user_id', 'amount'], severity=ValidationSeverity.CRITICAL)
validator.validate_unique(['transaction_id'])
validator.validate_range('amount', min_value=0, max_value=1000000)
validator.validate_format('email', r'^[\w\.-]+@[\w\.-]+\.\w+$')

# Check results
report = validator.get_report()
if validator.has_critical_failures():
    logging.error("Critical validation failures!")
    raise ValueError("Data validation failed")

Data Normalization

# Python 3.12 - Data normalization
from sklearn.preprocessing import StandardScaler, MinMaxScaler

class DataNormalizer:
    """
    Normalize data for consistency.
    Critical for ML features and cross-source comparisons.
    """
    
    @staticmethod
    def normalize_currency(
        df: pd.DataFrame,
        amount_col: str,
        currency_col: str,
        exchange_rates: dict
    ) -> pd.DataFrame:
        """
        Normalize all amounts to single currency.
        Real use case: multi-currency transactions.
        """
        df = df.copy()
        df['amount_usd'] = df.apply(
            lambda row: row[amount_col] * exchange_rates.get(row[currency_col], 1.0),
            axis=1
        )
        return df
    
    @staticmethod
    def normalize_dates(
        df: pd.DataFrame,
        date_columns: list[str],
        timezone: str = 'UTC'
    ) -> pd.DataFrame:
        """Normalize all dates to UTC."""
        for col in date_columns:
            df[col] = pd.to_datetime(df[col]).dt.tz_localize(timezone)
        return df
    
    @staticmethod
    def normalize_country_codes(
        df: pd.DataFrame,
        country_col: str
    ) -> pd.DataFrame:
        """
        Normalize country names to ISO codes.
        Handles variations like 'USA', 'United States', 'US'.
        """
        country_mapping = {
            'usa': 'US',
            'united states': 'US',
            'u.s.a.': 'US',
            'uk': 'GB',
            'united kingdom': 'GB',
            # Add more mappings
        }
        
        df[country_col] = (
            df[country_col]
            .str.lower()
            .str.strip()
            .map(country_mapping)
            .fillna(df[country_col])
        )
        
        return df
    
    @staticmethod
    def scale_numeric_features(
        df: pd.DataFrame,
        columns: list[str],
        method: str = 'standard'
    ) -> pd.DataFrame:
        """
        Scale numeric features.
        'standard': mean=0, std=1
        'minmax': scale to [0, 1]
        """
        scaler = StandardScaler() if method == 'standard' else MinMaxScaler()
        
        df[columns] = scaler.fit_transform(df[columns])
        return df

# Usage
df = DataNormalizer.normalize_currency(
    df,
    amount_col='amount',
    currency_col='currency',
    exchange_rates={'EUR': 1.1, 'GBP': 1.3, 'USD': 1.0}
)

Outlier Detection

# Python 3.12 - Outlier detection
class OutlierDetector:
    """
    Identify outliers that might indicate data quality issues.
    """
    
    @staticmethod
    def detect_iqr_outliers(
        df: pd.DataFrame,
        column: str,
        multiplier: float = 1.5
    ) -> pd.DataFrame:
        """
        Detect outliers using IQR method.
        Standard statistical approach.
        """
        Q1 = df[column].quantile(0.25)
        Q3 = df[column].quantile(0.75)
        IQR = Q3 - Q1
        
        lower_bound = Q1 - multiplier * IQR
        upper_bound = Q3 + multiplier * IQR
        
        outliers = df[
            (df[column] < lower_bound) |
            (df[column] > upper_bound)
        ]
        
        logging.info(f"Found {len(outliers)} outliers in {column}")
        return outliers
    
    @staticmethod
    def detect_zscore_outliers(
        df: pd.DataFrame,
        column: str,
        threshold: float = 3.0
    ) -> pd.DataFrame:
        """
        Detect outliers using Z-score method.
        Values more than 3 standard deviations from mean.
        """
        from scipy import stats
        
        z_scores = np.abs(stats.zscore(df[column].dropna()))
        outliers = df[z_scores > threshold]
        
        return outliers
    
    @staticmethod
    def cap_outliers(
        df: pd.DataFrame,
        column: str,
        lower_percentile: float = 0.01,
        upper_percentile: float = 0.99
    ) -> pd.DataFrame:
        """
        Cap outliers at percentile values.
        Less aggressive than removal.
        """
        lower_bound = df[column].quantile(lower_percentile)
        upper_bound = df[column].quantile(upper_percentile)
        
        df[column] = df[column].clip(lower=lower_bound, upper=upper_bound)
        return df

# Usage
outliers = OutlierDetector.detect_iqr_outliers(df, 'amount')
df = OutlierDetector.cap_outliers(df, 'amount', 0.01, 0.99)

Data Transformation Patterns

# Python 3.12 - Common transformations
class DataTransformer:
    """
    Standard transformations I use in pipelines.
    """
    
    @staticmethod
    def pivot_data(
        df: pd.DataFrame,
        index: str,
        columns: str,
        values: str,
        aggfunc: str = 'sum'
    ) -> pd.DataFrame:
        """
        Pivot long format to wide format.
        Example: pivot monthly sales by product.
        """
        return df.pivot_table(
            index=index,
            columns=columns,
            values=values,
            aggfunc=aggfunc,
            fill_value=0
        ).reset_index()
    
    @staticmethod
    def melt_data(
        df: pd.DataFrame,
        id_vars: list[str],
        value_vars: list[str],
        var_name: str = 'variable',
        value_name: str = 'value'
    ) -> pd.DataFrame:
        """
        Melt wide format to long format.
        Opposite of pivot.
        """
        return pd.melt(
            df,
            id_vars=id_vars,
            value_vars=value_vars,
            var_name=var_name,
            value_name=value_name
        )
    
    @staticmethod
    def create_time_features(
        df: pd.DataFrame,
        timestamp_col: str
    ) -> pd.DataFrame:
        """
        Extract time-based features.
        Useful for time series analysis.
        """
        df['year'] = df[timestamp_col].dt.year
        df['month'] = df[timestamp_col].dt.month
        df['day'] = df[timestamp_col].dt.day
        df['dayofweek'] = df[timestamp_col].dt.dayofweek
        df['hour'] = df[timestamp_col].dt.hour
        df['is_weekend'] = df['dayofweek'].isin([5, 6])
        
        return df
    
    @staticmethod
    def create_lag_features(
        df: pd.DataFrame,
        value_col: str,
        lags: list[int] = [1, 7, 30]
    ) -> pd.DataFrame:
        """
        Create lagged features for time series.
        Example: previous day/week/month values.
        """
        for lag in lags:
            df[f'{value_col}_lag_{lag}'] = df[value_col].shift(lag)
        
        return df

# Usage
df = DataTransformer.create_time_features(df, 'timestamp')
df = DataTransformer.create_lag_features(df, 'sales', lags=[1, 7, 30])

Complete Cleaning Pipeline

# Python 3.12 - End-to-end cleaning pipeline
class DataCleaningPipeline:
    """
    Complete cleaning pipeline.
    Real pipeline I use in production.
    """
    
    def __init__(self, df: pd.DataFrame):
        self.df = df
        self.original_count = len(df)
    
    def run(self) -> pd.DataFrame:
        """Execute complete cleaning pipeline."""
        logging.info(f"Starting cleaning pipeline with {self.original_count} records")
        
        # 1. Remove exact duplicates
        self.df = DuplicateHandler.remove_exact_duplicates(self.df)
        
        # 2. Type conversions
        self.df = TypeConverter.convert_to_numeric(
            self.df,
            ['amount', 'quantity']
        )
        self.df = TypeConverter.convert_to_datetime(
            self.df,
            ['created_at']
        )
        
        # 3. Handle missing data
        self.df = MissingDataHandler.fill_missing(self.df)
        
        # 4. Validation
        validator = DataValidator(self.df)
        validator.validate_not_null(['user_id', 'amount'])
        validator.validate_range('amount', 0, 1000000)
        
        if validator.has_critical_failures():
            raise ValueError("Critical validation failures")
        
        # 5. Outlier handling
        self.df = OutlierDetector.cap_outliers(self.df, 'amount')
        
        # 6. Normalization
        self.df = DataNormalizer.normalize_dates(self.df, ['created_at'])
        
        final_count = len(self.df)
        removed = self.original_count - final_count
        logging.info(f"Cleaning complete: {final_count} records ({removed} removed)")
        
        return self.df

# Usage
pipeline = DataCleaningPipeline(raw_df)
cleaned_df = pipeline.run()

Conclusion

Data cleaning is unglamorous but critical. The patterns here handle 90% of cleaning scenarios I encounter in production.

Key takeaways:

• Always analyze before cleaning

• Validate data quality with automated checks

• Handle missing data appropriately for context

• Remove or cap outliers based on business rules

• Document all transformations

---

Navigation:

• ← Previous: Data Ingestion & Sources

• Next: Data Modeling & Warehousing →

• Back to Overview