r flag the run as invalid rather than producing false discrepancies.

Step 2: Extraction & Transport

For batch workflows, extract snapshots via SQL queries or API connectors. For event-driven workflows, Debezium captures row-level changes from relational sources and publishes them to Apache Kafka. Kafka acts as the durable transport layer, buffering events until the reconciliation consumer is ready. Consumer group configuration and retention policies must align with your recovery window requirements.

Step 3: Comparison Engine

The core logic performs set-based matching and field-level diffing. Pandas handles in-memory joins efficiently using outer merges with indicator columns to classify rows as left-only, right-only, or matched. Matched rows undergo field-by-field comparison, accounting for type coercion and precision tolerance. For datasets exceeding memory limits, PySpark mirrors this logic across distributed partitions.

Step 4: Orchestration & State Tracking

Prefect manages scheduling, retry policies, and run monitoring. Each reconciliation step is modeled as a task with independent retry logic. Discrepancies are persisted to an append-only ledger or upsert table with run IDs, enabling historical tracking and alerting.

Implementation Example: In-Memory Reconciliation Engine

The following Python module demonstrates a production-ready comparison pattern. It replaces fragile script-level logic with a typed, state-aware processor that handles precision tolerance, explicit join keys, and structured output.

import pandas as pd
import logging
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime

logger = logging.getLogger(__name__)

@dataclass
class ReconciliationConfig:
    join_keys: List[str]
    tolerance: float = 0.0001
    numeric_columns: List[str] = None
    run_id: str = None

class DatasetComparator:
    def __init__(self, config: ReconciliationConfig):
        self.cfg = config
        self.numeric_cols = config.numeric_columns or []

    def _apply_precision_tolerance(self, left_val: float, right_val: float) -> bool:
        if pd.isna(left_val) or pd.isna(right_val):
            return pd.isna(left_val) == pd.isna(right_val)
        return abs(left_val - right_val) <= self.cfg.tolerance

    def execute(self, source_a: pd.DataFrame, source_b: pd.DataFrame) -> pd.DataFrame:
        logger.info(f"Starting reconciliation run {self.cfg.run_id}")
        
        merged = source_a.merge(
            source_b,
            on=self.cfg.join_keys,
            how="outer",
            suffixes=("_src", "_tgt"),
            indicator=True
        )

        merged["status"] = merged["_merge"].map({
            "left_only": "MISSING_IN_TARGET",
            "right_only": "MISSING_IN_SOURCE",
            "both": "MATCHED"
        })

        discrepancy_records = []
        matched_mask = merged["status"] == "MATCHED"
        
        for idx, row in merged[matched_mask].iterrows():
            diffs = {}
            for col in source_a.columns:
                if col in self.cfg.join_keys:
                    continue
                src_val = row.get(f"{col}_src")
                tgt_val = row.get(f"{col}_tgt")
                
                if col in self.numeric_cols:
                    if not self._apply_precision_tolerance(src_val, tgt_val):
                        diffs[col] = {"source": src_val, "target": tgt_val}
                else:
                    if src_val != tgt_val:
                        diffs[col] = {"source": src_val, "target": tgt_val}
            
            if diffs:
                discrepancy_records.append({
                    "run_id": self.cfg.run_id,
                    "detected_at": datetime.utcnow().isoformat(),
                    "key_values": {k: row[k] for k in self.cfg.join_keys},
                    "discrepancies": diffs
                })

        logger.info(f"Run {self.cfg.run_id} complete. Found {len(discrepancy_records)} discrepancies.")
        return pd.DataFrame(discrepancy_records)

Architecture Decisions & Rationale

1. Outer Join with Indicator Column: Using how="outer" ensures no rows are silently dropped. The _merge indicator explicitly classifies unmatched records, which is critical for audit trails.
2. Precision Tolerance Layer: Floating-point drift is a leading cause of false positives in financial and telemetry reconciliation. The tolerance check prevents trivial precision mismatches from triggering alerts.
3. Stateful Run Tracking: Embedding run_id and detected_at in the output enables historical trending, duplicate suppression, and downstream alert routing.
4. Separation of Validation & Comparison: Great Expectations runs before extraction. If source quality fails, the comparison engine never executes, saving compute and preventing noise.

Pitfall Guide

1. Non-Deterministic Join Keys

Explanation: Using composite keys with nullable columns or high-cardinality strings causes cross-joins or missed matches. Fix: Enforce NOT NULL constraints on join keys during extraction. Hash composite keys into a single deterministic identifier before comparison.

2. Floating-Point Precision Drift

Explanation: Direct equality checks on decimals fail due to IEEE 754 representation differences across systems. Fix: Implement epsilon-based tolerance for numeric columns. Round values to a consistent decimal place before comparison.

3. Premature Distributed Scaling

Explanation: Deploying Spark clusters for datasets under 10M rows introduces unnecessary dependency management, shuffle overhead, and cost. Fix: Profile memory usage first. Use Pandas or DuckDB for in-memory workloads. Only migrate to PySpark when heap allocation exceeds 70% of available RAM consistently.

4. Neglecting Kafka Retention & Consumer Lag

Explanation: Short retention windows cause event loss during consumer restarts. Unmonitored lag leads to reconciliation gaps. Fix: Set retention to at least 2x your maximum recovery window. Implement consumer lag alerting and configure auto.offset.reset=earliest for reconciliation consumers.

