Recovery Scenario: Streaming Job Dead for 6 Hours Overnight

Your streaming job died for 6 hours overnight. Do you restart cleanly or backfill — and how do you make sure downstream dashboards don't double-count?

The Setup

It's 5:00 AM. Your PagerDuty fires:

🚨 ALERT: Spark Structured Streaming Job "payment_processor" FAILED
   Status:     TERMINATED (OOM on Executor 3)
   Last Alive:  11:02 PM (6 hours ago)
   Consumer Lag: 2.16 BILLION records behind
   Affected:    Real-time dashboards, fraud detection, settlement reports

Your pipeline architecture:

Kafka (payment_events, 120 partitions)
    │
    ▼
Spark Structured Streaming
    │
    ├── Output 1: Delta Lake (s3://datalake/payments/)
    ├── Output 2: Real-time Dashboard (Grafana via ClickHouse)
    └── Output 3: Fraud Detection Alerts (SNS → Review Queue)

The 6-hour gap means:

• 2.16 billion events are sitting unprocessed in Kafka
• Real-time dashboards show zero transactions for the last 6 hours
• Fraud alerts were not generated — fraudsters had a free window
• Morning settlement reports will be incomplete

The Critical Decision: Restart vs. Backfill

graph TD
    A["Job Died 6 Hours Ago"] --> B{"Do you have<br/>checkpoints?"}
    B -->|"Yes"| C{"Is Kafka retention<br/>> 6 hours?"}
    B -->|"No"| D["❌ DATA LOST<br/>Backfill from S3 archive"]

    C -->|"Yes"| E["✅ CHECKPOINT RESTART<br/>Replay from last offset"]
    C -->|"No"| F["⚠️ PARTIAL LOSS<br/>Backfill what you can"]

    E --> G{"Can your pipeline<br/>handle 6h of backlog<br/>+ live traffic?"}
    G -->|"Yes"| H["Option A: Let it catch up"]
    G -->|"No"| I["Option B: Separate backfill job"]

Option A: Checkpoint Restart (The Default Path)

If your streaming job was properly configured with checkpoints, Spark saved the last committed Kafka offsets. On restart, it picks up exactly where it left off.

How Checkpoints Work

Normal operation (before crash):
    Micro-batch 1000: Read offsets 5,000,000 - 5,500,000 → Processed → Checkpoint saved
    Micro-batch 1001: Read offsets 5,500,001 - 6,000,000 → Processed → Checkpoint saved
    Micro-batch 1002: Read offsets 6,000,001 - 6,500,000 → Processing...
    ❌ OOM CRASH at 11:02 PM

    Checkpoint state on disk: "Last committed = offset 6,000,000"

After restart (5:00 AM):
    Spark reads checkpoint: "Resume from offset 6,000,001"
    Micro-batch 1002 (replayed): Read offsets 6,000,001 - 6,500,000 → Reprocessed
    Micro-batch 1003: Read offsets 6,500,001 - 7,000,000 → New data
    ... continues catching up through 6 hours of backlog

Restart Command

# The streaming job code — EXACTLY the same as before
# Spark will read the checkpoint and resume automatically

query = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092") \
    .option("subscribe", "payment_events") \
    .option("maxOffsetsPerTrigger", 2000000) \
    .option("failOnDataLoss", "false") \   # ← CRITICAL: Don't crash if offsets expired
    .load() \
    .writeStream \
    .format("delta") \
    .option("checkpointLocation", "s3://pipeline/checkpoints/payments") \  # ← Same checkpoint!
    .option("mergeSchema", "true") \
    .trigger(processingTime="30 seconds") \
    .start("s3://datalake/payments/")

The Catch-Up Math

Backlog:           2.16 billion events (6 hours × 100K/s)
Normal throughput: 100K events/sec (500K per 5-sec batch)
Backfill rate:     Can process 200K/sec with maxOffsetsPerTrigger=2M per 10-sec trigger

Time to catch up:
    2,160,000,000 events / 200,000 events/sec = 10,800 seconds = 3 hours

BUT: New events are still arriving at 100K/sec!
    Net catch-up rate: 200K/sec - 100K/sec = 100K/sec
    Adjusted time: 2,160,000,000 / 100,000 = 21,600 sec = 6 hours to fully catch up

Problem: With a single streaming job trying to both catch up AND process live traffic, it takes 6 hours to clear the backlog. Dashboards remain stale until then.

Option B: Separate Backfill Job (The Production Path)

For faster recovery, run two jobs in parallel:

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  JOB 1: LIVE STREAMING (starts immediately)                 │
│  ├── Reads from Kafka starting at LATEST offset             │
│  ├── Processes only NEW events from 5:00 AM onward          │
│  └── Dashboards show real-time data immediately             │
│                                                             │
│  JOB 2: BACKFILL BATCH (separate job, runs in parallel)     │
│  ├── Reads the 6-hour gap: 11:00 PM to 5:00 AM offsets     │
│  ├── Processes as a batch job (not streaming)               │
│  ├── Writes to the SAME Delta table with MERGE              │
│  └── Completes in ~1 hour (dedicated resources)             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Job 1: Live Stream (Restart from Latest)

# Start a NEW streaming job from the latest Kafka offset
# This ensures dashboards get real-time data IMMEDIATELY

live_query = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092") \
    .option("subscribe", "payment_events") \
    .option("startingOffsets", "latest") \          # ← Skip the backlog
    .option("maxOffsetsPerTrigger", 500000) \
    .load() \
    .writeStream \
    .format("delta") \
    .option("checkpointLocation", "s3://pipeline/checkpoints/payments_live_v2") \  # NEW checkpoint!
    .trigger(processingTime="10 seconds") \
    .start("s3://datalake/payments/")

Job 2: Backfill Batch Job

# Read ONLY the missed 6-hour window from Kafka as a BATCH read

# Step 1: Determine the exact offset range for the gap
from pyspark.sql import functions as F
import json

# The gap: 11:00 PM to 5:00 AM 
# Use specific offsets from the old checkpoint + current earliest available
starting_offsets = {
    "payment_events": {
        "0": 6000001,   # Last committed offset from checkpoint
        "1": 5800001,
        "2": 6200001,
        # ... all 120 partitions
    }
}

ending_offsets = {
    "payment_events": {
        "0": 24000000,   # Current latest offset at 5:00 AM
        "1": 23500000,
        "2": 24800000,
        # ... all 120 partitions
    }
}

# Step 2: Batch-read the exact gap from Kafka
backfill_df = spark.read \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092") \
    .option("subscribe", "payment_events") \
    .option("startingOffsets", json.dumps(starting_offsets)) \
    .option("endingOffsets", json.dumps(ending_offsets)) \
    .load()

print(f"📦 Backfill records to process: {backfill_df.count():,}")

# Step 3: Transform (same logic as streaming job)
parsed_backfill = backfill_df.select(
    F.from_json(F.col("value").cast("string"), payment_schema).alias("payment")
).select("payment.*")

# Apply the same transformations as the streaming pipeline
transformed = transform_payments(parsed_backfill)

# Step 4: Write using MERGE to prevent duplicates
transformed.createOrReplaceTempView("backfill_data")

spark.sql("""
    MERGE INTO delta.`s3://datalake/payments/` AS target
    USING backfill_data AS source
    ON target.transaction_id = source.transaction_id
    WHEN NOT MATCHED THEN 
        INSERT *
    WHEN MATCHED THEN 
        UPDATE SET *
""")

print("✅ Backfill complete. All 6 hours of missed data merged.")

Preventing Double-Counting: The Deduplication Strategy

The biggest risk during recovery is double-counting — processing the same event twice and inflating metrics on downstream dashboards.

Why Double-Counting Happens

Scenario: Micro-batch 1002 was PARTIALLY processed before the crash.

Records 6,000,001 - 6,250,000: Written to Delta ✅
Records 6,250,001 - 6,500,000: IN MEMORY when crash happened ❌

On restart, Spark replays from offset 6,000,001 (last checkpoint).
Records 6,000,001 - 6,250,000: WRITTEN AGAIN → DUPLICATES!

Solution 1: Idempotent Writes with MERGE (Delta Lake)

The gold standard for exactly-once semantics:

def write_with_dedup(batch_df, batch_id):
    """Write micro-batch to Delta Lake using MERGE for idempotency."""

    if batch_df.isEmpty():
        return

    batch_df.createOrReplaceTempView(f"batch_{batch_id}")

    spark.sql(f"""
        MERGE INTO delta.`s3://datalake/payments/` AS target
        USING batch_{batch_id} AS source
        ON target.transaction_id = source.transaction_id
           AND target.event_timestamp = source.event_timestamp
        WHEN NOT MATCHED THEN 
            INSERT *
    """)

# Use foreachBatch for idempotent writes
query = parsed_stream.writeStream \
    .foreachBatch(write_with_dedup) \
    .option("checkpointLocation", "s3://pipeline/checkpoints/payments") \
    .trigger(processingTime="30 seconds") \
    .start()

How MERGE prevents duplicates:

Batch 1002 (replayed after restart):
    Record txn_001: MERGE checks → EXISTS in target → SKIP (no insert)
    Record txn_002: MERGE checks → EXISTS in target → SKIP
    ...
    Record txn_250001: MERGE checks → NOT in target → INSERT ✅
    Record txn_250002: MERGE checks → NOT in target → INSERT ✅

Solution 2: Watermark-Based Deduplication (Streaming Native)

Spark Structured Streaming has built-in deduplication with watermarks:

# Deduplicate within a 24-hour window based on transaction id
deduped_stream = parsed_stream \
    .withWatermark("event_timestamp", "24 hours") \
    .dropDuplicates(["transaction_id"])

How it works:

Watermark = current event time - 24 hours

Any event with event_timestamp OLDER than the watermark is considered "late"
and is dropped if a duplicate already exists in state.

This prevents double-counting even if the same event arrives twice
(once from the partial batch, once from the replay).

Solution 3: Dashboard-Level Deduplication

As a final safety net, the dashboard query itself should deduplicate:

-- Grafana / ClickHouse dashboard query with deduplication
SELECT 
    date_trunc('hour', event_timestamp) AS hour,
    COUNT(DISTINCT transaction_id) AS unique_transactions,  -- ← DISTINCT!
    SUM(amount) AS total_revenue
FROM payments
WHERE event_date >= '2026-05-30'
GROUP BY 1
ORDER BY 1;

Key: Use COUNT(DISTINCT transaction_id) instead of COUNT(*). This makes the dashboard immune to duplicates regardless of what happens upstream.

The Recovery Checklist

When a streaming job dies overnight, follow this sequence:

╔══════════════════════════════════════════════════════════════════╗
║                  RECOVERY PLAYBOOK                              ║
╠══════════════════════════════════════════════════════════════════╣
║                                                                  ║
║  ⏱️ MINUTE 0-5: ASSESS                                          ║
║  ├── Check: When did the job die? (Spark UI history server)      ║
║  ├── Check: Is Kafka retention sufficient? (kafka-configs.sh)    ║
║  ├── Check: Is checkpoint intact? (ls checkpoint directory)      ║
║  └── Check: What caused the failure? (CloudWatch / driver logs)  ║
║                                                                  ║
║  ⏱️ MINUTE 5-10: START LIVE STREAM                               ║
║  ├── Start Job 1: Live stream from "latest" offset               ║
║  ├── Verify: Dashboard shows new real-time data                  ║
║  └── Verify: Fraud detection alerts flowing again                ║
║                                                                  ║
║  ⏱️ MINUTE 10-30: LAUNCH BACKFILL                                ║
║  ├── Calculate: Exact offset range for the gap                   ║
║  ├── Start Job 2: Batch backfill with MERGE INTO                 ║
║  └── Monitor: Backfill progress in Spark UI                     ║
║                                                                  ║
║  ⏱️ HOUR 1-2: VERIFY                                             ║
║  ├── Confirm: Backfill job completed successfully                ║
║  ├── Audit: Row count comparison (Kafka offsets vs Delta table)  ║
║  ├── Validate: No duplicates (SELECT COUNT(*) vs COUNT(DISTINCT))║
║  └── Check: Dashboard data is continuous (no gaps in timeline)   ║
║                                                                  ║
║  ⏱️ HOUR 2+: CLEANUP                                             ║
║  ├── Stop Job 1 (live stream with latest offsets)                ║
║  ├── Restart: Original streaming job from checkpoint             ║
║  ├── Confirm: Consumer lag is zero                               ║
║  └── Post-mortem: Fix the root cause (OOM → increase memory)    ║
║                                                                  ║
╚══════════════════════════════════════════════════════════════════╝

Verification Queries

Check for Data Gaps

-- Verify no hourly gaps exist after recovery
SELECT 
    date_trunc('hour', event_timestamp) AS hour,
    COUNT(*) AS txn_count
FROM delta.`s3://datalake/payments/`
WHERE event_date = '2026-05-30'
GROUP BY 1
ORDER BY 1;

-- Expected: Every hour should have data (no zeros)
-- Hour     | txn_count
-- 22:00    | 350,000
-- 23:00    | 380,000   ← Was missing, now backfilled ✅
-- 00:00    | 310,000   ← Was missing, now backfilled ✅
-- 01:00    | 280,000   ← Was missing, now backfilled ✅
-- ...
-- 05:00    | 340,000   ← Live stream picked up here
-- 06:00    | 390,000

Check for Duplicates

# Verify no duplicates exist
payments = spark.read.format("delta").load("s3://datalake/payments/")

total_rows = payments.filter(F.col("event_date") == "2026-05-30").count()
unique_txns = payments.filter(F.col("event_date") == "2026-05-30") \
    .select("transaction_id").distinct().count()

print(f"Total rows:       {total_rows:,}")
print(f"Unique txn IDs:   {unique_txns:,}")
print(f"Duplicate rate:   {(total_rows - unique_txns) / total_rows * 100:.4f}%")

# Expected output:
# Total rows:       8,640,000
# Unique txn IDs:   8,640,000
# Duplicate rate:   0.0000%   ← ✅ Zero duplicates!

Check Consumer Lag

# Verify Kafka consumer lag is back to zero
kafka-consumer-groups.sh \
    --bootstrap-server broker1:9092 \
    --describe \
    --group payment_processor_group

# Expected: LAG = 0 for all partitions
# TOPIC             PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# payment events    0          24000000        24000012        12
# payment events    1          23500000        23500008        8
# ...
# Total lag: ~100 (near-zero, processing in real-time) ✅

Prevention: Avoiding Future Overnight Failures

# Health check script (runs via cron every 5 minutes)
import subprocess
import json

result = subprocess.run(
    ["spark-submit", "--master", "yarn", "--status", "application_id"],
    capture_output=True, text=True
)

if "RUNNING" not in result.stdout:
    send_pager_alert("🚨 payment_processor streaming job is NOT RUNNING!")
    # Auto-restart
    subprocess.run(["spark-submit", "payment_processor.py"])

Summary

The Golden Rule of Recovery: Resume live traffic FIRST (dashboards matter), backfill the gap SECOND (completeness matters), and deduplicate ALWAYS (accuracy matters).

Follow-Up Questions & Answers

Q1: What if the checkpoint itself is corrupted or deleted?

A: If checkpoints are gone, Spark has no idea where it left off. You have two options:

Option A: Start from earliest available Kafka offset

query = spark.readStream \
    .format("kafka") \
    .option("startingOffsets", "earliest") \  # Read everything available in Kafka
    .option("failOnDataLoss", "false") \
    .load()

⚠️ Risk: You'll reprocess ALL data currently in Kafka retention (potentially days' worth). You MUST use idempotent writes (MERGE INTO) to prevent duplicates.

