Implementing Statistical Drift Detection for Incremental Hudi Ingests Using SciPy Pre-Commit Validators

The operational alert at 2 AM wasn’t for a system crash, but for a data pipeline that had silently started poisoning a production machine learning model. For two days, an upstream IoT sensor fleet had a firmware bug causing a subtle shift in the distribution of its pressure readings—not enough to generate nulls or out-of-range values that our basic validation would catch, but enough to skew the feature values and degrade model performance by 12%. The cost wasn’t just in engineering hours spent on a full backfill and model retrain; it was the erosion of trust in the data platform. The root cause was clear: our Incremental Source Reconciliation (ISR) pipeline, which efficiently upserted data into an Apache Hudi table, lacked any concept of statistical validation. It only cared about schema, not substance.

Our initial concept was to build a statistical gatekeeper. Before any micro-batch of data could be committed to our core Hudi dataset, it had to pass a statistical consistency check against a historical baseline. This meant moving beyond simple col IS NOT NULL checks and into the realm of distributional analysis. The goal was to programmatically answer the question: “Does this new sliver of data look like the data we’ve seen before?”

Technology selection was straightforward on the surface. We were committed to Apache Hudi for its transactional capabilities on the data lake, which were essential for managing our continuous stream of updates and deletes. The core of our processing was Apache Spark, for its scalability. For the statistical heavy lifting, the Python SciPy library was the obvious choice; it’s robust, battle-tested, and contains the exact non-parametric tests we needed. The real challenge, as always, was in the integration—making these disparate systems work together reliably in a production data pipeline. A common mistake is to prototype such a system in a notebook and assume it will translate directly to a production streaming job. The pitfalls lie in state management, performance, and error handling.

Our first major task was defining what “statistically normal” meant. This required creating a baseline profile of our core sensor_readings Hudi table. This profile couldn’t be static; it needed to be periodically recalculated to account for natural data evolution. We built a dedicated Spark job for this purpose.

import os
import json
import logging
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class BaselineProfileGenerator:
    """
    Calculates and persists statistical profiles for key columns in a Hudi table.
    The profile includes histograms, quantiles, and basic descriptive statistics.
    This is intended to be run as a periodic batch job (e.g., daily).
    """

    def __init__(self, spark: SparkSession, hudi_table_path: str, profile_output_path: str):
        if not spark:
            raise ValueError("SparkSession must be provided.")
        self.spark = spark
        self.hudi_table_path = hudi_table_path
        self.profile_output_path = profile_output_path
        self.target_columns = ["pressure_kpa", "temperature_celsius", "humidity_percent"]

    def _calculate_histograms(self, df, column_name, num_bins=20):
        """Calculates a histogram for a given column."""
        logging.info(f"Calculating histogram for column: {column_name}")
        min_max_df = df.agg(F.min(column_name).alias("min_val"), F.max(column_name).alias("max_val")).first()
        if not min_max_df or min_max_df.min_val is None:
            logging.warning(f"Could not calculate min/max for {column_name}, skipping histogram.")
            return []

        min_val, max_val = min_max_df.min_val, min_max_df.max_val
        bin_width = (max_val - min_val) / num_bins
        
        # Create bins and count elements in each
        histogram = (
            df.select(
                column_name,
                (F.floor((F.col(column_name) - min_val) / bin_width)).cast("int").alias("bin")
            )
            .groupBy("bin")
            .count()
            .orderBy("bin")
            .toJSON()
            .collect()
        )
        return [json.loads(row) for row in histogram]

    def _calculate_descriptive_stats(self, df, column_name):
        """Calculates mean, stddev, count, min, and max."""
        logging.info(f"Calculating descriptive stats for column: {column_name}")
        stats = df.select(
            F.mean(column_name).alias("mean"),
            F.stddev(column_name).alias("stddev"),
            F.count(column_name).alias("count"),
            F.min(column_name).alias("min"),
            F.max(column_name).alias("max")
        ).first()
        return stats.asDict() if stats else {}
        
    def generate_and_save_profile(self):
        """
        Main method to run the profiling process.
        It reads the Hudi table, calculates profiles for target columns,
        and saves the consolidated profile as a JSON file.
        """
        logging.info(f"Starting baseline profile generation for Hudi table at {self.hudi_table_path}")
        try:
            hudi_df = self.spark.read.format("hudi").load(self.hudi_table_path)
            # In a real-world project, you'd likely sample for performance on very large tables.
            # For this example, we use the full dataset.
            # hudi_df = hudi_df.sample(fraction=0.1, seed=42)
            hudi_df.cache()
            
            full_profile = {}
            for col_name in self.target_columns:
                logging.info(f"Processing column: {col_name}")
                if col_name not in hudi_df.columns:
                    logging.warning(f"Column {col_name} not found in DataFrame. Skipping.")
                    continue

                column_df = hudi_df.select(col_name).dropna()
                
                full_profile[col_name] = {
                    "descriptive_stats": self._calculate_descriptive_stats(column_df, col_name),
                    "histogram": self._calculate_histograms(column_df, col_name)
                    # Quantiles are also extremely useful but omitted here for brevity
                }

            hudi_df.unpersist()

            # Save the profile to a location accessible by the ingest pipeline (e.g., S3, HDFS)
            profile_json = json.dumps(full_profile, indent=4)
            # Writing through Spark to handle distributed filesystems correctly
            self.spark.sparkContext.parallelize([profile_json]).coalesce(1).saveAsTextFile(self.profile_output_path + "_temp")
            
            # A common pattern to get a single file out of saveAsTextFile
            fs = self.spark._jvm.org.apache.hadoop.fs.FileSystem.get(self.spark._jsc.hadoopConfiguration())
            temp_path = fs.globStatus(self.spark._jvm.org.apache.hadoop.fs.Path(self.profile_output_path + "_temp/part*"))[0].getPath()
            final_path = self.spark._jvm.org.apache.hadoop.fs.Path(self.profile_output_path)
            fs.rename(temp_path, final_path)
            fs.delete(self.spark._jvm.org.apache.hadoop.fs.Path(self.profile_output_path + "_temp"), True)
            
            logging.info(f"Successfully generated and saved baseline profile to {self.profile_output_path}")

        except Exception as e:
            logging.error(f"Failed to generate baseline profile: {e}", exc_info=True)
            raise

