Spark Structured Streaming to Delta Lake: Checkpoints, Exactly-Once, and the Transaction Log

Series: Building a Real-Time Data Platform from the Edge — Post 5 of 7. The full source is on GitHub: rpi-data-platform

Every Spark Structured Streaming tutorial tells you the framework guarantees exactly-once delivery. What the tutorials do not tell you is exactly-once guaranteed by what, enforced at which layer, and under what failure conditions it actually holds. Spark alone gives you at-least-once — it will re-process a batch after a crash. Delta Lake completes the guarantee, but only because its transaction log is append-only, content-addressed, and checked for duplicate writes at commit time. These two mechanisms together are what produce exactly-once. Neither one alone does it.

This post wires Spark Structured Streaming and Delta Lake into the stack. Then it opens both the checkpoint directory and the Delta transaction log and reads what is actually inside them, so that the claim is backed by evidence rather than documentation.

What we are adding

Two new services and one job file:

MinIO — S3-compatible object storage running locally on the Pi. Spark writes Delta Lake tables to MinIO via the S3A filesystem connector, exactly as it would write to S3 or ADLS in a cloud deployment. Swapping the endpoint is a one-line config change. The storage layer is intentionally decoupled from the processing layer.

Spark Structured Streaming — a single PySpark job running in local mode, reading from the sensor-events Kafka topic, flattening the nested payload into a typed schema, and writing to Delta Lake every 60 seconds. No standalone cluster, no worker nodes — local mode is the right choice for a single machine.

jobs/sensor_stream.py — the streaming job. Less than 100 lines. Most of the interesting content is in the SparkSession config, not the transformation logic, and we will go through it carefully.

jobs/optimize.py — a one-shot maintenance job that runs Delta Lake OPTIMIZE and VACUUM on a schedule. It is intentionally separate from the streaming job: mixing streaming writes and compaction in the same process complicates the memory budget without any benefit.

The repository structure grows by four files:

services/
├── minio/
│   └── docker-compose.yml   ← new
└── spark/
    ├── Dockerfile           ← new
    └── docker-compose.yml   ← new

jobs/
├── sensor_stream.py         ← new
└── optimize.py              ← new

MinIO: local S3

MinIO is a Go binary that speaks the S3 API. For Spark, it is indistinguishable from AWS S3 — same SDK, same Hadoop connector, same path format. The only configuration difference is the endpoint and one flag that most S3 tutorials do not mention.

# services/minio/docker-compose.yml

services:

  minio:
    image: minio/minio:latest
    container_name: minio
    restart: unless-stopped
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER:              ${MINIO_ROOT_USER:-minioadmin}
      MINIO_ROOT_PASSWORD:          ${MINIO_ROOT_PASSWORD:-minioadmin}
      # Exposes /metrics without auth so Prometheus can scrape it in Post 7
      MINIO_PROMETHEUS_AUTH_TYPE:   "public"
    ports:
      - "9000:9000"   # S3 API — what Spark and DuckDB connect to
      - "9001:9001"   # Web console — http://rpi-data-platform:9001
    volumes:
      - minio_data:/data
    networks:
      - data-platform
    healthcheck:
      # TCP check — works regardless of what utilities are bundled in the image
      test: ["CMD-SHELL", "bash -c 'cat < /dev/null > /dev/tcp/localhost/9000'"]
      interval: 10s
      timeout: 5s
      retries: 10
      start_period: 15s

  # ── Init: create the bucket once MinIO is healthy ─────────────────────────
  minio-init:
    image: minio/mc:latest
    container_name: minio-init
    restart: "no"
    depends_on:
      minio:
        condition: service_healthy
    environment:
      MINIO_ROOT_USER:     ${MINIO_ROOT_USER:-minioadmin}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:-minioadmin}
    networks:
      - data-platform
    entrypoint:
      - bash
      - -c
      - |
        mc alias set local http://minio:9000 ${MINIO_ROOT_USER} ${MINIO_ROOT_PASSWORD}
        mc mb --ignore-existing local/sensor-data
        echo "Bucket sensor-data ready."

volumes:
  minio_data:
    driver: local

The --ignore-existing flag on mc mb is the same idempotency principle as kafka-topic-init: the command always succeeds, and re-running docker compose up is always safe.

Add the MinIO credentials to .env (copy from .env.example):

MINIO_ROOT_USER=minioadmin
MINIO_ROOT_PASSWORD=minioadmin

Spark image on arm64

The Apache project publishes an official apache/spark image with native arm64 builds from 3.5.0 onwards. It bundles Java 17, Spark, PySpark, and all the environment wiring (SPARK_HOME, JAVA_HOME, PYSPARK_PYTHON) pre-configured. We use it directly.

The image weighs about 1.1GB. For a platform running unattended on a Pi that is already pulling Kafka, MinIO, and Grafana images, one extra gigabyte is not a meaningful constraint. The alternative — building from a bare JRE and installing PySpark via pip — saves disk space at the cost of a longer Dockerfile and one more layer to maintain. That trade is not worth it here.