Option B: Start from a specific timestamp

# Start from a known point in time (Kafka 0.10.1+)
query = spark.readStream \
    .format("kafka") \
    .option("startingOffsetsByTimestamp", 
            '{"payment_events":{"0":1780165200000,"1":1780165200000}}') \
    .load()

Prevention: Store checkpoints on durable, versioned storage:

• S3 with versioning enabled (recover deleted checkpoints from previous versions)
• HDFS with replication factor 3
• Never store checkpoints on local executor disks — they're lost if the node dies

Q2: How does MERGE INTO actually prevent double-counting? Isn't it expensive?

A: MERGE INTO performs an upsert — it inserts new records and updates existing ones:

MERGE INTO payments AS target
USING backfill_batch AS source
ON target.transaction_id = source.transaction_id
WHEN MATCHED THEN UPDATE SET *    -- Overwrite (idempotent)
WHEN NOT MATCHED THEN INSERT *    -- Insert new records

Cost analysis:

• MERGE requires a join between the source batch and the target table on the merge key.
• For a 10-million-row batch against a 1-billion-row target, this join scans only the relevant partitions (if the target is partitioned by date).
• Optimization: Partition the Delta table by event_date and add a filter:

spark.sql("""
    MERGE INTO payments AS target
    USING backfill_batch AS source
    ON target.transaction_id = source.transaction_id
       AND target.event_date = source.event_date   -- Partition pruning!
    WHEN NOT MATCHED THEN INSERT *
""")

