Definition
Data quality and failure handling are two sides of pipeline reliability.
- Schema validation – Confirms each record matches the expected structure and types, enforced by a schema registry with compatibility rules.
- Content checks – Null checks for mandatory fields, range checks for sensible values, uniqueness checks for keys like transaction ID.
- Reconciliation – Row-count and sum checks comparing output against the source to catch silent data loss.
- Data-contract tests – Automated tests formalizing the agreement between producing and consuming teams.
- Anomaly alerting – Watches metrics over time and flags sharp deviations from the norm.
- Failure recovery – Quarantine bad records to a DLQ, make every step idempotent, backfill missed data, and alert with clear ownership.
The goal is to catch problems early, isolate them, and recover safely without publishing incorrect data.
Think of the National Bank of India's data pipeline as a quality line in a factory that produces financial reports. Data quality is the row of inspectors along the belt: one checks the form has the right shape (schema), one checks no mandatory box is blank and amounts are sensible (null and range), one checks no transaction ID appears twice (uniqueness), one counts the output against what the source sent (reconciliation), and one rings an alarm when today's numbers look wildly off (anomaly). When something breaks, jammed forms go into a reject bin (DLQ), every station is built so re-running a batch does not duplicate products (idempotency), and a named supervisor is paged to fix it (ownership).
Understanding the Concepts
What is schema validation?
Schema validation confirms each record matches the expected structure and types. In a banking pipeline, a schema registry enforces compatibility rules so a producer change cannot silently break the pipeline.
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
# Schema Registry enforces that producer schema is compatible
schema_registry = SchemaRegistryClient({"url": "http://schema-registry:8081"})
# Avro schema for bank transactions
txn_schema = """
{
"type": "record",
"name": "Transaction",
"fields": [
{"name": "transaction_id", "type": "string"},
{"name": "account_id", "type": "string"},
{"name": "amount", "type": {"type": "bytes", "logicalType": "decimal", "precision": 18, "scale": 2}},
{"name": "txn_date", "type": "string"},
{"name": "branch_code", "type": "string"}
]
}
"""
# If a producer adds or removes a field incompatibly,
# the registry rejects the schema change before it reaches production
What are null, range, and uniqueness checks?
Beyond structure, you apply content checks on every record. At the National Bank of India, these are the day-to-day guards that catch the most common data corruption.
from pyspark.sql import functions as F
txns = spark.read.parquet("/data/bank/raw/transactions")
# Null checks: mandatory fields must be present
null_violations = txns.filter(
F.col("account_id").isNull() |
F.col("amount").isNull() |
F.col("transaction_id").isNull()
)
print(f"Null violations: {null_violations.count()}")
# Range checks: amounts must be sensible
range_violations = txns.filter(
(F.col("amount") < 0) |
(F.col("amount") > 100000000) # 10 crore upper bound
)
print(f"Range violations: {range_violations.count()}")
# Uniqueness checks: no duplicate transaction IDs
duplicates = txns.groupBy("transaction_id").count().filter("count > 1")
print(f"Duplicate transaction IDs: {duplicates.count()}")
How do reconciliation checks work?
Reconciliation and row-count checks compare the pipeline's output against the source. If the source sent a million rows and the warehouse has nine hundred thousand, you know data was lost. For money, you also reconcile totals.
import java.math.BigDecimal
# Row count reconciliation
source_count = spark.read.parquet("/data/bank/raw/transactions").count()
target_count = spark.read.parquet("/data/bank/gold/transactions").count()
if source_count != target_count:
raise Exception(
f"Row count mismatch: source={source_count}, target={target_count}. "
f"Pipeline halted for investigation."
)
# Sum reconciliation for money
source_total = spark.read.parquet("/data/bank/raw/transactions") \
.agg(F.sum("amount")).collect()[0][0]
target_total = spark.read.parquet("/data/bank/gold/transactions") \
.agg(F.sum("amount")).collect()[0][0]
if abs(source_total - target_total) > 0.01:
raise Exception(
f"Amount mismatch: source={source_total}, target={target_total}. "
f"Possible silent data loss."
)
print("Reconciliation passed: row counts and totals match.")
What are data-contract tests?
Data-contract tests formalize the agreement between the producing team and the consuming team. They run as automated tests in CI so a breaking change is caught before it reaches production.
# data_contract_test.py
import pytest
from pyspark.sql.types import StructType, StructField, StringType, DecimalType, DateType
EXPECTED_SCHEMA = StructType([
StructField("transaction_id", StringType(), nullable=False),
StructField("account_id", StringType(), nullable=False),
StructField("amount", DecimalType(18, 2), nullable=False),
StructField("txn_date", DateType(), nullable=False),
StructField("branch_code", StringType(), nullable=False),
])
def test_transaction_schema_contract():
"""Contract: transactions table must match the agreed schema."""
actual = spark.read.parquet("/data/bank/silver/transactions").schema
for expected_field in EXPECTED_SCHEMA.fields:
actual_field = actual[expected_field.name]
assert actual_field.dataType == expected_field.dataType, \
f"Type mismatch for {expected_field.name}"
assert actual_field.nullable == expected_field.nullable, \
f"Nullability mismatch for {expected_field.name}"
def test_no_null_account_ids():
"""Contract: account_id must never be null."""
nulls = spark.read.parquet("/data/bank/silver/transactions") \
.filter("account_id IS NULL").count()
assert nulls == 0, f"Found {nulls} null account_ids"
What is anomaly alerting?
Anomaly alerting watches metrics over time and alerts when today deviates sharply from the norm. It catches problems that pass structural checks but are still wrong.
# Simple anomaly detection: flag if today's row count
# deviates more than 3 standard deviations from the 30-day average
import statistics
historical_counts = [100000, 102000, 98000, 101000, 99000] # last 30 days
mean_count = statistics.mean(historical_counts)
std_count = statistics.stdev(historical_counts)
today_count = spark.read.parquet("/data/bank/gold/transactions") \
.filter("txn_date = current_date()").count()
if abs(today_count - mean_count) > 3 * std_count:
alert(f"Anomaly: today's count {today_count} deviates from "
f"mean {mean_count} by more than 3 std deviations")
How do you quarantine bad records with a DLQ?
When a pipeline fails, the first principle is isolation: quarantine bad records to a dead-letter queue so a single poison record does not block the whole flow.
from pyspark.sql import functions as F
txns = spark.read.parquet("/data/bank/raw/transactions")
# Separate good and bad records
good_records = txns.filter(
F.col("account_id").isNotNull() &
F.col("amount").isNotNull() &
(F.col("amount") > 0)
)
bad_records = txns.filter(
F.col("account_id").isNull() |
F.col("amount").isNull() |
(F.col("amount") <= 0)
)
# Good records continue through the pipeline
good_records.write.format("delta").mode("append").save("/data/bank/silver/transactions")
# Bad records go to the DLQ for investigation
bad_records.write.format("delta").mode("append").save("/data/bank/dlq/transactions")
print(f"Processed: {good_records.count()}, Quarantined: {bad_records.count()}")
How do you make pipeline steps idempotent?
Idempotency means re-running a step produces the same result. This is achieved with upserts keyed by a stable ID, so after a failure you can safely re-run without duplicating data.
-- Idempotent upsert: re-running produces the same result
MERGE INTO gold.daily_account_totals AS target
USING (
SELECT account_id, txn_date, SUM(amount) AS total
FROM silver.transactions
WHERE txn_date = '2025-06-15'
GROUP BY account_id, txn_date
) AS source
ON target.account_id = source.account_id
AND target.txn_date = source.txn_date
WHEN MATCHED THEN UPDATE SET total = source.total
WHEN NOT MATCHED THEN INSERT (account_id, txn_date, total)
VALUES (source.account_id, source.txn_date, source.total);
-- Running this twice with the same input: same result, no duplicates
How do you backfill after fixing a failure?
Once the root cause is fixed, you reprocess the missed window by replaying from the durable log or re-reading the lake. Because steps are idempotent, the backfill corrects the data without double-counting.
# Backfill: reprocess the missed date range
missed_dates = ["2025-06-10", "2025-06-11", "2025-06-12"]
for date in missed_dates:
raw = spark.read.parquet(f"/data/bank/raw/transactions/date={date}")
processed = transform_transactions(raw)
# Idempotent write: MERGE or replaceWhere
processed.write.format("delta") \
.mode("overwrite") \
.option("replaceWhere", f"txn_date = '{date}'") \
.save("/data/bank/gold/daily_account_totals")
print(f"Backfill complete for {missed_dates}")
What should a senior engineer know?
- Validate early so bad data is caught at ingestion rather than after it has spread through bronze, silver, and gold layers.
- Decide per check whether to fail the whole batch (for a critical reconciliation mismatch) or quarantine and continue (for a few bad records).
- Make quality results observable with dashboards and a data catalog that records freshness and quality status.
- In a bank, the bias is toward failing safe: halting or quarantining rather than publishing data you cannot vouch for.
- Monitor the DLQ itself: a rising rate signals upstream data quality problems.
# Decision matrix for quality check failures
quality_rules = {
"schema_mismatch": "FAIL_BATCH", # Critical: halt everything
"null_account_id": "QUARANTINE", # Isolate bad records
"negative_amount": "QUARANTINE", # Isolate bad records
"row_count_mismatch": "FAIL_BATCH", # Critical: investigate
"duplicate_txn_id": "QUARANTINE", # Isolate duplicates
"anomaly_detected": "ALERT_AND_CONTINUE" # Flag but do not halt
}
Visual Overview
Data Quality Pipeline:
Source ──► ┌─────────────────┐
│ Schema Validate │
└────────┬────────┘
▼
┌─────────────────┐
│ Null / Range / │
│ Uniqueness Check │
└───┬─────────┬───┘
│ │
PASS │ │ FAIL
▼ ▼
┌──────────────────┐ ┌──────────────┐
│ Continue Pipeline │ │ Dead-Letter │
│ (Silver → Gold) │ │ Queue (DLQ) │
└────────┬─────────┘ └──────┬───────┘
▼ ▼
┌─────────────────┐ ┌──────────────┐
│ Reconciliation │ │ Alert + │
│ (count + sum) │ │ Investigate │
└────────┬────────┘ └──────────────┘
│
PASS │ FAIL → Halt batch
▼
┌─────────────────┐
│ Publish to Gold │
└─────────────────┘
Key Takeaway
Data quality in a pipeline is enforced with layered automated checks. It begins with schema validation, then null, range, and uniqueness checks on every record, followed by reconciliation that compares row counts and sum totals against the source to catch silent data loss. Data-contract tests formalize agreements between teams and run in CI, and anomaly alerting flags when today's metrics deviate sharply from the norm.
When a pipeline fails in production, the four principles are isolation (quarantine bad records to a DLQ), idempotency (upserts keyed by a stable ID so re-runs are safe), backfill (reprocess the missed window once the root cause is fixed), and alerting with clear ownership so someone accountable responds with context and a runbook.
At the National Bank of India, quality checks quarantined duplicate, null, and out-of-range records to a dead-letter queue, and a reconciliation step caught a run where records had been missed by comparing row counts against the source, halting the batch and alerting the on-call team rather than publishing an incomplete report. After fixing the root cause, we backfilled safely because every write was an idempotent upsert.
The bias in banking is always to fail safe: halt or quarantine rather than publish data you cannot vouch for, and keep full auditability so you can prove what was checked, what was quarantined, and how it was corrected.
Remember one sentence: enforce quality with layered checks at ingestion, quarantine failures to a DLQ, make every step idempotent, and always fail safe rather than publish data you cannot vouch for.
---