delta-spark is the only addition. It is not bundled in the Apache image and must be installed on top.

# services/spark/Dockerfile

FROM apache/spark:3.5.1-python3
# apache/spark:3.5.1-python3 publishes native linux/arm64 builds.
# Includes Java 17, Spark 3.5.1, Python 3, and PySpark — pre-wired.
# Verify the arm64 manifest exists before deploying:
#   docker pull apache/spark:3.5.1-python3 --platform linux/arm64

USER root
# The image defaults to a non-root 'spark' user.
# Running as root keeps volume permissions consistent with the rest of the stack
# and avoids ownership conflicts on the ivy_cache mount.

# Pin pyspark to match the Spark JARs bundled in this image, then install
# delta-spark without letting pip pull in its own pyspark copy.
# Running python3 directly (not spark-submit) means pip must own the import;
# the image's bundled PySpark at /opt/spark/python/ is not on sys.path.
RUN pip3 install --no-cache-dir pyspark==3.5.1 && \
    pip3 install --no-cache-dir --no-deps delta-spark==3.2.0

COPY jobs/sensor_stream.py /opt/spark/work-dir/sensor_stream.py

CMD ["python3", "/opt/spark/work-dir/sensor_stream.py"]

The ivy cache pattern

Spark’s dependency resolution — used when you specify spark.jars.packages — is handled by Apache Ivy, which caches downloaded JARs in ~/.ivy2/cache. On first startup, Ivy downloads roughly 130MB of JARs from Maven Central:

delta-spark_2.12:3.2.0 — the Delta Lake core and Spark integration
spark-sql-kafka-0-10_2.12:3.5.1 — the Spark ↔ Kafka integration layer
hadoop-aws:3.3.4 — the S3A filesystem implementation (version must match Spark’s bundled Hadoop)
aws-java-sdk-bundle:1.12.367 — the AWS SDK that hadoop-aws delegates to

Delta comes from Ivy rather than from configure_spark_with_delta_pip (the helper bundled with the delta-spark pip package). The reason: configure_spark_with_delta_pip locates the Delta JAR by inspecting the pip package’s install path — a layout that works when Spark is launched via spark-submit or when the pip package owns the only PySpark installation. Inside the apache/spark image, PySpark ships at /opt/spark/python/ and is not installed via pip, so the helper cannot find the JAR and registers the Delta extensions before the classpath is ready. Letting Ivy resolve Delta alongside the other connectors avoids this entirely.

The version constraint on hadoop-aws matters. PySpark 3.5.1 bundles Hadoop 3.3.4. If you specify a different hadoop-aws version, you get classpath conflicts that produce cryptic NoSuchMethodError exceptions. Pin it to 3.3.4.

Rather than baking these JARs into the Docker image (which would make the image ~720MB heavier and couple image rebuilds to JAR updates), we mount a named volume at /root/.ivy2. On first docker compose up, Ivy downloads once and the volume persists. On every subsequent startup, the JARs are already there — cold start drops from roughly 2 minutes to under 15 seconds.

# services/spark/docker-compose.yml

services:

  spark-stream:
    build:
      # Context is repo root so COPY jobs/sensor_stream.py resolves correctly.
      # Relative paths in build.context are resolved from the location of this file.
      # apache/spark sets WORKDIR to /opt/spark/work-dir — the Dockerfile COPYs there.
      context: ../../
      dockerfile: services/spark/Dockerfile
    container_name: spark-stream
    restart: unless-stopped
    depends_on:
      kafka:
        condition: service_healthy
      minio-init:
        condition: service_completed_successfully
    environment:
      KAFKA_BOOTSTRAP:      kafka:9092
      MINIO_ENDPOINT:       http://minio:9000
      MINIO_ACCESS_KEY:     ${MINIO_ROOT_USER:-minioadmin}
      MINIO_SECRET_KEY:     ${MINIO_ROOT_PASSWORD:-minioadmin}
      # Caps driver memory. In local mode the driver IS the executor —
      # spark.executor.memory is ignored. SPARK_DRIVER_MEMORY is read by
      # the PySpark launcher before the JVM starts.
      SPARK_DRIVER_MEMORY:  "512m"
      PYSPARK_PYTHON:       python3
      PYTHONUNBUFFERED:     "1"

    networks:
      - data-platform

    volumes:
      # Connector JARs downloaded once, persisted across container restarts.
      # Running as root (per Dockerfile USER root), so ivy resolves to /root/.ivy2.
      - ivy_cache:/root/.ivy2
      # Spark shuffle and spill directory — separate from the container filesystem
      - spark_tmp:/tmp/spark-local

volumes:
  ivy_cache:
    driver: local
  spark_tmp:
    driver: local