This limits the MERGE to scanning only the 1 day's partition (~10M rows) instead of the entire table.

Q3: What if Kafka retention expired and the 6-hour data is GONE from Kafka?

A: If Kafka retention is too short, the data is irretrievably lost FROM KAFKA. But you may have other options:

# Recovery from S3 raw archive
archive_df = spark.read \
    .format("json") \
    .load("s3://raw-archive/payment_events/2026/05/30/23/",  # 11 PM
           "s3://raw-archive/payment_events/2026/05/31/00/",  # 12 AM
           "s3://raw-archive/payment_events/2026/05/31/01/",  # 1 AM
           "s3://raw-archive/payment_events/2026/05/31/02/",  # 2 AM
           "s3://raw-archive/payment_events/2026/05/31/03/",  # 3 AM
           "s3://raw-archive/payment_events/2026/05/31/04/")  # 4 AM

# Process and merge into the main table
transformed = transform_payments(archive_df)
transformed.createOrReplaceTempView("archive_recovery")
spark.sql("MERGE INTO payments USING archive_recovery ON ...")

Q4: During the backfill, won't the parallel live stream and batch job create a write conflict on the same Delta table?

A: Delta Lake handles concurrent writes using Optimistic Concurrency Control (OCC):

• Both jobs can write to the same Delta table simultaneously.
• Delta uses transaction logs (_delta_log/) to ensure ACID consistency.
• If both jobs try to modify the same partition at the same time, one will retry automatically (delta.retryCommit).
• Append-only writes (both jobs just INSERT) almost never conflict.
• MERGE operations have a higher conflict chance but Delta handles retries automatically.

