Custom Micro-Batch Writing with foreachBatch

How can we write streaming micro-batches to multiple destination tables or perform Delta Lake MERGE operations?

Production⏱ 3 Minutes⭐ 880 XP

🤔 THE QUESTION

How can we write streaming micro-batches to multiple destination tables or perform Delta Lake MERGE operations?

💡 WHAT IS IT?

foreachBatch(process_batch_fn) executes custom programmatic logic on each micro-batch DataFrame as standard batch data.

🎯 WHAT IS IT USED FOR?

Upserting stream data into Lakehouse tables (MERGE INTO) and writing simultaneously to Redis cache and Parquet lake.

💻 EXAMPLE

def process_batch(batch_df, batch_id):
    batch_df.persist()
    batch_df.write.mode("append").parquet("s3://lakehouse/silver/events/")
    batch_df.unpersist()

query = streaming_df.writeStream.foreachBatch(process_batch).option("checkpointLocation", "/tmp/cp/").start()

🎯 Mission Objectives

Practice typing production-grade PySpark code for Custom Micro-Batch Writing with foreachBatch.

✓ foreachBatch() custom sink
✓ Micro-batch processing function
✓ Multi-sink streaming publishing

🚀 Start Mission

← Prev (Stream-Stream Join with Watermarking and Interval Constraints)Next Lesson (Production Real-Time Streaming Pipeline) →