CDC Pipeline for Global commodity trade and logistics

[Harshal1309]

Summary

I have designed and implemented a Change Data Capture (CDC) Lakehouse synchronization pipeline under submission/harshal-logistics tailored for ROBUSTRADE's core business of international commodity trade, logistics, and border customs clearances.

The pipeline ingests data from a transactional source system, writes it to a raw append-only Lake, processes it under safety constraints, and updates a near real-time analytical Warehouse. The implementation features comprehensive Point-in-Time recovery (Time-Travel) and strict business rule data quality validators.

---

Source Schema Design

• Domain Chosen: Global Commodity Trade & Cargo Logistics.
• Strong Entities:
  • trade_partners: Profiles buyers and sellers.
  • trade_contracts: Commodity sales contracts (e.g. Copper, Cocoa) including pricing and tons.
• Weak Entities (tied to trade_contracts):
  • shipment_cargo: Container freight carrier numbers, load ports, discharge ports, and actual delivery dates.
  • customs_clearances: Border clearances, duty fee pricing, and approval statuses.
  • settlement_transactions: Wire transfer and letter of credit payments.
• Keys, relationships, and indexes: All tables enforce primary keys, referential foreign key integrity, and check constraints on the source side.
• Source validation rules: Status fields restrict values (e.g. compliance_status as active/suspended/under_review), metric tons and monetary values must be non-negative, and shipment delivery dates are checked.

---

CDC Strategy

• How changes are captured: CDC captures every source insert, update, and delete in transaction commit order as a CDCRecord carrying a monotonic sequence number (our Log Sequence Number/LSN).
• How inserts, updates, and deletes are handled:
  • Inserts and Updates are upserted into the warehouse.
  • Deletes are mapped to soft-deletes (_deleted = true) to preserve audit capabilities.
• How replay/restart works: The extraction offset can be checkpointed. Consumers use records_since(offset) to resume ingestion after a restart.
• How duplicates / out-of-order logs are handled: The warehouse ingestion enforces a strict merge guard: record.sequence > current_warehouse_row._cdc_seq. Duplicate arrivals and older out-of-order LSN messages are safely discarded.
  • Early Deletes: If a DELETE arrives before its corresponding INSERT, a placeholder soft-deleted row is created which gets updated by the subsequent insert.
  • Resurrection Safety: If a row is deleted at LSN 5, a late-arriving insert at LSN 3 is ignored, preventing row resurrection.

---

Lake and Warehouse Modeling

• How the lake captures every change: The Lake preserves raw CDC events immutably in the lake_cdc_events table, serializing row changes as raw JSON events alongside captured timestamps.
• How the warehouse maintains latest snapshot: Warehouse tables (prefixed with wh_) keep the consolidated latest status state and track CDC metadata.
• How time travel / restore is supported: The function reconstruct_warehouse_at(conn, lsn) truncates warehouse tables and replays lake records sequentially up to lsn to restore the exact historical database snapshot at that point in time.

---

Schema Change Safety

• How incompatible changes are detected: The pre-check scanner scans the source schema against the contract registry before pulling changes.
• How ingestion is stopped: Mismatches (such as dropped columns, renamed fields, or mismatched data types) trigger a fatal error, halting the ingestion pipeline.
• How warnings/failures are surfaced: The contract validator outputs contract violation details to stdout/stderr and exits with status code 1, halting the CI/CD build.

---

Validation Parity

• Which source system validations were mirrored downstream:
  • Uniqueness of primary keys.
  • Referential integrity (verifying child contracts, cargo, clearances, and transactions resolve to active parents).
  • NOT NULL constraints on active records.
• Business validations in Warehouse:
  • Math Check: total_value_usd == weight_metric_tons * price_per_ton_usd.
  • Overpayment Block: Total settled payments for a contract must be <= total_value_usd.
  • Customs Clearance Block: A shipment cannot be marked as delivered (actual_delivery is set) if customs clearance status is rejected or held_in_customs.
• How failures are checked and reported: Handled automatically by run_data_quality_checks.py. Mismatches list violations and return exit code 1.

---

Catalog Exposure

Lake and warehouse datasets are published in catalog/catalog.json with dataset descriptions, data layer tags (lake vs warehouse), schemas, update cadences, owners, and downstream consumers.

---

Validation

All checks and pipeline validations were executed and verified locally:

1. Pytest suite (29 tests):

   pytest -v submission/harshal-logistics/tests
   # Result: 29 passed in 3.26s

2. Schema contracts scanner:

   python submission/harshal-logistics/scripts/check_schema_contracts.py
   # Result: All schema contracts passed (5 tables checked).

3. Data Quality Auditor:

   python submission/harshal-logistics/scripts/run_data_quality_checks.py
   # Result: All data quality checks passed (system validations and business rules verified).

4. Catalog validator:

   python submission/harshal-logistics/scripts/validate_catalog.py
   # Result: Catalog validation passed (6 datasets verified).

---

Known Limitations / Next Steps

• In-memory WAL queue simulation is used. In a production environment, this would be backed by Postgres WAL replica logs (using logical replication) streamed via Debezium and Apache Kafka.
• Catalog metadata registry is currently stored in a local static JSON file. In production, this would be registered dynamically on a platform catalog like AWS Glue, Apache Atlas, or Unity Catalog.