# Delta concurrency settings for parallel writers
spark.conf.set("spark.databricks.delta.retryWriteConflict.enabled", "true")
spark.conf.set("spark.databricks.delta.retryWriteConflict.limit", 3)

Best practice: Partition the Delta table by event_date or event_hour. The live stream writes to the current hour's partition while the backfill writes to past hours — no physical overlap.

Q5: How do you handle the situation where the job keeps crashing in a restart loop (crash → restart → crash)?

A: This is a crash loop — the root cause of the crash (e.g., a poison pill record) is still in Kafka, so every restart immediately hits the same bad record and crashes again.

Fix: Skip the poison pill:

# Option 1: Use failOnDataLoss=false and manually advance the offset
# past the bad records using kafka-consumer-groups.sh

# Option 2: Add a try/except in foreachBatch
def safe_process(batch_df, batch_id):
    try:
        process_batch(batch_df, batch_id)
    except Exception as e:
        # Log the error, send alert, but DON'T crash the whole job
        bad_records = batch_df.limit(10).toPandas()
        send_alert(f"⚠️ Batch {batch_id} failed: {e}. Sample bad records: {bad_records}")
        # Write bad records to DLQ instead of crashing
        batch_df.write.mode("append").json("s3://datalake/dead_letter_queue/crash_loop/")