if __name__ == '__main__':
    spark = (
        SparkSession.builder.appName("HudiBaselineGenerator")
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .config("spark.sql.hive.convertMetastoreParquet", "false")
        .getOrCreate()
    )

    HUDI_TABLE_PATH = "s3://my-data-lake/hudi/sensor_readings"
    PROFILE_OUTPUT_PATH = "s3://my-data-lake/validation_profiles/sensor_readings_baseline.json"

    generator = BaselineProfileGenerator(spark, HUDI_TABLE_PATH, PROFILE_OUTPUT_PATH)
    generator.generate_and_save_profile()

With a mechanism to generate our baseline, the next critical step was the validator itself. This is where we needed to integrate SciPy. Our initial thought was to use Hudi’s native HoodiePreCommitValidator interface. The pitfall here is that this is a Java/Scala interface. While technically possible to bridge from Python, it’s fraught with complexity and brittleness. A pragmatic senior engineer knows when to avoid fighting the framework. We pivoted to a simpler, more robust architecture: the validation would be a discrete step in our PySpark application, executed just before the Hudi write operation. If validation fails, the save() call is never made for the main table; instead, the micro-batch is diverted to a quarantine area. This achieves the same “pre-commit” logical outcome without the cross-language integration headache.

The core of the validator uses the two-sample Kolmogorov-Smirnov (KS) test. It’s a powerful non-parametric test that compares two distributions. Our hypothesis is that the distribution of the incoming micro-batch should be statistically indistinguishable from the historical baseline distribution. If the KS test returns a very small p-value (e.g., < 0.05), we reject this hypothesis and flag the data as anomalous.

import json
import logging
import pandas as pd
from scipy.stats import ks_2samp
from pyspark.sql import DataFrame, SparkSession