On depends_on across included files: kafka is defined in services/kafka/docker-compose.yml and minio-init is defined in services/minio/docker-compose.yml. Docker Compose resolves these by service name within the project namespace. All services defined via include: share one network and one name registry — cross-file depends_on works without any extra configuration.

The streaming job

# jobs/sensor_stream.py

"""
Spark Structured Streaming: Kafka sensor-events → Delta Lake on MinIO.

Reads the sensor-events topic, parses the nested JSON payload into a flat
typed schema, and writes to Delta Lake every 60 seconds with device_type
and date partitioning. Checkpoints to MinIO so offset state survives
container restarts and image rebuilds.
"""

import os
import pyspark
from pyspark.sql import functions as F
from pyspark.sql.streaming import StreamingQueryListener
from pyspark.sql.types import (
    StructType, StructField,
    StringType,
    DoubleType, IntegerType, BooleanType,
)

# ── Configuration ────────────────────────────────────────────────────────────

KAFKA_BOOTSTRAP  = os.getenv("KAFKA_BOOTSTRAP",  "kafka:9092")
KAFKA_TOPIC      = "sensor-events"
KAFKA_GROUP_ID   = "spark-sensor-consumer"

MINIO_ENDPOINT   = os.getenv("MINIO_ENDPOINT",   "http://minio:9000")
MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY",  "minioadmin")
MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY",  "minioadmin")

DELTA_TABLE_PATH = "s3a://sensor-data/delta/sensor_events"
CHECKPOINT_PATH  = "s3a://sensor-data/checkpoints/sensor_stream"
# Checkpoint goes to MinIO — not a Docker volume.
# If checkpoint lived in a local volume, an image rebuild would delete it
# and the next startup would re-read the entire Kafka topic from earliest.
# Storing checkpoint in MinIO means it survives any container lifecycle event.

TRIGGER_SECONDS  = "60 seconds"
# 60s trigger produces at most 2 Parquet files per batch (one per device_type).
# Lower triggers produce more small files. OPTIMIZE (covered below) fixes that,
# but starting with a reasonable trigger reduces the maintenance surface.

WATERMARK_DELAY  = "5 minutes"
# How long to hold state waiting for late events before advancing the watermark.
# Door sensor messages arrive sporadically — 5 minutes is generous without
# holding unbounded state in the streaming query.

# All JARs resolved via Ivy at startup and cached in the ivy_cache volume.
# hadoop-aws must match Spark's bundled Hadoop (3.3.4 for PySpark 3.5.1).
# Delta is included here rather than via configure_spark_with_delta_pip because
# the pip helper can't locate its JAR inside the apache/spark image layout,
# causing the Delta extensions to be registered before the JAR is on the
# classpath. Ivy resolves all packages before extensions are instantiated.
PACKAGES = ",".join([
    "io.delta:delta-spark_2.12:3.2.0",
    "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1",
    "org.apache.hadoop:hadoop-aws:3.3.4",
    "com.amazonaws:aws-java-sdk-bundle:1.12.367",
])

# ── SparkSession ─────────────────────────────────────────────────────────────

builder = (
    pyspark.sql.SparkSession.builder
    .appName("sensor-stream")
    .master("local[2]")
    # local[2]: 2 threads. The sensor-events topic has 3 partitions, but only
    # 2 contain data (door-sensor-01 hashes to one partition, smart-plug-01 to
    # another, the third is empty). A third thread would idle. Two threads match
    # the actual parallelism available, leaving the Pi's remaining cores for
    # Kafka, MinIO, and the OS.

    # S3A filesystem — points Spark at MinIO instead of AWS S3
    .config("spark.hadoop.fs.s3a.endpoint",               MINIO_ENDPOINT)
    .config("spark.hadoop.fs.s3a.access.key",             MINIO_ACCESS_KEY)
    .config("spark.hadoop.fs.s3a.secret.key",             MINIO_SECRET_KEY)
    .config("spark.hadoop.fs.s3a.path.style.access",      "true")
    # path.style.access is required for MinIO. Without it, S3A constructs
    # virtual-hosted-style URLs: sensor-data.minio:9000/delta/...
    # That hostname does not resolve inside Docker. Path-style addressing
    # uses minio:9000/sensor-data/delta/... which does.
    .config("spark.hadoop.fs.s3a.impl",                   "org.apache.hadoop.fs.s3a.S3AFileSystem")
    .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
    # MinIO runs HTTP locally — SSL handshake would fail and produce a
    # misleading "connection reset" error rather than an explicit SSL error.

    # Shuffle partition reduction
    .config("spark.sql.shuffle.partitions", "2")
    .config("spark.default.parallelism",    "2")
    # Default is 200. A streaming append job with no groupBy or join
    # does not shuffle, so this setting is irrelevant for our current query.
    # It is set anyway because future additions (aggregations, deduplication)
    # would otherwise produce 200 empty output files per batch.

    .config("spark.local.dir", "/tmp/spark-local")

    # Pin driver to localhost so Spark never tries to resolve the container
    # hostname — unresolvable hostnames cause a NullPointerException in
    # BlockManagerMasterEndpoint.register when running in local mode on Docker.
    .config("spark.driver.host",        "localhost")
    .config("spark.driver.bindAddress", "127.0.0.1")

    .config("spark.jars.packages",              PACKAGES)
    # Use in-memory catalog — prevents Spark from defaulting to Hive metastore,
    # which causes errors in the apache/spark image when no Hive setup exists.
    .config("spark.sql.catalogImplementation", "in-memory")
    .config("spark.sql.extensions",             "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog",  "org.apache.spark.sql.delta.catalog.DeltaCatalog")
)