# Option 3: Reset the consumer group offset
# kafka-consumer-groups.sh --reset-offsets --to-offset <offset+1> --execute

Sub-Scenarios

Sub-Scenario A: The Outage Was Caused by a Bad Deployment, Not an OOM

Situation: At 11 PM, someone deployed a new version of the streaming job with a bug. The job crashed immediately. The bug was fixed at 5 AM.

Additional concern: The NEW code may process data differently than the OLD code. If you backfill the 6-hour gap with the NEW code, results may be subtly different from what the old code would have produced.

Solution: Document the "version boundary" in the data:

backfill_df = backfill_df.withColumn("processing_version", F.lit("v2.1.0"))
backfill_df = backfill_df.withColumn("backfill_flag", F.lit(True))

This lets downstream analysts know that data in the 11 PM - 5 AM window was processed with a different code version and may have slight differences.

Sub-Scenario B: Multiple Streaming Jobs Failed (Not Just One)

Situation: A Kubernetes node failure took down 5 streaming jobs simultaneously: payments, refunds, fraud detection, settlement, and notifications.

Recovery priority order:

Priority 1: Fraud Detection    → Resume IMMEDIATELY (security critical)
Priority 2: Payments Pipeline   → Resume within 5 min (revenue critical)
Priority 3: Settlement Reports  → Backfill within 2 hours (SLA: daily)
Priority 4: Refunds Pipeline    → Backfill within 4 hours
Priority 5: Notifications      → Backfill within 24 hours (lowest priority)

