Data Pipeline Integration (OLTP, OLAP, NoSQL)

Both streaming and batch paths exist :

• Streaming path (Debezium → Kafka → Snowflake): delivers near real-time, row-level updates for low-latency use cases (e.g., live dashboards, fraud detection).
• Batch path (Spark → Snowflake): runs nightly for heavy aggregations, full historical recalculations, and backfills — workloads too resource-intensive or complex to run continuously in the streaming layer.

Together, these form a Lambda architecture: the streaming layer keeps data fresh, while the batch layer periodically reconciles and recomputes for accuracy and completeness.

---

🔄 Data Integration Architecture: Real-Time and Batch Pipeline

Goal:

• Ensure consistent, real-time, and batch data flow between OLTP, OLAP, and NoSQL systems.

• Supports low-latency processing (<500ms for fraud detection) and scalable ETL/ELT for analytics.

---

Real-Time streaming

This is how OLTP changes instantly arrive in OLAP/NoSql database.

• Debezium uses Change Data Capture (CDC) — a non-intrusive technique that reads changes (inserts/updates/deletes) directly from OLTP logs (e.g., PostgreSQL WAL) — and publishes them to Kafka topics (e.g., payments, refunds). Each change triggers downstream processing.
• Kafka provides scalable, low-latency event streaming for these transactional events.
• Kafka Streams enriches events in-flight (e.g., adding region from Merchant_Dim).
• Kafka Connect (Sink Connectors) writes the enriched events into both the OLAP system (Snowflake) and NoSQL (MongoDB) in parallel.

Example flow:

flowchart LR
    PostgreSQL[OLTP] -->|WAL Logs| Debezium
    Debezium -->|"topics: payments, refunds"| Kafka
    Kafka --> Enrich[Kafka Streams/Flink Enrichment]
    Enrich -->|Sink Connector| OLAP[Snowflake]
    Enrich -->|Sink Connector| NoSQL[MongoDB]

Workflow:

1. OLTP (PostgreSQL) → Debezium (tails WAL logs) → Kafka (topics: payments, refunds).

2. Kafka → OLAP (Snowflake) and NoSQL (MongoDB) via Kafka Connect/Sink Connectors.

3. Enrichment: Kafka Streams/Flink adds metadata (e.g., geolocation, fraud scores).

Example Event (Kafka):

{
  "payload": {
    "before": {"status": "pending", "amount": 99.99},
    "after": {"status": "completed", "amount": 99.99},
    "op": "u",
    "source": {"table": "transactions", "db": "stripe_oltp"},
    "ts_ms": 1682090401000
  }
}

Linking Systems Across OLTP/OLAP/NoSQL

Document NoSQL are linked to OLAP databases. To keep records joinable across systems, NoSQL documents embed the relevant foreign keys (e.g., merchant_id, customer_id) so they can still be linked back to Merchant_Dim / Customer_Dim in OLAP database.

Example:

{
  "_id": "txn_12345",
  "merchant_id": "m_67890",
  "customer_id": "c_54321",
  "amount": 99.99
}

Write path: enriched events are written to MongoDB via the MongoDB Kafka Connector.

{
  "_id": "txn_12345",
  "status": "refunded",
  "refund_reason": "customer_request",
  "merchant_id": "m_67890",
  "customer_id": "c_54321",
  "metadata": {
    "source": "OLTP",
    "change_type": "update",
    "change_timestamp": "2026-04-21T14:30:01Z"
  }
}

Failure Handling & Consistency

Here is how mistakes and duplicates are handled.

Challenge	Mitigation
Data Consistency	Idempotent writes + unique event IDs to avoid duplicates + reconciliation jobs.
High Throughput	Scale Kafka brokers/partitions + NoSQL write optimization.
Event Ordering	Use Kafka's partition keys to order events for a single record.
Schema Evolution	Schema registry (e.g., Confluent) to manage compatibility.

End-to-End Example: Payment Status Update

Here is an example which illustrates the streaming path for a refund.

1. Merchant refunds a transaction in OLTP (PostgreSQL).
2. Debezium captures the UPDATE event → Kafka topic (payments).
3. Kafka Streams enriches the event (e.g., adds refund_reason_code).
4. MongoDB Sink Connector writes the updated document to NoSQL.
5. The fraud detection ML model (querying MongoDB) adjusts the customer's risk profile.

Batch Sync (Spark + Airflow)

This is the complementary path for large calculus in OLAP/NoSQL database.

It consists of nightly ETL for historical data and heavy aggregations/transformations (e.g., revenue by region) and performs data enrichment (e.g., geolocation lookup, currency conversion).

Pipeline Diagram:

flowchart LR
    PostgreSQL[OLTP] -->|Batch Extract| Spark
    Spark -->|ETL: Aggregate & Transform| Snowflake[OLAP]
    Spark -->|ETL: Aggregate & Transform| MongoDB[NoSQL]
    Airflow[Airflow Scheduler] -.->|Triggers Nightly| Spark

• Airflow schedules and triggers the Spark job nightly.
• Spark extracts historical data directly from PostgreSQL and runs heavy aggregations/transformations.
• Results are written to both Snowflake (OLAP) for analytical queries and MongoDB (NoSQL) for downstream applications (e.g., fraud ML models needing pre-aggregated features).