5. Stateless Discrepancy Tracking

Explanation: Logging discrepancies to stdout or ephemeral tables prevents historical analysis and duplicate alerting. Fix: Persist findings to an append-only ledger with run IDs, timestamps, and resolution states. Implement idempotent upserts keyed by (run_id, join_key_hash).

6. Alert Fatigue from Silent Failures

Explanation: Orchestration tools retry failed runs without distinguishing between transient network errors and systemic data corruption. Fix: Classify exceptions into retryable (timeouts, connection drops) and terminal (schema mismatch, validation failure). Route terminal failures to incident channels immediately.

7. Ignoring Late-Arriving Data

Explanation: Event-driven pipelines assume strict ordering. Out-of-order CDC events cause temporary reconciliation mismatches. Fix: Implement watermarking or a propagation delay window before triggering comparison jobs. Buffer events until the expected consistency window closes.

Production Bundle

Action Checklist

• Define explicit join keys with NOT NULL enforcement and deterministic hashing
• Configure Great Expectations suites to validate source cardinality and null rates before extraction
• Set Kafka retention to 2x maximum recovery window and enable consumer lag monitoring
• Implement epsilon tolerance for all numeric/decimal comparison columns
• Persist discrepancies to an append-only table with run IDs and resolution states
• Classify orchestration exceptions into retryable vs terminal categories
• Add propagation delay windows for event-driven reconciliation to handle late-arriving CDC events
• Profile memory usage before selecting comparison engine; scale to Spark only when RAM thresholds are breached

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Daily batch < 5M rows, warehouse-native team	dbt + SQL relationships/audit tests	Leverages existing DW compute, zero additional infra	Low (DW credits only)
Sub-10M rows, Python stack, scheduled runs	Prefect + Pandas + Postgres ledger	Fast development, explicit state tracking, low ops overhead	Low (single node + managed DB)
Real-time detection required, relational sources	Debezium + Kafka + Prefect consumer	Event-driven propagation, parallel consumer groups, durable transport	Medium (Kafka cluster + CDC connectors)
> 100M rows, multi-terabyte datasets	PySpark on EMR/Dataproc + S3 staging	Distributed shuffle, partitioned comparison, horizontal scaling	High (cluster compute + storage)
Multi-cloud/hybrid sources, strict compliance	Great Expectations + Prefect + custom connectors	Declarative audit trails, version-controlled validation, centralized orchestration	Medium (orchestration + validation infra)

Configuration Template

Prefect flow with Great Expectations validation gate and structured discrepancy persistence:

from prefect import flow, task
from prefect.tasks import task_input_hash
import pandas as pd
import psycopg2
from datetime import datetime

@task(retries=2, retry_delay_seconds=30)
def validate_source_quality(suite_name: str, batch_id: str) -> bool:
    # Placeholder for Great Expectations validation run
    # Returns True if expectations pass, False otherwise
    return True

@task(cache_key_fn=task_input_hash)
def extract_source_a(run_id: str) -> pd.DataFrame:
    # Replace with actual connector logic
    return pd.DataFrame({"id": [1, 2, 3], "amount": [100.0, 200.5, 300.0]})

@task(cache_key_fn=task_input_hash)
def extract_source_b(run_id: str) -> pd.DataFrame:
    return pd.DataFrame({"id": [1, 2, 4], "amount": [100.0, 200.6, 400.0]})

@task
def persist_discrepancies(records: pd.DataFrame, db_conn_string: str):
    if records.empty:
        return
    conn = psycopg2.connect(db_conn_string)
    records.to_sql("reconciliation_ledger", conn, if_exists="append", index=False)
    conn.close()

@flow(name="financial-reconciliation-v1")
def reconciliation_flow():
    run_id = f"run_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}"
    
    if not validate_source_quality("source_quality_suite", run_id):
        raise ValueError("Source validation failed. Aborting reconciliation.")
        
    df_a = extract_source_a(run_id)
    df_b = extract_source_b(run_id)
    
    config = ReconciliationConfig(
        join_keys=["id"],
        tolerance=0.01,
        numeric_columns=["amount"],
        run_id=run_id
    )
    
    comparator = DatasetComparator(config)
    discrepancies = comparator.execute(df_a, df_b)
    
    persist_discrepancies(discrepancies, "postgresql://user:pass@localhost/recon_db")
    return discrepancies

Quick Start Guide

1. Initialize Validation Suite: Create a Great Expectations checkpoint for your primary source. Define expectations for row count bounds, null thresholds, and key uniqueness. Run locally to verify baseline quality.
2. Configure Extraction Tasks: Replace placeholder extractors with your actual connectors (SQL queries, API clients, or S3 loaders). Ensure join keys are explicitly selected and typed.
3. Deploy Orchestration: Install Prefect OSS or use the cloud free tier. Register the flow, configure a cron schedule matching your data propagation window, and set up Slack/email notifications for terminal failures.
4. Provision State Storage: Create a PostgreSQL or Snowflake table for the reconciliation ledger. Include columns for run_id, detected_at, key_hash, discrepancy_json, and resolution_status.
5. Execute & Monitor: Trigger the first run. Verify discrepancy output matches expected drift. Adjust tolerance thresholds and retention windows based on initial run metrics. Scale to Spark or Kafka only when batch duration or data volume breaches defined SLAs.