tarting 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:

```python
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.