Dirty Data Scenario: 30% Malformed Timestamps

30% of incoming events have a malformed timestamp. Do you drop, quarantine, or repair — and what's the line between salvageable and toxic?

The Setup

You're running a real-time payment processing pipeline. Your events look like this:

{
    "transaction_id": "txn_20260530_001",
    "user_id": "usr_42981",
    "merchant_id": "mch_amazon",
    "amount": 89.50,
    "currency": "INR",
    "event_timestamp": "2026-05-30T13:45:22.000Z",   // ← Expected format
    "event_type": "purchase"
}

But after a partner API upgrade, 30% of events start arriving with broken timestamps:

// Type 1: Epoch milliseconds instead of ISO string (15% of bad data)
{"event_timestamp": 1780165200000}

// Type 2: Wrong date format (8% of bad data)
{"event_timestamp": "30/05/2026 13:45:22"}

// Type 3: Completely null (4% of bad data)
{"event_timestamp": null}

// Type 4: Future dates / obviously wrong (2% of bad data)
{"event_timestamp": "2099-01-01T00:00:00.000Z"}

// Type 5: Garbage / totally unparseable (1% of bad data)
{"event_timestamp": "not_a_date_lol"}

The Decision Framework: Drop vs. Quarantine vs. Repair

This is NOT a binary decision. You need a triage system — different levels of data corruption require different responses.

The Data Quality Triage Matrix

graph TD
    A["Incoming Event"] --> B{"Is timestamp<br/>parseable in<br/>ANY known format?"}
    B -->|Yes| C{"Is timestamp<br/>within valid range?<br/>(not future, not 1970)"}
    B -->|No| D{"Are ALL other<br/>fields valid?"}

    C -->|Yes| E["✅ REPAIR & PASS<br/>Normalize to ISO-8601"]
    C -->|No| F["⚠️ QUARANTINE<br/>Salvageable with context"]

    D -->|Yes| G["⚠️ QUARANTINE<br/>Use Kafka ingest timestamp"]
    D -->|No| H["❌ DEAD LETTER QUEUE<br/>Toxic / Unrecoverable"]

Step 1: Build the Data Quality Gate

The Data Quality Gate is a function that classifies every record into one of four categories:

from pyspark.sql import functions as F
from pyspark.sql.types import StringType
import re
from datetime import datetime

# ============================================================
# DATA QUALITY CLASSIFICATION UDF
# ============================================================
def classify_timestamp_quality(ts_value):
    """
    Classifies a timestamp value into quality tiers:

    - 'CLEAN':      Valid ISO-8601 timestamp within acceptable range
    - 'REPAIRABLE': Valid data in non-standard format (epoch, dd/MM/yyyy, etc.)
    - 'QUARANTINE': Null or out-of-range but other fields may be valid
    - 'TOXIC':      Completely unparseable garbage
    """
    if ts_value is None:
        return 'QUARANTINE'

    ts_str = str(ts_value).strip()

    if ts_str == '' or ts_str.lower() in ('null', 'none', 'undefined', 'nan'):
        return 'QUARANTINE'

    # Try ISO-8601 parsing
    iso_patterns = [
        r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}',  # 2026-05-30T13:45:22
        r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}',   # 2026-05-30 13:45:22
    ]
    for pattern in iso_patterns:
        if re.match(pattern, ts_str):
            try:
                dt = datetime.fromisoformat(ts_str.replace('Z', '+00:00'))
                # Range check: must be between 2020 and next year
                if datetime(2020, 1, 1) <= dt <= datetime(2027, 12, 31):
                    return 'CLEAN'
                else:
                    return 'QUARANTINE'  # Out of range (future or ancient)
            except:
                return 'REPAIRABLE'

    # Try epoch milliseconds (13-digit number)
    if ts_str.isdigit() and len(ts_str) == 13:
        return 'REPAIRABLE'

    # Try epoch seconds (10-digit number)
    if ts_str.isdigit() and len(ts_str) == 10:
        return 'REPAIRABLE'

    # Try common alternate formats: dd/MM/yyyy, MM-dd-yyyy, etc.
    alt_patterns = [
        r'^\d{2}/\d{2}/\d{4}',   # 30/05/2026
        r'^\d{2}-\d{2}-\d{4}',   # 30-05-2026
        r'^\w+ \d{1,2}, \d{4}',  # May 30, 2026
    ]
    for pattern in alt_patterns:
        if re.match(pattern, ts_str):
            return 'REPAIRABLE'

    # Nothing matched → toxic
    return 'TOXIC'