class HudiStatisticalValidator:
    """
    Validates an incoming Spark DataFrame micro-batch against a pre-calculated baseline profile.
    Uses the Kolmogorov-Smirnov test to detect distributional drift.
    """

    def __init__(self, spark: SparkSession, profile_path: str, p_value_threshold: float = 0.05):
        self.spark = spark
        self.profile_path = profile_path
        self.p_value_threshold = p_value_threshold
        self.baseline_profile = self._load_profile()
        self.target_columns = list(self.baseline_profile.keys())

    def _load_profile(self) -> dict:
        """Loads the JSON baseline profile from a distributed filesystem."""
        try:
            logging.info(f"Loading baseline profile from {self.profile_path}")
            profile_str = self.spark.sparkContext.textFile(self.profile_path).collect()[0]
            return json.loads(profile_str)
        except Exception as e:
            logging.error(f"Could not load or parse baseline profile at {self.profile_path}. Error: {e}")
            # In a production system, this should trigger a high-priority alert.
            # The pipeline cannot validate without a baseline.
            raise IOError("Failed to load baseline profile.")

    def _reconstruct_baseline_sample(self, column_name: str, sample_size: int = 10000) -> pd.Series:
        """
        Reconstructs a numeric sample from the stored histogram. This is a crucial step.
        The KS test needs raw data points, not aggregated stats. We generate a representative
        sample from our stored histogram to avoid pulling massive amounts of historical data
        during the real-time ingest. This is an approximation, but a very effective one.
        """
        profile = self.baseline_profile.get(column_name)
        if not profile or "histogram" not in profile or "descriptive_stats" not in profile:
            raise ValueError(f"Incomplete profile for column {column_name}")

        hist_data = profile["histogram"]
        stats = profile["descriptive_stats"]
        
        if not hist_data or stats['count'] == 0:
            return pd.Series([], dtype=float)

        # Calculate bin width from min/max and number of bins
        num_bins = len(hist_data)
        min_val, max_val = stats['min'], stats['max']
        bin_width = (max_val - min_val) / num_bins if num_bins > 0 else 0

        # Create arrays of bin midpoints and their corresponding frequencies (counts)
        bin_midpoints = [min_val + (bin_info['bin'] + 0.5) * bin_width for bin_info in hist_data]
        counts = [bin_info['count'] for bin_info in hist_data]
        total_count = sum(counts)

        # Create a probability distribution and sample from it
        probabilities = [c / total_count for c in counts]
        
        # Use pandas/numpy for efficient sampling
        reconstructed_sample = pd.Series(
            pd.np.random.choice(bin_midpoints, size=sample_size, p=probabilities, replace=True)
        )
        return reconstructed_sample

    def validate_micro_batch(self, micro_batch_df: DataFrame) -> (bool, dict):
        """
        Performs statistical validation on the incoming micro-batch.

        Returns:
            - A boolean indicating if the validation passed.
            - A dictionary containing detailed results for logging/quarantining.
        """
        logging.info("Starting statistical validation of micro-batch.")
        if micro_batch_df.isEmpty():
            logging.warning("Micro-batch is empty, skipping validation.")
            return True, {"status": "SKIPPED", "reason": "Empty batch"}

        # Collect data for validation. For performance, we limit the sample size
        # from the incoming batch as well. 5000 records is often enough for the KS test.
        # This is a critical performance tuning parameter.
        validation_results = {}
        overall_pass = True

        for column in self.target_columns:
            if column not in micro_batch_df.columns:
                continue

            try:
                # We need to bring the data to the driver to use SciPy.
                # This is a bottleneck and its size must be controlled.
                incoming_data_pd = micro_batch_df.select(column).dropna().sample(fraction=1.0, seed=42).limit(5000).toPandas()
                
                if incoming_data_pd.empty:
                    validation_results[column] = {"status": "SKIPPED", "reason": "No data in batch for this column"}
                    continue
                
                incoming_sample = incoming_data_pd[column]
                baseline_sample = self._reconstruct_baseline_sample(column, sample_size=len(incoming_sample))

                if len(baseline_sample) == 0:
                    validation_results[column] = {"status": "SKIPPED", "reason": "No baseline data available"}
                    continue

                # The core statistical test
                ks_statistic, p_value = ks_2samp(incoming_sample, baseline_sample)

                is_valid = p_value >= self.p_value_threshold

                if not is_valid:
                    overall_pass = False

                validation_results[column] = {
                    "status": "PASS" if is_valid else "FAIL",
                    "p_value": p_value,
                    "ks_statistic": ks_statistic,
                    "threshold": self.p_value_threshold
                }
                logging.info(f"Validation for column '{column}': {'PASS' if is_valid else 'FAIL'} (p-value: {p_value:.4f})")

            except Exception as e:
                logging.error(f"Error validating column {column}: {e}", exc_info=True)
                validation_results[column] = {"status": "ERROR", "reason": str(e)}
                overall_pass = False # Fail open or fail closed? We choose to fail closed.

        return overall_pass, validation_results

Finally, we integrate this validator into our main ISR ingestion pipeline. The pipeline reads a micro-batch from a source like Kafka, applies transformations, and then, before writing to Hudi, calls our validator.

