Transformation: from hourly batch ETL to streaming with Kafka and Spark Structured Streaming

#Is streaming worth it?

A middle ground: run streaming code with trigger(availableNow=True) on a schedule — incremental, exactly-once, but without an always-on cluster.

#1. Read from Kafka

raw = (spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092")
  .option("subscribe", "orders")
  .option("startingOffsets", "earliest")
  .option("maxOffsetsPerTrigger", 500000)
  .load())

from pyspark.sql import functions as F, types as T
schema = T.StructType([
  T.StructField("order_id", T.StringType()), T.StructField("customer_id", T.StringType()),
  T.StructField("amount", T.DecimalType(18, 2)), T.StructField("event_time", T.TimestampType())])

orders = (raw.select(F.from_json(F.col("value").cast("string"), schema).alias("o"), "timestamp")
             .select("o.*", F.col("timestamp").alias("kafka_ts")))

#2. Bronze: append everything

(orders.writeStream.format("delta")
  .option("checkpointLocation", "/Volumes/main/stream/_chk/orders_bronze")
  .trigger(processingTime="1 minute")
  .toTable("main.bronze.orders"))

The checkpoint plus Delta's transactional writes give exactly-once into the table.

#3. Silver: dedupe and upsert

def upsert(batch_df, batch_id):
    (batch_df.dropDuplicates(["order_id"]).createOrReplaceTempView("updates"))
    batch_df.sparkSession.sql("""
      MERGE INTO main.silver.orders t USING updates s ON t.order_id = s.order_id
      WHEN MATCHED AND s.event_time > t.event_time THEN UPDATE SET *
      WHEN NOT MATCHED THEN INSERT *""")

(spark.readStream.table("main.bronze.orders")
  .writeStream.foreachBatch(upsert)
  .option("checkpointLocation", "/Volumes/main/stream/_chk/orders_silver")
  .trigger(processingTime="1 minute").start())

#4. Aggregations with late data

per_minute = (spark.readStream.table("main.silver.orders")
  .withWatermark("event_time", "15 minutes")
  .groupBy(F.window("event_time", "1 minute"), "customer_id")
  .agg(F.sum("amount").alias("revenue")))

The watermark bounds state size: events older than 15 minutes behind the max seen event time are dropped from aggregation. Choose it from real lateness data.

#5. Operate it

• Monitoring: query.lastProgress (input rate vs. processing rate, batch duration), Kafka consumer lag, alerts when processing rate < input rate for 10 minutes.
• Restarts: run as a job with retries; streams resume from the checkpoint.
• Changes: some query changes are incompatible with an existing checkpoint (changing stateful operations, sources). Plan a new checkpoint plus a backfill when that happens.
• Small files: optimise/auto-compaction on Delta targets.

Warning

Never delete or share a checkpoint directory between queries. A lost checkpoint means reprocessing or a gap.

#6. Retire the batch job

Run batch and stream in parallel, reconcile daily totals, then switch consumers and turn off the batch schedule.

#Checklist

• Freshness requirement justifies streaming (or use availableNow)
• Bronze append, silver MERGE, gold aggregates with watermarks
• One checkpoint per query on durable storage
• Lag and processing-rate alerts
• Reconciled against batch; batch retired