classify_udf = F.udf(classify_timestamp_quality, StringType())

Step 2: Apply the Classification

# Apply quality classification to every record
classified_df = raw_df.withColumn(
    "data_quality_tier",
    classify_udf(F.col("event_timestamp"))
)

# Audit the distribution
classified_df.groupBy("data_quality_tier").count().show()

Sample Output:

+-------------------+--------+
|data_quality_tier  |  count |
+-------------------+--------+
|CLEAN              | 700,000|
|REPAIRABLE         | 230,000|
|QUARANTINE         |  60,000|
|TOXIC              |  10,000|
+-------------------+--------+

Step 3: Route Each Tier to Its Destination

# ============================================================
# TIER 1: CLEAN DATA → Pass straight through
# ============================================================
clean_df = classified_df.filter(F.col("data_quality_tier") == "CLEAN") \
    .withColumn("event_timestamp", F.to_timestamp("event_timestamp")) \
    .drop("data_quality_tier")

# ============================================================
# TIER 2: REPAIRABLE DATA → Parse and normalize
# ============================================================
repairable_df = classified_df.filter(F.col("data_quality_tier") == "REPAIRABLE")

# Repair logic: try multiple parsing strategies
repaired_df = repairable_df.withColumn(
    "event_timestamp_repaired",
    F.coalesce(
        # Strategy 1: Try epoch milliseconds
        F.when(
            F.col("event_timestamp").rlike("^\\d{13}$"),
            F.from_unixtime(F.col("event_timestamp").cast("bigint") / 1000)
        ),
        # Strategy 2: Try epoch seconds
        F.when(
            F.col("event_timestamp").rlike("^\\d{10}$"),
            F.from_unixtime(F.col("event_timestamp").cast("bigint"))
        ),
        # Strategy 3: Try dd/MM/yyyy HH:mm:ss
        F.to_timestamp(F.col("event_timestamp"), "dd/MM/yyyy HH:mm:ss"),
        # Strategy 4: Try dd-MM-yyyy HH:mm:ss
        F.to_timestamp(F.col("event_timestamp"), "dd-MM-yyyy HH:mm:ss"),
        # Strategy 5: Try MM/dd/yyyy
        F.to_timestamp(F.col("event_timestamp"), "MM/dd/yyyy HH:mm:ss")
    )
).withColumn(
    "event_timestamp", F.col("event_timestamp_repaired")
).withColumn(
    "was_repaired", F.lit(True)
).drop("event_timestamp_repaired", "data_quality_tier")

# Merge clean + repaired into the main pipeline
main_pipeline_df = clean_df.withColumn("was_repaired", F.lit(False)) \
    .unionByName(repaired_df)

# ============================================================
# TIER 3: QUARANTINE → Salvage with Kafka ingest timestamp
# ============================================================
quarantine_df = classified_df.filter(F.col("data_quality_tier") == "QUARANTINE") \
    .withColumn(
        "event_timestamp", F.col("kafka_ingest_timestamp")  # Substitute!
    ) \
    .withColumn("timestamp_source", F.lit("KAFKA_INGEST_PROXY")) \
    .withColumn("quarantine_reason", 
        F.when(F.col("event_timestamp").isNull(), "NULL_TIMESTAMP")
         .when(F.year("event_timestamp") > 2027, "FUTURE_DATE")
         .otherwise("OUT_OF_RANGE")
    )

# Write quarantine to a separate location for review
quarantine_df.write \
    .mode("append") \
    .partitionBy("quarantine_reason") \
    .parquet("s3://datalake/quarantine/timestamps/")

# ============================================================
# TIER 4: TOXIC → Dead Letter Queue (DLQ)
# ============================================================
toxic_df = classified_df.filter(F.col("data_quality_tier") == "TOXIC") \
    .withColumn("dlq_reason", F.lit("UNPARSEABLE_TIMESTAMP")) \
    .withColumn("dlq_ingested_at", F.current_timestamp())

# Write to DLQ for root cause investigation
toxic_df.write \
    .mode("append") \
    .json("s3://datalake/dead_letter_queue/timestamps/")

Step 4: The Quarantine Review Process

Quarantined data isn't forgotten — it's actively reviewed and either repaired or escalated:

Quarantine Zone (s3://datalake/quarantine/timestamps/)
│
├── quarantine_reason=NULL_TIMESTAMP/
│   ├── part-00000.parquet    (40,000 records)
│   └── → Action: Substitute with Kafka ingest time, 
│                  merge back into main table after review
│
├── quarantine_reason=FUTURE_DATE/
│   ├── part-00000.parquet    (15,000 records)
│   └── → Action: Investigate clock skew on producer nodes,
│                  fix source system, reprocess
│
└── quarantine_reason=OUT_OF_RANGE/
    ├── part-00000.parquet    (5,000 records)
    └── → Action: Dates before 2020 suggest test data leaked
                  into production. Flag for data steward.

Automated Quarantine Repair Job (runs daily):

# Daily job: attempt to repair quarantined records
quarantine_data = spark.read.parquet("s3://datalake/quarantine/timestamps/")

# Re-attempt parsing with updated rules
re_repaired = quarantine_data.withColumn(
    "event_timestamp_fixed",
    F.coalesce(
        F.col("kafka_ingest_timestamp"),  # Use Kafka time as fallback
        F.current_timestamp()              # Last resort: use processing time
    )
)

# Records that can be rescued → merge into main table
rescued_count = re_repaired.filter(F.col("event_timestamp_fixed").isNotNull()).count()
print(f"✅ Rescued {rescued_count} records from quarantine")

# Write rescued data to main table using MERGE (Delta Lake)
re_repaired.createOrReplaceTempView("rescued")
spark.sql("""
    MERGE INTO main_payments AS target
    USING rescued AS source
    ON target.transaction_id = source.transaction_id
    WHEN NOT MATCHED THEN INSERT *
""")

The Line Between Salvageable and Toxic

This is the critical decision boundary:

SALVAGEABLE ← ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ → TOXIC

"Can I recover the        "Does the record
 BUSINESS VALUE of         still represent a
 this record?"             REAL event?"

  ┌─────────────────┐        ┌─────────────────┐
  │ YES to both     │        │ NO to either     │
  │ → REPAIR or     │        │ → DEAD LETTER    │
  │   QUARANTINE    │        │   QUEUE          │
  └─────────────────┘        └─────────────────┘

The Rules:

The Guiding Principle: If the record represents a real business event (a real payment, a real user action) and the corrupted field can be reasonably approximated, it's salvageable. If the record's authenticity itself is questionable, it's toxic.

Data Quality Metrics Dashboard

Track these metrics to monitor the health of your data quality gate:

# Compute data quality metrics per micro-batch
def compute_dq_metrics(batch_df, batch_id):
    total = batch_df.count()
    clean = batch_df.filter(F.col("data_quality_tier") == "CLEAN").count()
    repaired = batch_df.filter(F.col("data_quality_tier") == "REPAIRABLE").count()
    quarantined = batch_df.filter(F.col("data_quality_tier") == "QUARANTINE").count()
    toxic = batch_df.filter(F.col("data_quality_tier") == "TOXIC").count()

    metrics = {
        "batch_id": batch_id,
        "total_records": total,
        "clean_pct": round(clean / total * 100, 2) if total > 0 else 0,
        "repaired_pct": round(repaired / total * 100, 2) if total > 0 else 0,
        "quarantine_pct": round(quarantined / total * 100, 2) if total > 0 else 0,
        "toxic_pct": round(toxic / total * 100, 2) if total > 0 else 0,
    }

    print(f"[Batch {batch_id}] DQ Metrics: {metrics}")

    # ALERT: If toxic rate exceeds 5%, something is seriously wrong upstream
    if metrics["toxic_pct"] > 5:
        send_pager_alert(f"🚨 TOXIC DATA RATE AT {metrics['toxic_pct']}% — Investigate source!")

    # ALERT: If clean rate drops below 60%, quality degradation
    if metrics["clean_pct"] < 60:
        send_slack_alert(f"⚠️ Clean data rate dropped to {metrics['clean_pct']}%")

Alert Thresholds:

Summary

The Cardinal Rule of Data Quality: Never silently drop data. Every record that enters your system must exit through exactly one of three doors: the main pipeline, the quarantine, or the dead letter queue. No record should vanish without a trace.

Follow-Up Questions & Answers

Q1: What if the 30% bad data rate suddenly jumps to 80%? Should the pipeline keep running?

A: No. An 80% corruption rate means something catastrophic happened upstream — a deployment bug, a schema migration gone wrong, or a compromised data source. The pipeline should circuit-break.

# Circuit breaker pattern
def process_with_circuit_breaker(batch_df, batch_id):
    total = batch_df.count()
    toxic = batch_df.filter(F.col("data_quality_tier").isin("TOXIC", "QUARANTINE")).count()
    corruption_rate = toxic / total if total > 0 else 0

    if corruption_rate > 0.50:  # > 50% bad data = circuit OPEN
        send_pager_alert(f"🚨 CIRCUIT BREAKER OPEN: {corruption_rate*100:.1f}% corruption rate!")
        # STOP processing — don't pollute the data lake with mostly bad data
        raise Exception(f"Circuit breaker tripped: {corruption_rate*100:.1f}% corruption rate")

    # Normal processing continues...
    process_batch(batch_df, batch_id)

Why circuit-break instead of quarantining everything?

• Quarantining 80% of data means your main pipeline output is unreliable (only 20% of real events).
• Downstream dashboards would show a massive drop in activity, triggering false business alarms.
• It's better to pause and investigate than to produce a misleading partial view.

Q2: How do you handle timestamps in different timezones across different data sources?

A: This is one of the most common real-world dirty data problems. The fix has two parts:

Part 1: Normalize to UTC at ingestion time

from pyspark.sql import functions as F

# Convert all timestamps to UTC during ingestion
df_normalized = df.withColumn(
    "event_timestamp_utc",
    F.to_utc_timestamp(
        F.coalesce(
            F.to_timestamp("event_timestamp", "yyyy-MM-dd'T'HH:mm:ssXXX"),  # ISO with TZ
            F.to_timestamp("event_timestamp", "yyyy-MM-dd HH:mm:ss"),        # No TZ (assume source TZ)
        ),
        F.coalesce(F.col("source_timezone"), F.lit("Asia/Kolkata"))  # Default TZ if not provided
    )
)

Part 2: Store the original timezone for audit

df_final = df_normalized.withColumn(
    "original_timezone", F.coalesce(F.col("source_timezone"), F.lit("UNKNOWN"))
)

Rule: All internal processing uses UTC. The original timezone is preserved as metadata for debugging but never used in joins or aggregations.

Q3: What's the difference between schema validation and data validation? When does each apply?

A:

Schema failures → Dead Letter Queue (the record is structurally broken). Data failures → Quarantine (the record's structure is fine, but a value is suspicious).

# Schema validation (structure check)
schema_valid = raw_df.filter(
    F.col("transaction_id").isNotNull() &
    F.col("amount").cast("double").isNotNull() &
    F.col("user_id").isNotNull()
)

# Data validation (business rule check)
data_valid = schema_valid.filter(
    (F.col("amount") > 0) &
    (F.col("amount") < 10000000) &  # No single transaction above ₹1 crore
    (F.col("event_timestamp") > "2020-01-01")
)

Q4: How do you prevent the quarantine zone from growing indefinitely?

A: The quarantine zone needs its own lifecycle policy:

Quarantine Lifecycle:
Day 0:     Record enters quarantine
Day 1-7:   Automated repair attempts run daily
Day 7:     If still not repaired → escalate to data steward
Day 14:    If still unresolved → move to cold storage (S3 Glacier)
Day 90:    Auto-delete (with audit log retention)

# Quarantine cleanup job (runs weekly)
quarantine = spark.read.parquet("s3://datalake/quarantine/")

# Records older than 14 days → move to cold storage
old_records = quarantine.filter(
    F.datediff(F.current_date(), F.col("quarantined_at")) > 14
)
old_records.write.parquet("s3://datalake-glacier/quarantine_archive/")

# Delete from active quarantine
# (Use Delta Lake DELETE for ACID compliance)
spark.sql("DELETE FROM quarantine WHERE datediff(current_date(), quarantined_at) > 14")

Q5: In a streaming context, how do you handle late-arriving data that passes the quality gate but arrives after the window has closed?

A: This is the intersection of data quality and watermarking:

# Watermark: accept late events up to 2 hours after event time
windowed = parsed_stream \
    .withWatermark("event_timestamp", "2 hours") \
    .groupBy(
        F.window("event_timestamp", "1 hour"),
        "merchant_id"
    ).agg(F.sum("amount").alias("total_revenue"))

• Within watermark (< 2 hours late): Event is processed normally. Dashboard updates retroactively.
• Beyond watermark (> 2 hours late): Event is dropped by Spark — it cannot update a closed window.
• Fix for ultra-late data: Route events older than the watermark to a late-arrivals table and run a nightly batch reconciliation job.

Sub-Scenarios

Sub-Scenario A: The Malformed Data Is Coming From a Specific Partner API

Situation: After investigation, you discover that ALL 30% of malformed timestamps are coming from Partner X's API, which deployed a breaking change without notifying you.

Response:

1. Immediate: Add a partner-specific parsing rule to your quality gate that handles Partner X's format.
2. Short-term: Set up a schema contract with Partner X (e.g., using AsyncAPI or JSON Schema validation).
3. Long-term: Implement a schema registry (e.g., Confluent Schema Registry with Avro) that enforces schema compatibility at the Kafka producer level — before bad data even enters your pipeline.

# Partner-specific repair rule
df_repaired = df.withColumn(
    "event_timestamp",
    F.when(F.col("source_partner") == "PARTNER_X",
           F.from_unixtime(F.col("event_timestamp").cast("bigint") / 1000))
     .otherwise(F.to_timestamp("event_timestamp"))
)

Sub-Scenario B: Timestamps Are Valid But Semantically Wrong (Clock Drift)

Situation: A fleet of IoT devices has clock drift — their timestamps are consistently 3 hours ahead of real time. The records pass schema validation perfectly, but the data is semantically wrong.

Detection:

# Detect clock drift: compare event timestamp vs kafka ingest timestamp
df_drift = df.withColumn(
    "clock_drift_seconds",
    F.abs(F.unix_timestamp("event_timestamp") - F.unix_timestamp("kafka_timestamp"))
)

# Flag records with > 30 minutes of drift
suspicious = df_drift.filter(F.col("clock_drift_seconds") > 1800)
print(f"Records with clock drift > 30 min: {suspicious.count()}")

Fix: Use the Kafka ingest timestamp as the authoritative timestamp for devices with known clock drift, and file a ticket to fix the device firmware.

Sub-Scenario C: The Repair Logic Itself Has a Bug

Situation: Your epoch-to-ISO conversion has a bug — it's converting epoch seconds as if they're milliseconds, making all repaired timestamps land in 1970.

Safeguard: Always add a post-repair validation step:

# Post-repair range check
repaired = repaired.withColumn(
    "repair_valid",
    (F.year("event_timestamp_repaired") >= 2020) &
    (F.year("event_timestamp_repaired") <= 2027)
)

# If repair produced invalid timestamps, route to quarantine instead
bad_repairs = repaired.filter(~F.col("repair_valid"))
if bad_repairs.count() > 0:
    send_alert(f"⚠️ REPAIR BUG: {bad_repairs.count()} records have invalid post-repair timestamps!")

Sub-Scenario D: Downstream ML Model Is Sensitive to Timestamp Quality

Situation: A fraud detection ML model uses time_since_last_transaction as a feature. If timestamps are repaired with Kafka ingest time (which is always a few seconds late), the feature is systematically biased.

Solution: Add a timestamp_confidence column so the ML model can weight features appropriately:

df_with_confidence = df.withColumn(
    "timestamp_confidence",
    F.when(F.col("data_quality_tier") == "CLEAN", 1.0)
     .when(F.col("data_quality_tier") == "REPAIRABLE", 0.8)
     .when(F.col("data_quality_tier") == "QUARANTINE", 0.3)
     .otherwise(0.0)
)

The ML model can then discount features derived from low-confidence timestamps rather than treating all timestamps as equally reliable.

Sub-Scenario E: Regulatory Requirement — You Must Keep ALL Original Raw Data

Situation: Your compliance team says you can NEVER modify or discard the original raw event — even toxic ones — because auditors need to see exactly what was received.

Solution: Implement a raw archive + processed pipeline pattern:

Raw Events (Kafka)
    │
    ├──► S3 Raw Archive (IMMUTABLE — exact copy of every message)
    │    └── s3://archive/raw/yyyy/MM/dd/HH/
    │
    └──► Quality Gate → Clean / Quarantine / DLQ
         └── s3://datalake/processed/

The raw archive is write-once, never modified. The processed pipeline can drop, quarantine, or repair freely because the original is always recoverable.