Definition
- Ingestion – a durable, partitioned log like Kafka or Kinesis decouples producers from consumers and supports replay.
- Partition key – typically account id; sets parallelism and per-key ordering for stateful processing.
- Processing mode – streaming for low-latency fraud detection, batch for completeness and reprocessing.
- Delivery semantics – exactly-once via idempotent producers, transactional writes, and checkpointed offsets.
- Storage – columnar Parquet in the lake, partitioned by date, with a lakehouse format for ACID.
- Reliability – schema registry, dead-letter queues, backpressure, monitoring, and reprocessing via replay.
Imagine the National Bank of India receiving millions of trade slips a day. You set up a conveyor belt with many lanes (Kafka partitions) so slips flow in order and nothing is lost. Then you decide: inspect each slip the moment it lands (streaming for fraud) or sort whole batches at night (batch for reports). Finally you file the processed slips in a smart cabinet where similar fields sit together (columnar Parquet) so later questions are quick.
Understanding the Concepts
How do you design the ingestion layer?
Ingestion starts with a durable, partitioned log that decouples producers from consumers, buffers spikes, and retains events for replay.
# Kafka producer: key by account_id for per-account ordering
from kafka import KafkaProducer
import json
producer = KafkaProducer(
bootstrap_servers="kafka.nbi.com:9092",
key_serializer=str.encode,
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
acks="all", # Wait for all replicas
enable_idempotence=True # Prevent duplicates
)
trade = {
"trade_id": "TRD-2024-001",
"account_id": "SAV2001",
"amount": 50000.00,
"timestamp": "2024-01-15T10:30:00Z"
}
# Partition by account_id: all events for SAV2001 go to one partition
producer.send("bank-trades", key="SAV2001", value=trade)
What is the first key decision: partition key?
The partition key sets both parallelism and ordering. Choosing the right key is critical for correctness and performance.
Partition Key Decision:
account_id:
+ All events for one account are ordered (correct for stateful)
+ Good for account-level aggregation
- Hot accounts can skew partitions
trade_id:
+ Even distribution
- No per-account ordering (wrong for stateful logic)
date:
+ Simple
- Only one partition per day (no parallelism)
Recommendation: account_id with enough partitions for peak throughput
National Bank of India: 64 partitions for 10M events/day
What is the second decision: processing mode?
Choose streaming for low latency or batch for completeness. Many banks run both.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window, sum as spark_sum
spark = SparkSession.builder.appName("PipelineDemo").getOrCreate()
# STREAMING: real-time fraud detection and alerts
stream = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "kafka.nbi.com:9092") \
.option("subscribe", "bank-trades") \
.option("startingOffsets", "latest") \
.load()
fraud_alerts = stream \
.withWatermark("timestamp", "5 minutes") \
.groupBy(window("timestamp", "10 minutes"), "account_id") \
.agg(spark_sum("amount").alias("total")) \
.filter(col("total") > 1000000) # Flag accounts > 10 lakh in 10 min
fraud_alerts.writeStream \
.format("delta") \
.option("checkpointLocation", "/checkpoints/fraud") \
.start("/data/fraud_alerts")
# BATCH: nightly end-of-day report
txns = spark.read.parquet("/data/lake/transactions") \
.filter(col("txn_date") == "2024-01-15")
daily_totals = txns.groupBy("account_id") \
.agg(spark_sum("amount").alias("daily_total"))
daily_totals.write.parquet("/reports/daily_totals/date=2024-01-15")
What is the third decision: delivery semantics?
For financial correctness, target exactly-once processing or at minimum idempotent writes.
# Exactly-once with Structured Streaming and Delta Lake
stream.writeStream \
.format("delta") \
.outputMode("append") \
.option("checkpointLocation", "/checkpoints/trades") \
.start("/data/lake/transactions")
# Checkpoint tracks Kafka offsets; Delta provides transactional writes
# Together: exactly-once from Kafka to Delta
# Alternative: idempotent writes with upsert
from delta.tables import DeltaTable
delta_table = DeltaTable.forPath(spark, "/data/lake/transactions")
# Upsert: insert new trades, update existing ones
# Even if a trade is replayed, the result is the same
delta_table.alias("target").merge(
new_trades.alias("source"),
"target.trade_id = source.trade_id"
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
What is the fourth decision: storage layout?
Raw events land in columnar format in the lake, partitioned for efficient querying.
# Write to the lake in Parquet, partitioned by date
txns.write \
.format("delta") \
.partitionBy("txn_date") \
.option("overwriteSchema", "true") \
.save("/data/lake/transactions")
# Partition by date: low cardinality, enables date-range pruning
# ZORDER by account_id: high cardinality, enables data skipping
# File size: target 256 MB - 1 GB with OPTIMIZE
# Curated layer: modeled subset for fast serving
curated = txns.groupBy("account_id", "txn_date") \
.agg(spark_sum("amount").alias("daily_total"))
curated.write.format("delta").save("/data/warehouse/daily_totals")
What reliability controls are essential?
A senior engineer never ships a pipeline without these controls.
# 1. Schema registry: prevent breaking changes
# Kafka Schema Registry enforces compatibility rules
# so a producer change does not break consumers
# 2. Dead-letter queue: handle poison records
def process_with_dlq(batch_df, batch_id):
try:
clean = batch_df.filter(col("amount").isNotNull())
bad = batch_df.filter(col("amount").isNull())
bad.write.mode("append").parquet("/data/dlq/trades")
clean.write.format("delta").mode("append") \
.save("/data/lake/transactions")
except Exception as e:
# Log error, alert, do not crash the stream
print(f"Batch {batch_id} error: {e}")
# 3. Monitoring: consumer lag, throughput, error rates
# Monitor Kafka consumer lag to detect falling behind
# Monitor Spark streaming batch duration vs trigger interval
# Alert if batch duration > trigger interval (backpressure)
# 4. Reprocessing: replay from the log
# Kafka retains events for N days
# To reprocess: reset offsets and replay
# Idempotent writes ensure replay does not duplicate data
What does the end-to-end architecture look like?
The complete pipeline has three layers: ingestion, processing, and storage, with reliability controls at each.
End-to-End Pipeline at the National Bank of India:
Producers -> Kafka (partitioned by account_id, 64 partitions)
│
┌──────────┴──────────┐
▼ ▼
Streaming Path Batch Path
(Structured Streaming) (Spark SQL)
- Fraud detection - Daily totals
- Real-time alerts - Regulatory reports
- Low latency - Completeness
│ │
▼ ▼
Delta Lake (Bronze) Delta Lake (Silver/Gold)
- Raw events - Curated aggregates
- Partitioned by date - Served to warehouse
- ZORDER by account - ZORDER by account
What should a senior engineer know?
- The partition key choice is the most consequential early decision because it is hard to change later.
- At the National Bank of India, the overarching principle was that the pipeline must be scalable and fast, but above all correct and recoverable, because its output drives money movement, fraud decisions, and regulatory reports.
- Plan for reprocessing from the start: retain events in Kafka, use idempotent writes, and design replay as a tested operation.
- The lambda architecture (streaming + batch) is common in banks because streaming provides immediacy and batch provides correctness with full reprocessing.
Visual Overview
End-to-End Pipeline Architecture
┌─────────────────────────────────────────────────────┐
│ Producers (trading systems, payment gateways) │
└────────────────────────┬────────────────────────────┘
▼
┌─────────────────────────────────────────────────────┐
│ Kafka / Kinesis (durable, partitioned log) │
│ Key: account_id Partitions: 64 Retention: 7 days│
└───────────┬──────────────────────┬──────────────────┘
│ │
┌─────────▼─────────┐ ┌─────────▼─────────┐
│ Streaming Path │ │ Batch Path │
│ Structured │ │ Spark SQL │
│ Streaming │ │ Nightly jobs │
│ - Fraud alerts │ │ - Daily totals │
│ - Real-time │ │ - Regulatory │
└─────────┬─────────┘ └─────────┬─────────┘
│ │
┌─────────▼──────────────────────▼─────────┐
│ Delta Lake │
│ Bronze: raw events (partitioned by date) │
│ Silver: cleaned, deduplicated │
│ Gold: curated aggregates for serving │
└───────────────────────────────────────────┘
Key Takeaway
Designing a pipeline for millions of trade events per day requires deliberate decisions at three layers. Ingestion uses a durable, partitioned log like Kafka, keyed by account id for per-account ordering and parallelism. Processing is chosen as streaming for low-latency fraud detection or batch for completeness, and many banks run both. Storage lands raw events in columnar format partitioned by date, with a lakehouse table format for ACID and schema evolution.
The delivery semantics decision is critical for financial correctness. Exactly-once is the target, achieved through idempotent producers, transactional writes to Delta Lake, and checkpointed offsets. When exactly-once is not achievable end to end, idempotent downstream writes with upserts keyed by event id ensure replays do not duplicate data.
Reliability controls are non-negotiable. A schema registry prevents producer changes from breaking consumers. Dead-letter queues quarantine poison records. Monitoring tracks consumer lag, batch duration, and error rates. Reprocessing is planned from the start by retaining events and designing idempotent writes so a replay is a tested operation, not an emergency.
At the National Bank of India, the overarching principle was that the pipeline must be scalable and fast, but above all correct and recoverable, because its output drives money movement, fraud decisions, and regulatory reports.
Remember one sentence: ingest through a partitioned durable log, process as streaming or batch per the latency need, store in columnar format with ACID, and build in reliability controls for a pipeline that is correct and recoverable above all.
---