Here’s the structure of the main ingest job, demonstrating the quarantine logic.

graph TD
    A[Kafka Source] --> B{Spark Streaming Job};
    B --> C[Apply Transformations];
    C --> D{HudiStatisticalValidator};
    D -- PASS --> E[Write to Production Hudi Table];
    D -- FAIL --> F[Write to Quarantine Hudi Table];
    F --> G[Trigger Alert];

And the corresponding PySpark code:

# This is a simplified representation of the main ingest application
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import lit, to_json, struct

# Assume the validator classes from above are available
# from validator import HudiStatisticalValidator
# from baseline import BaselineProfileGenerator

def write_to_hudi(df: DataFrame, table_path: str, table_name: str, record_key: str, partition_key: str):
    """Encapsulates the Hudi write logic."""
    hudi_options = {
        'hoodie.table.name': table_name,
        'hoodie.datasource.write.recordkey.field': record_key,
        'hoodie.datasource.write.partitionpath.field': partition_key,
        'hoodie.datasource.write.table.name': table_name,
        'hoodie.datasource.write.operation': 'upsert',
        'hoodie.datasource.write.precombine.field': 'updated_at',
        'hoodie.upsert.shuffle.parallelism': 2,
        'hoodie.insert.shuffle.parallelism': 2,
    }
    
    logging.info(f"Writing {df.count()} records to Hudi table {table_name} at {table_path}")
    df.write.format("hudi").options(**hudi_options).mode("append").save(table_path)

def main():
    spark = (
        SparkSession.builder.appName("ValidatedHudiIngest")
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .getOrCreate()
    )

    # Common paths and configurations
    PROD_TABLE_PATH = "s3://my-data-lake/hudi/sensor_readings"
    QUARANTINE_TABLE_PATH = "s3://my-data-lake/hudi/quarantine/sensor_readings"
    PROFILE_PATH = "s3://my-data-lake/validation_profiles/sensor_readings_baseline.json"

    # 1. Read a micro-batch from Kafka (or another source)
    # This part is highly dependent on the source system.
    # For this example, let's create a dummy DataFrame representing a new batch of data.
    # In a real scenario, this would come from `spark.readStream.format("kafka")...`
    micro_batch_df = spark.createDataFrame(...) # Assume this has new sensor data

    # 2. Initialize the validator
    try:
        validator = HudiStatisticalValidator(spark, PROFILE_PATH, p_value_threshold=0.05)
    except IOError as e:
        logging.critical(f"Pipeline cannot start: {e}. Shutting down.")
        # Terminate the job if the baseline is missing.
        return

    # 3. Perform validation
    is_valid, validation_details = validator.validate_micro_batch(micro_batch_df)

    # 4. Route data based on validation outcome
    if is_valid:
        logging.info("Validation passed. Writing to production table.")
        write_to_hudi(
            df=micro_batch_df,
            table_path=PROD_TABLE_PATH,
            table_name="sensor_readings",
            record_key="sensor_id",
            partition_key="event_date"
        )
    else:
        logging.warning("Validation FAILED. Diverting batch to quarantine.")
        # Add validation metadata to the quarantined data for easier debugging
        details_json = to_json(struct([lit(v).alias(k) for k, v in validation_details.items()]))
        quarantined_df = micro_batch_df.withColumn("quarantine_reason", details_json)
        
        write_to_hudi(
            df=quarantined_df,
            table_path=QUARANTINE_TABLE_PATH,
            table_name="quarantined_sensor_readings",
            record_key="sensor_id",
            partition_key="event_date"
        )
        # Here, you would integrate with an alerting system (PagerDuty, Slack, etc.)
        logging.error(f"Data quarantine alert. Details: {validation_details}")

if __name__ == '__main__':
    main()

This system isn’t without its limitations. The primary trade-off is latency. Performing statistical tests, even on samples, adds overhead to the ingestion job. For pipelines with sub-second latency requirements, this approach would need significant performance tuning, perhaps by moving the validation to a more performant language or using approximate algorithms. Furthermore, the baseline reconstruction from a histogram is an approximation. It’s effective for many use cases but could miss very subtle, complex drifts that a test on the full historical dataset might catch.

Future iterations could explore developing an adaptive baseline that learns from recently validated data, reducing the reliance on a stale, periodically generated profile. Another avenue is expanding from univariate validation (one column at a time) to multivariate anomaly detection, which could identify corrupt data by spotting breakdowns in the relationships between columns—a far more complex but powerful form of data quality assurance.