print("[sensor_stream] Building SparkSession (Ivy JAR resolution may take ~2 min on first run)...")
spark = builder.getOrCreate()
print(f"[sensor_stream] SparkSession ready — Spark {spark.version}")
spark.sparkContext.setLogLevel("WARN")


class BatchLogger(StreamingQueryListener):
    def onQueryStarted(self, event):
        print(f"[sensor_stream] Query started  id={event.id}")

    def onQueryProgress(self, event):
        p = event.progress
        rows = p.numInputRows
        batch = p.batchId
        duration_ms = p.durationMs.get("triggerExecution", 0)
        print(f"[sensor_stream] Batch {batch} complete — {rows} rows — {duration_ms}ms")

    def onQueryTerminated(self, event):
        print(f"[sensor_stream] Query terminated id={event.id} exception={event.exception}")


spark.streams.addListener(BatchLogger())

# ── Input schema ─────────────────────────────────────────────────────────────
# The producer (Post 4) wraps each Zigbee message as:
# {
#   "device_id":   "smart-plug-01",
#   "ingested_at": "2026-06-22T09:15:30.123456+00:00",
#   "payload":     "{\"power\": 84.3, \"voltage\": 230.1, ...}"
# }
# payload arrives as a JSON string — the producer serialised it that way.

EVENT_SCHEMA = StructType([
    StructField("device_id",   StringType(), False),
    StructField("device_type", StringType(), False),
    StructField("ingested_at", StringType(), False),
    StructField("payload",     StringType(), True),
])

# ── Read from Kafka ──────────────────────────────────────────────────────────

raw = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers",  KAFKA_BOOTSTRAP)
    .option("subscribe",                KAFKA_TOPIC)
    .option("kafka.group.id",           KAFKA_GROUP_ID)
    .option("startingOffsets",          "earliest")
    # earliest applies only on first startup (no checkpoint exists yet).
    # Once the checkpoint directory exists in MinIO, Spark ignores this
    # option and resumes from the checkpointed offset. Setting earliest
    # ensures the full history is processed on initial deployment.
    .option("maxOffsetsPerTrigger",     1000)
    # Safety ceiling. At sensor throughput (~1 message/10 seconds from the
    # plug, sporadic from the door sensor), we accumulate ~360 messages per
    # hour. 1000 per trigger is far above that — this guard only activates
    # if Spark is restarted after a multi-day outage with a large backlog.
    .load()
)

# ── Parse and flatten ────────────────────────────────────────────────────────
# Schema strategy: extract all known typed columns from both sensor schemas.
# Door sensor fields are null in smart plug rows, and vice versa.
# raw_payload preserves the original JSON for forward compatibility —
# adding a new sensor type does not require a schema migration.

parsed = (
    raw
    .select(
        F.from_json(F.col("value").cast("string"), EVENT_SCHEMA).alias("e")
    )
    .select(
        F.col("e.device_id"),
        F.col("e.device_type"),
        F.to_timestamp("e.ingested_at").alias("event_time"),

        # Shared across both devices
        F.get_json_object("e.payload", "$.linkquality").cast(IntegerType()).alias("linkquality"),

        # SNZB-04P (door/window) fields — null for smart_plug rows
        F.get_json_object("e.payload", "$.contact").cast(BooleanType()).alias("contact"),
        F.get_json_object("e.payload", "$.battery").cast(IntegerType()).alias("battery"),

        # S60ZBTPF (smart plug) fields — null for door_window rows
        F.get_json_object("e.payload", "$.power").cast(DoubleType()).alias("power"),
        F.get_json_object("e.payload", "$.voltage").cast(DoubleType()).alias("voltage"),
        F.get_json_object("e.payload", "$.current").cast(DoubleType()).alias("current"),
        F.get_json_object("e.payload", "$.energy").cast(DoubleType()).alias("energy"),
        F.get_json_object("e.payload", "$.state").alias("plug_state"),

        # Full payload JSON preserved
        F.col("e.payload").alias("raw_payload"),
        F.current_timestamp().alias("processed_at"),
    )
    .withWatermark("event_time", WATERMARK_DELAY)
    .withColumn("event_date", F.to_date("event_time"))
    # event_date is a partition column — derived here so it is available
    # to partitionBy() without storing a redundant column in each row
)

