Handling late / out-of-order data and reprocessing in a streaming pipeline

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window, sum as _sum

spark = SparkSession.builder.appName("BankEventTime").getOrCreate()

# Each transaction carries its event-time (txn_timestamp)
txns = spark.readStream.format("kafka") \
    .option("subscribe", "bank-transactions") \
    .load()

parsed = txns.selectExpr(
    "CAST(value AS STRING)",
    "CAST(timestamp AS TIMESTAMP) AS kafka_time"  # processing-time
).select(
    col("value").alias("raw"),
    col("kafka_time"),
    # Event-time from the record itself
    col("value").getField("txn_timestamp").cast("timestamp").alias("event_time")
)

# Using event-time ensures a transaction stamped 10:00
# is counted in the 10:00 window even if it arrives at 10:04

# Spark Structured Streaming: 5-second watermark
watermarked = parsed.withWatermark("event_time", "5 seconds")

# Flink equivalent:
# stream.assignTimestampsAndWatermarks(
#     WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))
# )

import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

// Flink: 1-minute tumbling windows with 30-second allowed lateness
DataStream<TransactionTotal> totals = txnStream
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((txn, ts) -> txn.getTimestamp()))
    .keyBy(Transaction::getAccountId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.seconds(30))
    .sideOutputLateData(lateOutputTag)
    .sum("amount");

// Late events arriving within 30s after window fires: update result
// Late events arriving after 30s: routed to side output for investigation

from pyspark.sql.functions import window, sum as _sum

# Tumbling window: fixed, non-overlapping 1-minute buckets
tumbling = watermarked.groupBy(
    window("event_time", "1 minute"),
    "branch_code"
).agg(_sum("amount").alias("total_amount"))

# Sliding window: 1-minute window every 10 seconds (overlapping)
sliding = watermarked.groupBy(
    window("event_time", "1 minute", "10 seconds"),
    "branch_code"
).agg(_sum("amount").alias("rolling_total"))

# Session window: group bursts separated by 5-minute gaps
# (Flink natively supports session windows)

Tumbling (1 min):  |--10:00--|--10:01--|--10:02--|
Sliding (1 min/10s): |--10:00--|
                       |--10:00:10--|
                         |--10:00:20--|
Session (5 min gap): |--burst1--|  gap  |--burst2--|

// Flink: capture very late data in a side output
OutputTag<Transaction> lateOutputTag = new OutputTag<>("late-transactions") {};

SingleOutputStreamOperator<TransactionTotal> result = txnStream
    .keyBy(Transaction::getAccountId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.seconds(30))
    .sideOutputLateData(lateOutputTag)
    .sum("amount");

// Process late data separately
DataStream<Transaction> lateData = result.getSideOutput(lateOutputTag);
lateData.addSink(new LateDataSink());  // Write to DLQ or reconciliation table

# Replay from Kafka at a specific offset for backfill
backfill = spark.read.format("kafka") \
    .option("subscribe", "bank-transactions") \
    .option("startingOffsets", '{"bank-transactions":{"0":1000,"1":2000}}') \
    .option("endingOffsets", '{"bank-transactions":{"0":5000,"1":6000}}') \
    .load()

# Or replay from the lake
backfill_lake = spark.read.parquet("/data/bank/raw/transactions/date=2025-06-*")

# Recompute and write with idempotent upsert
backfill_lake.groupBy("account_id", "txn_date").agg(_sum("amount").alias("total")) \
    .write.format("delta").mode("overwrite") \
    .option("replaceWhere", "txn_date >= '2025-06-01' AND txn_date < '2025-07-01'") \
    .save("/data/bank/gold/account_daily_totals")

-- Idempotent upsert in Delta Lake
MERGE INTO gold.account_daily_totals AS target
USING backfill_results 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);

-- Replay produces the same result: no double-counting

# Version your transformation logic for reproducible backfills
TRANSFORM_VERSION = "v2.1"

def compute_daily_totals(df):
    """Deterministic transformation for bank daily totals."""
    return df.groupBy("account_id", "txn_date") \
        .agg(_sum("amount").alias("total")) \
        .withColumn("transform_version", lit(TRANSFORM_VERSION))