Reading and writing streams — Kafka, files, and Delta

What we're doing

You'll learn the mechanics of file sources, Kafka sources and sinks, and Delta Lake as a streaming destination. Then you'll build a real end-to-end streaming pipeline with checkpointing so it can survive failures.

Step 1: File sources

Common file source setup:

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("event_type", StringType()),
    StructField("user_id", IntegerType()),
    StructField("value", IntegerType())
])

stream_df = spark.readStream \
    .schema(schema) \
    .csv("/path/to/landing/directory")

Key details for file sources:

• Schema is mandatory — Spark can't infer it from streaming data
• Files are processed once — Spark tracks which files it has seen, so you can't reprocess by dropping the same file twice
• File formats supported — CSV, JSON, Parquet, ORC, plus text
• Partial file writes — files being written should be moved into the directory atomically (write elsewhere, then mv)

Step 2: Kafka source

kafka_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "user-events") \
    .option("startingOffsets", "earliest") \
    .load()

Key options:

• kafka.bootstrap.servers — where your Kafka brokers are
• subscribe — the topic to read from (can be multiple, comma-separated, or use subscribePattern for regex)
• startingOffsets — where to start reading. earliest reads all history, latest only reads new messages

The resulting DataFrame has a fixed schema regardless of what's in your topic:

• key (binary) — message key
• value (binary) — message value (your actual payload)
• topic (string)
• partition (int)
• offset (long)
• timestamp (timestamp)

Almost always you need to cast value to a string and parse it as JSON to get to your actual data:

from pyspark.sql.functions import col, from_json

parsed = kafka_stream.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

Step 3: Publishing back to Kafka

You can also write streaming results back to Kafka.

from pyspark.sql.functions import to_json, struct

query = enriched.select(
    col("user_id").cast("string").alias("key"),
    to_json(struct("*")).alias("value")
).writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "enriched-events") \
    .option("checkpointLocation", "/tmp/kafka-checkpoint") \
    .start()

Step 4: Delta Lake — the sink built for streaming

What Delta adds over plain Parquet:

• ACID transactions — writes either fully succeed or fully fail. No partial writes corrupting your table.
• Schema enforcement — Delta rejects rows that don't match the table schema. Prevents bad data from getting through.
• Time travel — read the table as it existed at any point in the past.
• Streaming + batch compatibility — the same Delta table can be read as a stream or as a batch table.
• Efficient updates and deletes — plain Parquet doesn't support these; Delta does.

Writing a stream to Delta:

query = df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/tmp/delta-checkpoint") \
    .start("/tmp/delta-output")

Reading a Delta table as a batch:

batch_df = spark.read.format("delta").load("/tmp/delta-output")

Reading a Delta table as a stream — for downstream consumers:

stream_df = spark.readStream.format("delta").load("/tmp/delta-output")

Step 5: Checkpointing — how streams survive failures

Spark writes progress information, which offsets it has processed, which micro-batches completed, current aggregation state — to a checkpoint directory. When the query restarts, it reads the checkpoint and resumes from where it stopped.

.option("checkpointLocation", "/tmp/my-checkpoint")

Rules for checkpoint locations:

• Must be a durable filesystem — HDFS, S3, or persistent local disk. Not /tmp in production.
• Must be unique per query — never share checkpoints between two different queries.
• Never delete manually while the query is running — deletion corrupts the query.

Step 6: Start PySpark with Delta support

We need to include the Delta package when starting PySpark:

pyspark --master spark://localhost:7077 \
  --packages io.delta:delta-spark_2.12:3.2.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

Step 7: Set up directories

In another terminal:

mkdir -p ~/streaming-input
mkdir -p ~/delta-output
mkdir -p ~/delta-checkpoint

---

Step 8: File source to Delta sink

Back in PySpark:

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("event_type", StringType()),
    StructField("user_id", IntegerType()),
    StructField("value", IntegerType())
])

stream_df = spark.readStream \
    .schema(schema) \
    .csv("/home/preparesh/streaming-input")

query = stream_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/home/preparesh/delta-checkpoint") \
    .start("/home/preparesh/delta-output")

Step 9: Drop files and verify

In the other terminal:

cat > ~/streaming-input/events1.csv << 'EOF'
click,1,10
view,2,5
click,3,20
EOF

Wait a few seconds for Spark to process. Then verify the Delta table by reading it as a batch:

batch = spark.read.format("delta").load("/home/preparesh/delta-output")
batch.show()

You should see the three rows. Drop another file:

cat > ~/streaming-input/events2.csv << 'EOF'
purchase,4,100
click,5,30
EOF

Wait a few seconds, then re-read:

spark.read.format("delta").load("/home/preparesh/delta-output").show()