# ── Write to Delta Lake ──────────────────────────────────────────────────────

query = (
    parsed.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", CHECKPOINT_PATH)
    .partitionBy("device_type", "event_date")
    .trigger(processingTime=TRIGGER_SECONDS)
    .start(DELTA_TABLE_PATH)
)

print(f"[sensor_stream] Streaming query started — id={query.id} runId={query.runId}")
try:
    query.awaitTermination()
except Exception as exc:
    print(f"[sensor_stream] Fatal: {exc}")
    raise

Connecting the services

Add the two new service files to the root docker-compose.yml:

# docker-compose.yml

include:
  - services/kafka/docker-compose.yml
  - services/mqtt/docker-compose.yml
  - services/producer/docker-compose.yml
  - services/minio/docker-compose.yml   ← new
  - services/spark/docker-compose.yml   ← new

networks:
  data-platform:
    driver: bridge

The dependency chain the runtime now enforces:

kafka-init → kafka (healthy) → kafka-topic-init
                          ↓
minio → minio-init (completed)
                          ↓
                    spark-stream

spark-stream does not start until both Kafka is healthy and the MinIO bucket exists. Without the bucket, the first write to s3a://sensor-data/... would fail with a NoSuchBucketException that looks like an S3 authentication error and wastes twenty minutes of debugging.

Starting the stack

ssh tomvapon@192.168.1.50
cd rpi-data-platform

# Build the Spark image and start everything
docker compose up -d --build

# Watch the dependency chain resolve in real time
docker compose logs -f minio-init spark-stream

A clean first-run sequence:

minio-init   | Bucket `local/sensor-data` created successfully.
minio-init   | Bucket sensor-data ready.
spark-stream | [IvyResolver] :: loading settings :: url = jar:...
spark-stream | Downloading: org/apache/spark/spark-sql-kafka-0-10_2.12/3.5.1/...
spark-stream | Downloading: org/apache/hadoop/hadoop-aws/3.3.4/...
spark-stream | Downloading: com/amazonaws/aws-java-sdk-bundle/1.12.367/...
spark-stream | :: resolution report :: resolve 12341ms
spark-stream | Streaming query [id = <uuid>, runId = <uuid>] started
spark-stream | Batch 0 starting. Offsets: {"sensor-events":{"0":0,"1":0,"2":0}}
spark-stream | Batch 0 complete. Committed offsets: {"sensor-events":{"0":15,"1":12,"2":0}}

On subsequent startups (JARs already cached):

spark-stream | Resuming from checkpoint at s3a://sensor-data/checkpoints/sensor_stream
spark-stream | Batch 3 starting. Offsets: {"sensor-events":{"0":45,"1":36,"2":0}}

The first run takes about 2 minutes for JAR resolution. Every restart after that is under 15 seconds.

What is actually inside the Delta transaction log

The Delta transaction log lives at s3a://sensor-data/delta/sensor_events/_delta_log/. Every commit that Spark makes to the Delta table produces one JSON file in this directory. The file name is the commit number, zero-padded to 20 digits.

Inspect it directly from the Pi:

# From a second terminal — while spark-stream is running
docker exec minio-init mc alias set local http://minio:9000 minioadmin minioadmin

# List the transaction log after the first batch completes
docker exec minio-init mc ls local/sensor-data/delta/sensor_events/_delta_log/
# 2026-06-22 09:16:01    2.1 KiB 00000000000000000000.json

docker exec minio-init mc cat \
  local/sensor-data/delta/sensor_events/_delta_log/00000000000000000000.json

The file contains four newline-delimited JSON objects. One per line:

{"commitInfo":{"timestamp":1750583761000,"operation":"STREAMING UPDATE","operationParameters":{"outputMode":"Append","queryId":"a8f3c2d1-...","epochId":"0"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numRemovedFiles":"0","numAddedFiles":"2","numOutputRows":"27"},"clientVersion":"delta-spark/3.2.0"}}
{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
{"metaData":{"id":"7e2b4a91-...","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"device_id\",\"type\":\"string\",\"nullable\":false},{\"name\":\"device_type\",\"type\":\"string\",\"nullable\":false},{\"name\":\"event_time\",\"type\":\"timestamp\",\"nullable\":true},{\"name\":\"linkquality\",\"type\":\"integer\",\"nullable\":true},{\"name\":\"contact\",\"type\":\"boolean\",\"nullable\":true},{\"name\":\"battery\",\"type\":\"integer\",\"nullable\":true},{\"name\":\"power\",\"type\":\"double\",\"nullable\":true},{\"name\":\"voltage\",\"type\":\"double\",\"nullable\":true},{\"name\":\"current\",\"type\":\"double\",\"nullable\":true},{\"name\":\"energy\",\"type\":\"double\",\"nullable\":true},{\"name\":\"plug_state\",\"type\":\"string\",\"nullable\":true},{\"name\":\"raw_payload\",\"type\":\"string\",\"nullable\":true},{\"name\":\"processed_at\",\"type\":\"timestamp\",\"nullable\":true}]}","partitionColumns":["device_type","event_date"],"configuration":{},"createdTime":1750583760000}}
{"add":{"path":"device_type=smart_plug/event_date=2026-06-22/part-00000-4c9a2e7f-...snappy.parquet","partitionValues":{"device_type":"smart_plug","event_date":"2026-06-22"},"size":5234,"modificationTime":1750583761000,"dataChange":true,"stats":"{\"numRecords\":24,\"minValues\":{\"event_time\":\"2026-06-22T08:16:00.000Z\",\"power\":0.1,\"voltage\":229.4,\"linkquality\":231},\"maxValues\":{\"event_time\":\"2026-06-22T09:15:50.000Z\",\"power\":128.6,\"voltage\":230.8,\"linkquality\":255},\"nullCount\":{\"contact\":24,\"battery\":24,\"plug_state\":0,\"power\":0,\"voltage\":0}}"}}
{"add":{"path":"device_type=door_window/event_date=2026-06-22/part-00001-4c9a2e7f-...snappy.parquet","partitionValues":{"device_type":"door_window","event_date":"2026-06-22"},"size":1802,"modificationTime":1750583761000,"dataChange":true,"stats":"{\"numRecords\":3,\"minValues\":{\"event_time\":\"2026-06-22T08:21:10.000Z\",\"linkquality\":159,\"battery\":97},\"maxValues\":{\"event_time\":\"2026-06-22T09:02:44.000Z\",\"linkquality\":186,\"battery\":97},\"nullCount\":{\"contact\":0,\"battery\":0,\"power\":3,\"voltage\":3,\"current\":3,\"energy\":3}}"}}

Each line is independent and serves a specific purpose:

commitInfo — the audit record. It contains the operation type (STREAMING UPDATE), the query ID, and crucially the epochId. The epoch ID is Spark’s batch number. Delta uses (queryId, epochId) as a uniqueness key for idempotent writes — we will return to this when we discuss exactly-once semantics.

protocol — declares the minimum Delta reader and writer versions required to open this table. minWriterVersion: 2 enables append-only writes. Any tool that reads the table must support at least these protocol versions.

metaData — the full schema definition and partition column list, serialised as JSON-within-JSON inside schemaString. This is why Delta can detect schema mismatches before writing a single row: it compares the incoming DataFrame schema against the schema recorded here. If they differ and mergeSchema is not enabled, the write fails before touching any data files.

add — one record per Parquet file written. The stats field is what makes Delta’s data skipping work: per-file min/max values and null counts for every column. When a query includes a filter like WHERE power > 100, Delta reads the stats from the transaction log and skips any file whose maxValues.power is below 100. No data file is opened. The scanning happens at log level, which fits entirely in memory.

What is actually inside the checkpoint directory

The checkpoint lives at s3a://sensor-data/checkpoints/sensor_stream/. Its structure:

docker exec minio-init mc ls --recursive \
  local/sensor-data/checkpoints/sensor_stream/
# metadata
# offsets/0
# offsets/1
# commits/0
# commits/1

There are three components: metadata, offsets/, and commits/.

metadata — created once when the query starts. Contains the query ID and Spark version:

{"id":"a8f3c2d1-...","runId":"b9e5f3a2-...","sparkVersion":"3.5.1","..."}

offsets/N — written before batch N executes. Contains the Kafka offsets that will be consumed in this batch:

v1
{"batchWatermarkMs":0,"batchTimestampMs":1750583700000,"conf":{"spark.sql.shuffle.partitions":"2",...}}
{"sensor-events":{"0":0,"1":0,"2":0}}

The last line is the key: a JSON object mapping topic name → partition → start offset. Batch 0 will read from offset 0 on each partition up to the offsets observed at query start.

commits/N — written after batch N successfully completes. Contains only a version marker:

v1

The relationship between these two files is the recovery protocol. When Spark restarts, it scans both directories:

If offsets/N exists but commits/N does not: batch N was started but did not complete. Spark re-runs it.
If both offsets/N and commits/N exist: batch N is done. Spark moves to N+1.
If neither exists for N: Spark starts fresh from the highest complete batch.

Exactly-once: the full picture

Spark’s checkpoint gives at-least-once from Kafka’s perspective: if a batch fails after writing some data but before writing commits/N, Spark will re-read those Kafka offsets and re-process those records.

Delta Lake turns at-least-once into exactly-once by rejecting duplicate writes. When Spark re-runs batch N, it passes the same (queryId, epochId=N) pair to Delta’s write path. Delta’s transaction log writer checks whether a commit with that pair already exists in the _delta_log/. If it does — meaning the previous run succeeded in writing the files but crashed before Spark could write commits/N — Delta returns success without writing any files. The re-run is a no-op from the sink’s perspective.

This is why isBlindAppend: true in the commitInfo matters. It declares that the write does not need to check for conflicting concurrent writes — it is an unconditional append. Delta can process these at high throughput because no read-modify-write cycle is needed. The idempotency check is the only coordination required.

The two offset stores — Spark’s checkpoint in MinIO and Kafka’s __consumer_offsets topic — are intentionally independent. Spark’s checkpoint is the authoritative record of what has been processed. Kafka’s consumer group offset is a secondary record used by monitoring tools (kafka-consumer-groups.sh, kafka_exporter) to show consumer lag. If the two disagree after a crash — which can happen if Spark committed to Delta but crashed before committing to __consumer_offsets — Spark’s checkpoint wins. The Kafka consumer group offset is eventually re-aligned on the next successful batch.

Verifying end-to-end

After the first batch completes, verify the table is readable:

# Check the Parquet files landed in MinIO
docker exec minio-init mc ls --recursive local/sensor-data/delta/sensor_events/
# device_type=smart_plug/event_date=2026-06-22/part-00000-xxx.parquet
# device_type=door_window/event_date=2026-06-22/part-00001-xxx.parquet
# _delta_log/00000000000000000000.json

# Count records using the Spark container itself
docker exec spark-stream python3 - <<'EOF'
import os
import pyspark

PACKAGES = ",".join([
    "io.delta:delta-spark_2.12:3.2.0",
    "org.apache.hadoop:hadoop-aws:3.3.4",
    "com.amazonaws:aws-java-sdk-bundle:1.12.367",
])

spark = (
    pyspark.sql.SparkSession.builder
    .appName("read-check")
    .master("local[1]")
    .config("spark.hadoop.fs.s3a.endpoint",               os.getenv("MINIO_ENDPOINT"))
    .config("spark.hadoop.fs.s3a.access.key",             os.getenv("MINIO_ACCESS_KEY"))
    .config("spark.hadoop.fs.s3a.secret.key",             os.getenv("MINIO_SECRET_KEY"))
    .config("spark.hadoop.fs.s3a.path.style.access",      "true")
    .config("spark.hadoop.fs.s3a.impl",                   "org.apache.hadoop.fs.s3a.S3AFileSystem")
    .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
    .config("spark.driver.host",        "localhost")
    .config("spark.driver.bindAddress", "127.0.0.1")
    .config("spark.jars.packages",              PACKAGES)
    .config("spark.sql.catalogImplementation", "in-memory")
    .config("spark.sql.extensions",             "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog",  "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")

df = spark.read.format("delta").load("s3a://sensor-data/delta/sensor_events")
df.groupBy("device_type").count().show()
df.select("device_type","event_time","power","contact","linkquality").show(5)
EOF

Expected output:

+-----------+-----+
|device_type|count|
+-----------+-----+
| smart_plug|   24|
|door_window|    3|
+-----------+-----+

+-----------+------------------------+------+-------+------------+
|device_type|event_time              |power |contact|linkquality |
+-----------+------------------------+------+-------+------------+
|smart_plug |2026-06-22 08:16:00.000 |84.3  |null   |248         |
|smart_plug |2026-06-22 08:16:10.123 |83.9  |null   |251         |
|door_window|2026-06-22 08:21:10.441 |null  |false  |159         |
|smart_plug |2026-06-22 08:16:20.891 |84.1  |null   |247         |
|smart_plug |2026-06-22 08:16:30.002 |85.0  |null   |249         |
+-----------+------------------------+------+-------+------------+

The null values are correct: contact does not exist in smart plug payloads, power does not exist in door sensor payloads. get_json_object returns null for missing keys rather than throwing.

The small file problem

Each 60-second trigger that writes new data produces two Parquet files — one per active device_type partition. After 24 hours of continuous operation that is 288 small files. After a week, over 2000. Small files degrade read performance and waste metadata overhead.

Delta Lake solves this with OPTIMIZE, which rewrites a partition’s small files into a smaller number of target-size files. OPTIMIZE is additive: it writes new compacted files and adds remove entries to the transaction log for the old ones. The old files remain on disk until VACUUM runs.

VACUUM is the only destructive operation in Delta Lake. Deleted files cannot be recovered. The default 7-day safety floor means Delta refuses to vacuum files younger than that — overriding it requires spark.databricks.delta.retentionDurationCheck.enabled: false. Post 6 covers time-travel in the context of DuckDB queries.

jobs/optimize.py handles both operations as a self-contained one-shot job, designed to run as a separate container on a daily schedule:

# jobs/optimize.py

"""
Delta Lake maintenance: OPTIMIZE + VACUUM for sensor_events table.

Run as a one-shot container (restart: no) on a daily schedule.
OPTIMIZE compacts the small files produced by the 60-second streaming
trigger. VACUUM deletes the old pre-compaction files after the retention
window, keeping disk usage flat over time.
"""

import os
import sys
import pyspark
from delta import configure_spark_with_delta_pip
from delta.tables import DeltaTable

MINIO_ENDPOINT   = os.getenv("MINIO_ENDPOINT",   "http://minio:9000")
MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY",  "minioadmin")
MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY",  "minioadmin")

DELTA_TABLE_PATH  = "s3a://sensor-data/delta/sensor_events"
VACUUM_RETAIN_DAYS = int(os.getenv("VACUUM_RETAIN_DAYS", "30"))

PACKAGES = ",".join([
    "io.delta:delta-spark_2.12:3.2.0",
    "org.apache.hadoop:hadoop-aws:3.3.4",
    "com.amazonaws:aws-java-sdk-bundle:1.12.367",
])

print("[optimize] Building SparkSession...")
spark = (
    pyspark.sql.SparkSession.builder
    .appName("delta-optimize")
    .master("local[1]")
    .config("spark.hadoop.fs.s3a.endpoint",               MINIO_ENDPOINT)
    .config("spark.hadoop.fs.s3a.access.key",             MINIO_ACCESS_KEY)
    .config("spark.hadoop.fs.s3a.secret.key",             MINIO_SECRET_KEY)
    .config("spark.hadoop.fs.s3a.path.style.access",      "true")
    .config("spark.hadoop.fs.s3a.impl",                   "org.apache.hadoop.fs.s3a.S3AFileSystem")
    .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
    .config("spark.driver.host",                          "localhost")
    .config("spark.driver.bindAddress",                   "127.0.0.1")
    .config("spark.sql.shuffle.partitions",               "2")
    .config("spark.jars.packages",                        PACKAGES)
    .config("spark.sql.catalogImplementation",            "in-memory")
    .config("spark.sql.extensions",             "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog",  "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    # Allow VACUUM to delete files newer than the default 7-day safety floor
    .config("spark.databricks.delta.retentionDurationCheck.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")
print(f"[optimize] SparkSession ready — Spark {spark.version}")

dt = DeltaTable.forPath(spark, DELTA_TABLE_PATH)

print("[optimize] Running OPTIMIZE (compaction)...")
dt.optimize().executeCompaction()
print("[optimize] OPTIMIZE complete")

retain_hours = VACUUM_RETAIN_DAYS * 24
print(f"[optimize] Running VACUUM (retaining {VACUUM_RETAIN_DAYS} days = {retain_hours}h)...")
dt.vacuum(retentionHours=retain_hours)
print(f"[optimize] VACUUM complete — files older than {VACUUM_RETAIN_DAYS} days deleted")

spark.stop()
print("[optimize] Done")
sys.exit(0)

A few things worth noting:

VACUUM_RETAIN_DAYS defaults to 30 days and is configurable via environment variable. 30 days means you can time-travel back 30 versions; adjust down if disk space is a constraint.

retentionDurationCheck.enabled: false bypasses Delta’s built-in 7-day safety floor. Without it, any VACUUM call with retentionHours < 168 raises an error. This flag is required here because we want VACUUM_RETAIN_DAYS to be freely configurable — including values shorter than 7 days for low-disk scenarios.

local[1] — one thread. OPTIMIZE reads existing Parquet files and rewrites them; it does not read from Kafka. One thread is enough, and it leaves more cores for the streaming job running in parallel.

spark-sql-kafka is absent from PACKAGES. The optimize job never touches Kafka, so there is no reason to download 60MB of Kafka connectors on its startup path.

To trigger it manually:

docker exec spark-stream python3 /opt/spark/work-dir/optimize.py

Or run it as a one-shot container on a daily cron schedule — separate from spark-stream so it does not share the 512MB memory cap.

Memory at full stack

With Spark now running, the Pi’s 4GB is allocated as follows:

Service	Heap / RAM	Notes
Kafka	512MB JVM heap	`KAFKA_HEAP_OPTS`
Spark	512MB JVM heap	`JAVA_TOOL_OPTIONS -Xmx512m`
MinIO	~256MB	Object storage server
Mosquitto	~10MB	Negligible
Z2M + producer	~100MB	Python processes
OS + Docker	~400MB
Remaining	~1.8GB	Comfortable headroom

The JVM overhead beyond the heap (metaspace, code cache, direct buffers) typically adds 150–300MB per JVM process. Total actual consumption sits around 2.2–2.4GB at steady state, well within the Pi’s 4GB plus the 2GB swap configured in Post 2.

What comes next

The Delta table is live and accumulating sensor data. Post 6 connects DuckDB to it — reading Delta files directly from MinIO with no Spark involved, no extra infrastructure, and far less memory pressure than a Spark read path. That is the point of keeping the serving layer separate from the ingestion layer: for analytics on sensor-scale data, Spark is the wrong tool. DuckDB handles it in a shell.

Previous: Post 4 — Kafka on arm64: KRaft Mode, Topic Design, and the Python Producer Bridge Next: Post 6 — DuckDB + Grafana: Serving Delta Lake Data Without Spark