Key principle: Not all streaming jobs have the same SLA. Prioritize by business impact.

Sub-Scenario C: The Dashboard Team Notices the Gap Before You Do

Situation: At 6 AM, the analytics team Slacks you: "Why does the revenue dashboard show zero for 11 PM to 5 AM?" They're panicking because it looks like the company made zero revenue for 6 hours.

Immediate response:

1. Acknowledge and communicate: "The streaming pipeline was down from 11 PM to 5 AM. Data is being backfilled. Dashboards will be accurate within 2 hours."
2. Add a banner to the dashboard: Use a dashboard annotation to mark the affected time window:

-- Dashboard query with gap annotation
SELECT 
    hour,
    total_revenue,
    CASE 
        WHEN hour BETWEEN '2026-05-30 23:00' AND '2026-05-31 05:00'
        THEN '⚠️ Backfill in progress'
        ELSE '✅ Real-time'
    END AS data_status
FROM hourly_revenue

Sub-Scenario D: Exactly-Once Semantics Across Multiple Sinks

Situation: Your streaming job writes to THREE sinks: Delta Lake, ClickHouse, and SNS. How do you ensure exactly-once across ALL of them?

The hard truth: True exactly-once across multiple heterogeneous sinks is impossible without a distributed transaction coordinator. Instead, use at-least-once + idempotent sinks:

def write_to_all_sinks(batch_df, batch_id):
    # Sink 1: Delta Lake (idempotent via MERGE)
    batch_df.createOrReplaceTempView(f"batch_{batch_id}")
    spark.sql(f"MERGE INTO payments USING batch_{batch_id} ON ...")

    # Sink 2: ClickHouse (idempotent via ReplacingMergeTree)
    # ClickHouse's ReplacingMergeTree deduplicates by primary key
    batch_df.write \
        .format("jdbc") \
        .option("dbtable", "payments") \
        .option("driver", "com.clickhouse.jdbc.ClickHouseDriver") \
        .mode("append") \
        .save()

    # Sink 3: SNS (NOT idempotent — use dedup at consumer side)
    # Add a unique message_dedup_id to prevent duplicate notifications
    for row in batch_df.select("transaction_id", "alert_message").collect():
        sns_client.publish(
            TopicArn="arn:aws:sns:...:fraud_alerts",
            Message=row["alert_message"],
            MessageDeduplicationId=row["transaction_id"]  # SNS FIFO dedup
        )

Sub-Scenario E: The Backfill Job Is Too Slow (2 Billion Records in Kafka)

Situation: The 6-hour gap has 2 billion records. Your backfill batch job is estimated to take 8 hours — which means dashboards stay incomplete until the afternoon.

Speed optimizations:

1. Parallelize by partition: Split the backfill into 120 parallel jobs (one per Kafka partition).
2. Use more DPUs: Temporarily double the cluster size for the backfill.
3. Skip non-critical transformations: Run a "fast backfill" that only does essential transformations, then run a "full enrichment" job later.
4. Prioritize partitions: Process the most recent hours first (4-5 AM) so dashboards show the most recent data fastest.

# Prioritized backfill: process most recent hours first
for hour in reversed(range(23, 29)):  # 5 AM, 4 AM, 3 AM, 2 AM, 1 AM, 12 AM, 11 PM
    actual_hour = hour % 24
    backfill_hour(actual_hour)
    print(f"✅ Hour {actual_hour}:00 backfilled — dashboard updated")