All posts

Streaming into Iceberg: availableNow, checkpoints, and the snapshot count nobody watches

A streaming write into Iceberg commits a snapshot per micro-batch, so the trigger you choose decides how fast your metadata grows. Here is what availableNow actually does, why re-running it changes nothing, and the accounting that makes a streaming table slow down over weeks.

26 min read Iceberg

TL;DR

  • A streaming write commits one Iceberg snapshot per micro-batch, not per file and not per row. Three input files consumed in one availableNow run produced 1 snapshot; a fourth file arriving later produced a second.
  • Trigger.AvailableNow drains everything currently available, then stops. It is the batch-shaped way to run a streaming job, and it is what you want in an orchestrator.
  • Re-running the same query over the same input added 0 rows. The checkpoint records what was read, and the Iceberg sink records each committed batch in the snapshot summary, so a replayed batch is skipped. A new checkpoint is a new query, and it re-ingested everything.
  • The trigger interval is therefore a metadata-growth decision. A one-minute trigger is 1,440 snapshots a day, each with its own manifest list.
  • The file source does not look inside subdirectories unless you ask it to. Files staged one directory deep were silently never read, and the query reported success.

What a streaming write actually commits

Spark’s Structured Streaming writes to Iceberg through a normal sink, and the unit of work is the micro-batch. Each batch that produces data ends in an Iceberg commit, which means a new snapshot, a new manifest list, and a new metadata file.

That single fact explains most of what goes right and wrong with streaming Iceberg tables. It is not “streaming” in the sense of rows trickling into a file. It is a sequence of small, atomic batch commits, and everything about the table’s long-term health follows from how many of them you make.

flowchart LR
    F[new files arrive] --> MB[micro-batch]
    MB --> W[write data files]
    W --> C["commit: snapshot + manifest list + metadata.json"]
    C --> CK[checkpoint records progress]
    CK --> MB

The run, measured

Three Parquet files of 1,000 order events each, staged flat in a landing directory, a table created ahead of time, and a query with availableNow. This is the core of streaming_file_ingest.py, which is complete in run it yourself:

SCHEMA = "order_id BIGINT, customer_id BIGINT, status STRING, amount DOUBLE, event_ts TIMESTAMP"

query = (spark.readStream.schema(SCHEMA).parquet("s3a://warehouse/blog_streaming.db/_stream/landing")
              .writeStream.format("iceberg").outputMode("append")
              .trigger(availableNow=True)
              .option("checkpointLocation", "s3a://warehouse/blog_streaming.db/_stream/ckpt")
              .toTable("ice.blog_streaming.order_events"))
query.awaitTermination()

files staged                    : 3
micro-batches in the run        : 1
rows after first run            : 3000
snapshots                       : 1
data files in table             : 2

Three files, 3,000 rows, one snapshot. availableNow did not create a snapshot per file; it consumed everything available in one micro-batch and committed once. The batch wrote two data files, one per task of the local[2] run, not one per input file. That is the behaviour you want for a scheduled job, and the opposite of what a short processing-time trigger gives you.

Running the identical query a second time, with no new input:

rows after re-run               : 3000
snapshots                       : 1

Nothing was added and nothing was duplicated. Then a fourth file appeared:

rows after 4th file             : 4000
snapshots                       : 2
data files in table             : 3

A second snapshot, for a second run that found work. The snapshot count tracks runs that found data, not files and not rows.

Why re-running is safe

Two records of progress are doing that work. The first is the checkpoint:

checkpoint entries              : ['commits', 'metadata', 'offsets', 'sources']
offsets log                     : ['0', '1']
commits log                     : ['0', '1']

offsets records what each batch was told to read, before it ran. commits records which batches finished. sources tracks which files the file source has already seen, and metadata holds the query’s id. On restart, the engine compares the two logs, re-runs anything that started and did not finish, and skips everything already committed.

The second is in the table. The Iceberg sink writes the query id and the batch (epoch) id into every snapshot’s summary:

+-------------------+---------+-------------+--------+-----+
|snapshot_id        |operation|same_query_id|epoch_id|added|
+-------------------+---------+-------------+--------+-----+
|5434260312021626810|append   |true         |0       |3000 |
|7286108917502102620|append   |true         |1       |1000 |
+-------------------+---------+-------------+--------+-----+

Before it commits, the sink looks up the last epoch committed for this query id and skips any epoch at or below it (BaseStreamingWrite in Iceberg’s Spark module). That closes the gap the checkpoint alone leaves: a batch that committed to Iceberg but crashed before Spark wrote its commits entry is re-run on restart, and the sink recognises it and does not append it twice.

Both records are keyed to the query, not to the data. Point the same query at a new checkpoint directory and it gets a new query id, the sink finds no earlier epochs for it, and every file is ingested again:

rows after new-checkpoint run   : 8000
snapshots                       : 3

Two consequences follow. A checkpoint belongs to one query against one table: reusing it across two different sinks corrupts both notions of progress. And deleting a checkpoint to “reset” a job is a data-duplication event, not a clean slate, unless you also truncate the target.

The full anatomy of that directory, including why the offset log pins configuration, is in Structured Streaming internals.

Choosing the trigger

This is the decision that determines whether the table stays healthy.

Trigger Behaviour Metadata cost
availableNow drain everything available, then stop one snapshot per run, or per maxFilesPerTrigger slice
processingTime='5 minutes' a batch every 5 minutes, forever up to 288 snapshots/day
processingTime='1 minute' a batch every minute up to 1,440 snapshots/day
default (no trigger) batches as fast as they complete unbounded

A processing-time trigger commits only when a batch finds data, so “up to” is the ceiling for a source that always has something new.

availableNow is the right default for most pipelines. It turns a streaming query into something an orchestrator can schedule: it starts, it catches up, it exits with a status. You get the incremental-processing semantics of streaming with the operational shape of a batch job, and you commit once per run rather than once per micro-batch.

Reach for a continuous processing-time trigger when latency genuinely matters, and then accept that you have signed up for the maintenance in the next section.

The cost that shows up in week three

Every commit writes a metadata file, a manifest list, and at least one manifest. At one snapshot a minute, a table accumulates 1,440 of each per day. Nothing fails. Queries just get slower, because planning a query means reading the current metadata file, its manifest list, and the manifests it points at.

The symptom is distinctive: query planning time grows while the data volume is flat, and the slowdown affects every query against the table equally, including trivial ones.

Two jobs keep it in check, and a streaming table needs both scheduled from the day it is created:

Compaction merges the many small files each micro-batch produced into fewer large ones. Without it, a table ingesting every minute accumulates a file per partition per task per minute.

Snapshot expiry removes old snapshots and the files only they reference. This is the one that controls storage, and compaction without expiry grows the bill while shrinking the file count. The mechanics of both are in table maintenance for Iceberg and Hudi.

There is a third knob worth knowing: write.metadata.previous-versions-max and write.metadata.delete-after-commit.enabled bound how many old metadata JSON files are kept, which is separate from snapshot expiry and is the specific fix for a metadata directory with tens of thousands of files in it. The maintenance section measures both.

Configuring the stream

Three options change the shape of what a streaming write produces.

query = (stream
         .writeStream
         .format("iceberg")
         .outputMode("append")
         .trigger(availableNow=True)
         .option("checkpointLocation", "s3a://warehouse/blog_streaming.db/_stream2/ckpt_true")
         .option("fanout-enabled", "true")
         .toTable("ice.blog_streaming.events_true"))

fanout-enabled decides how a task writes rows for many partitions. With fanout, the writer keeps one open file per partition, so input can arrive in any order. Without it, the writer closes a partition’s file when the next partition starts, so rows must be grouped by partition. In Iceberg 1.11 fanout is the default for a table with no sort order: SparkWriteUtil.writeOrdering asks Spark for no ordering, and the writer then fans out. Setting fanout-enabled=false does not make unsorted input fail, either. Iceberg asks Spark to order rows by partition, and Spark adds a sort to every micro-batch. The same unsorted batch, spread over three days, into a table partitioned by days(event_ts):

default                rows=3000 partitions=3 data files=3 sort in plan=False
fanout-enabled=true    rows=3000 partitions=3 data files=3 sort in plan=False
fanout-enabled=false   rows=3000 partitions=3 data files=3 sort in plan=True

So the choice is not correctness but cost: fanout keeps a file open per partition in every task, which costs memory on a batch that touches many partitions; turning it off costs a sort per batch. For a stream landing in a handful of partitions per batch, the default is right.

maxFilesPerTrigger bounds how much one batch consumes, which is what stops a backlog from being processed as one enormous batch. It is an option of the file source, so it goes on readStream:

stream = (spark.readStream.schema(SCHEMA)
          .option("maxFilesPerTrigger", 2)
          .parquet(landing))

With availableNow, four staged files became two micro-batches and two commits:

micro-batches in the run        : 2
rows per batch                  : [2000, 2000]
rows                            : 4000
snapshots                       : 2

Put the same option on writeStream and the sink ignores it without a word: a run that should have produced four batches produced one.

The trigger determines the commit rate, which is the metadata growth rate. Worth restating as arithmetic: a one-minute trigger is 1,440 commits a day, each writing three metadata files, so 4,320 metadata files a day on a table that may only receive a few thousand rows.

Watching a running stream

StreamingQueryProgress answers whether the stream is keeping up, and the Iceberg side answers what it is doing to the table.

p = query.lastProgress
print("batch", p["batchId"], "rows", p["numInputRows"])
print("batch duration ms", p["durationMs"]["triggerExecution"])
print("sources", [(s["description"][:40], s["endOffset"]) for s in p["sources"]])

There is no batchDuration key in the progress PySpark returns. The keys are batchId, durationMs, id, inputRowsPerSecond, name, numInputRows, processedRowsPerSecond, runId, sink, sources, stateOperators and timestamp, and a batch’s wall time is durationMs["triggerExecution"].

-- how fast is the table accumulating commits?
SELECT date_trunc('hour', committed_at) AS hour, count(*) AS commits,
       sum(CAST(summary['added-data-files'] AS INT)) AS files
FROM ice.blog_streaming.order_events.snapshots
GROUP BY 1 ORDER BY 1 DESC LIMIT 12;

+-------------------+-------+-----+
|hour               |commits|files|
+-------------------+-------+-----+
|2026-09-30 05:00:00|3      |5    |
+-------------------+-------+-----+

The summary values are strings; without the cast, sum returns 5.0. That query is the one to put on a dashboard. Commits per hour is the number that predicts every metadata problem this table will have, and it is visible long before anything gets slow.

Batch duration approaching the trigger interval is the other early warning: once processing takes longer than the trigger, batches queue and latency grows without bound.

Recovering a stream

Three situations, three different answers.

The query failed and the data is fine. Restart it with the same checkpoint. Offsets and commits reconcile, an incomplete batch re-runs, and the sink skips any epoch it had already committed.

The output is wrong and must be replayed. Do not simply delete the checkpoint: that re-ingests everything into a table that still holds the bad rows. Roll the table back first, then reset the stream to a matching position. After a bad batch of 100 cancelled orders landed through an upsert stream:

before_bad = sql(f"SELECT snapshot_id FROM {TARGET}.refs WHERE name = 'main'").collect()[0][0]
# ... the bad batch lands through the upsert stream ...
sql(f"CALL ice.system.rollback_to_snapshot('blog_streaming.orders_current', {before_bad})")

after a bad batch               : rows=1300 shipped=500
after rollback_to_snapshot      : rows=1200 shipped=500

The schema changed upstream. A new column in the source is additive: add it to the table, and restart. A type change is not, and needs the add-migrate-drop pattern rather than a stream restart.

The checkpoint’s offsets directory records what each batch was told to read, and it is readable when you need to reason about where a stream is. Here the checkpoint lives in MinIO, so read it with the stack’s mc client:

docker exec mc /usr/bin/mc cat minio/warehouse/blog_streaming.db/_stream/ckpt/offsets/1

Trimmed in the middle, where it lists the pinned configuration:

v1
{"batchWatermarkMs":0,"batchTimestampMs":1790745631873,"conf":{"spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider", ... ,"spark.sql.shuffle.partitions":"200"}}
{"logOffset":1}

The first line is the log format version, the second the batch’s timestamp and the configuration the query is pinned to, and the last the file source’s offset.

Maintenance is part of the pipeline

A streaming table without scheduled maintenance is a table with a deadline. Both jobs, from the day it is created:

-- compact what the micro-batches produced, scoped to recent partitions
CALL ice.system.rewrite_data_files(
  table   => 'blog_streaming.events_compact',
  where   => 'event_ts >= current_date() - INTERVAL 1 DAYS',
  options => map('min-input-files','2'));

-- and bound the metadata, which snapshot expiry alone does not touch
ALTER TABLE ice.blog_streaming.events_compact SET TBLPROPERTIES (
  'write.metadata.delete-after-commit.enabled' = 'true',
  'write.metadata.previous-versions-max' = '200');

On a table fed by four micro-batches, each spreading over the last three days, the scoped rewrite compacted only the two days the where clause selected:

+----------+-----+
|       day|files|
+----------+-----+
|2026-09-28|    4|
|2026-09-29|    4|
|2026-09-30|    4|
+----------+-----+
+--------------------------+----------------------+---------------------+-----------------------+--------------------------+
|rewritten_data_files_count|added_data_files_count|rewritten_bytes_count|failed_data_files_count|removed_delete_files_count|
+--------------------------+----------------------+---------------------+-----------------------+--------------------------+
|8                         |2                     |19580                |0                      |0                         |
+--------------------------+----------------------+---------------------+-----------------------+--------------------------+
+----------+-----+
|       day|files|
+----------+-----+
|2026-09-28|    4|
|2026-09-29|    1|
|2026-09-30|    1|
+----------+-----+

The dates are relative: the script stages events for the two days before the day you run it, so your dates will be different and the counts the same.

The metadata properties are the ones people miss. Twenty single-row commits, then expire_snapshots keeping one snapshot:

defaults  after 20 commits: metadata.json=21; after expire_snapshots: metadata.json=22, snapshots=1
bounded   after 20 commits: metadata.json=6; after expire_snapshots: metadata.json=6, snapshots=1

Expiry removed 19 snapshots and deleted no metadata.json files; it added one of its own. With the two properties set (to 5 here, to make the effect visible), the table kept six: the current one and five previous. A per-minute stream without them leaves a new metadata.json behind every minute, forever.

One side effect to expect: on the bounded table, expire_snapshots logs an ERROR from ReachableFileUtil with a NotFoundException stack trace for each older metadata.json it tries to follow in the metadata log and finds already deleted, five of them here. The procedure still succeeds. It is noise from the action’s walk of the metadata history, not a broken table.

The trap that produces a healthy-looking no-op

Staging the input files one directory deep produced this:

layout                          : ['batch0', 'batch1', 'batch2']
micro-batches in the run        : 1
rows read from landing_nested   : 0
snapshots                       : 0
batch read of the same path     : 0

The query started, created its checkpoint, ran, and terminated normally. It read nothing, because the files were in landing_nested/batch0/part-*.parquet and the source was watching landing_nested/. The subdirectories are not in key=value form, so Spark does not treat them as partitions, and it does not descend into them. The last line shows this is not a streaming quirk: a batch spark.read.parquet of the same path also returned 0 rows.

Nothing in that output says “I found no input”. The checkpoint exists, the query exited zero, and only the row count reveals the problem, which is why a silent zero-row ingestion is exactly the failure that survives into production unnoticed.

Three defences, all measured:

recursiveFileLookup=true rows   : 3000
glob landing_nested/* rows      : 3000

Set recursiveFileLookup to true on the source, or point it at a glob such as landing_nested/*, or stage input flat in the watched directory, which is what the measured run above does. Whichever you choose, assert on the row count or the snapshot count after the run rather than on the exit status.

A related trap in this stack: the Spark image’s core-site.xml sets fs.defaultFS to s3a://warehouse/, so a path written as /tmp/landing is not the container’s disk. It is s3a://warehouse/tmp/landing in MinIO. Write the scheme explicitly, as the scripts do.

Row-level operations alongside a stream

A streaming append is the simple case. Streaming upserts go through foreachBatch and MERGE INTO, because the sink itself only appends:

def upsert(batch_df, batch_id):
    batch_df.createOrReplaceTempView("changes")
    batch_df.sparkSession.sql(f"""
        MERGE INTO {TARGET} t USING changes s ON t.order_id = s.order_id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *""")


(spark.readStream.schema(SCHEMA).parquet(changes_dir)
      .writeStream.foreachBatch(upsert)
      .trigger(availableNow=True)
      .option("checkpointLocation", checkpoint)
      .start().awaitTermination())

A thousand new orders, then a batch of 300 updates to existing orders plus 200 new ones, then that second batch applied a second time by hand:

after 1,000 new orders          : rows=1000 shipped=0
after 300 updates + 200 new     : rows=1200 shipped=500
after replaying that batch      : rows=1200 shipped=500
+---------+-----+-------+
|operation|added|deleted|
+---------+-----+-------+
|   append| 1000|   NULL|
|overwrite| 1028|    828|
|overwrite| 1028|   1028|
+---------+-----+-------+

foreachBatch gives at-least-once by default: the same batch_id can run twice after a failure. A MERGE keyed on a primary key is idempotent in its result, which is why this pattern works, and an append inside foreachBatch would not be. It is not free, though: the replay changed nothing and still rewrote 1,028 rows in a new overwrite snapshot, because the table is copy-on-write.

The batch runs in a different session. foreachBatch hands you a DataFrame whose sparkSession is the streaming query’s own, cloned session, with its own Iceberg catalog and table cache. The driver’s spark session had loaded the target before the stream ran, and kept serving that cached state:

foreachBatch session is spark   : False
driver reads, before REFRESH    : rows=0 shipped=0
after 1,000 new orders          : rows=1000 shipped=0

The 1,000 rows were committed; the driver saw none of them until REFRESH TABLE. Worse, a MERGE issued from the driver’s session without that refresh plans against the stale snapshot and fails with ValidationException: Found conflicting files that can contain records matching true. After any foreachBatch stream, refresh the target before the driver reads or writes it.

The delete-file accumulation that this pattern causes on a merge-on-read table is its own problem, covered in copy-on-write or merge-on-read.

Run it yourself

Both scripts run on the datalake-playground stack, with Spark 3.5.9 and Iceberg 1.11.0:

git clone https://github.com/rangareddy/datalake-playground.git
cd datalake-playground
SPARK_VERSION=3.5.9 sh run_datalake.sh start
sh run_datalake.sh status

They stage their input, keep their checkpoints and write their tables under one prefix, s3a://warehouse/blog_streaming.db/, in the namespace ice.blog_streaming.

streaming_file_ingest.py covers the measured run, re-runs, the checkpoint and the sink’s summary, a new checkpoint, the dashboard query, maxFilesPerTrigger and the subdirectory trap:

"""Stream order events from a landing directory into Iceberg with availableNow.

Run inside the datalake-playground stack:
  docker cp streaming_file_ingest.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/streaming_file_ingest.py'

The landing directory, the checkpoints and the table all live in MinIO. Paths
are written with an explicit s3a:// scheme: the stack's core-site.xml sets
fs.defaultFS to s3a://warehouse/, so a bare /tmp/... path would land in MinIO
anyway, just not where you expect.
"""
import json

from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder.appName("iceberg-streaming-ingest")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql

BASE = "s3a://warehouse/blog_streaming.db/_stream"
LANDING = f"{BASE}/landing"            # files are staged flat, directly in here
TABLE = "ice.blog_streaming.order_events"
SCHEMA = "order_id BIGINT, customer_id BIGINT, status STRING, amount DOUBLE, event_ts TIMESTAMP"


jvm = spark._jvm
hconf = spark._jsc.hadoopConfiguration()


def ls(path):
    """Names under a Hadoop path, hidden files (. and _) excluded."""
    p = jvm.org.apache.hadoop.fs.Path(path)
    fs = p.getFileSystem(hconf)
    if not fs.exists(p):
        return []
    return sorted(st.getPath().getName() for st in fs.listStatus(p)
                  if not st.getPath().getName().startswith((".", "_")))


def read_text(path):
    p = jvm.org.apache.hadoop.fs.Path(path)
    stream = p.getFileSystem(hconf).open(p)
    reader = jvm.java.io.BufferedReader(jvm.java.io.InputStreamReader(stream, "UTF-8"))
    lines, line = [], reader.readLine()
    while line is not None:
        lines.append(line)
        line = reader.readLine()
    reader.close()
    return lines


def rm(path):
    p = jvm.org.apache.hadoop.fs.Path(path)
    p.getFileSystem(hconf).delete(p, True)


def stage(directory, file_no, rows=1000):
    """Write one Parquet file of order events into `directory`."""
    (spark.range(rows)
          .select((F.lit(file_no * rows) + F.col("id")).alias("order_id"),
                  (F.col("id") % 250 + 1).alias("customer_id"),
                  F.element_at(F.array(*[F.lit(s) for s in ("PLACED", "PAID", "SHIPPED")]),
                               (F.col("id") % 3 + 1).cast("int")).alias("status"),
                  F.round(F.col("id") % 500 + 9.99, 2).alias("amount"),
                  (F.lit("2026-09-28 00:00:00").cast("timestamp")
                   + F.make_interval(secs=F.col("id") * 60 + file_no)).alias("event_ts"))
          .coalesce(1)
          .write.mode("append").parquet(directory))


def run_available_now(source_dir, table, checkpoint, **source_options):
    """One availableNow run: drain what is in `source_dir`, commit, stop."""
    reader = spark.readStream.schema(SCHEMA)
    for key, value in source_options.items():
        reader = reader.option(key, value)
    query = (reader.parquet(source_dir)
                   .writeStream.format("iceberg").outputMode("append")
                   .trigger(availableNow=True)
                   .option("checkpointLocation", checkpoint)
                   .toTable(table))
    query.awaitTermination()
    return query


def rows(table):
    return sql(f"SELECT count(*) FROM {table}").collect()[0][0]


def snapshots(table):
    return sql(f"SELECT count(*) FROM {table}.snapshots").collect()[0][0]


def data_files(table):
    return sql(f"SELECT count(*) FROM {table}.files").collect()[0][0]


rm(BASE)
sql("CREATE NAMESPACE IF NOT EXISTS ice.blog_streaming")
sql(f"DROP TABLE IF EXISTS {TABLE} PURGE")
sql(f"CREATE TABLE {TABLE} ({SCHEMA}) USING iceberg")

print("== 1. three files, one availableNow run ==")
for n in range(3):
    stage(LANDING, n)
print("files staged                    :", len([f for f in ls(LANDING) if f.endswith(".parquet")]))
q = run_available_now(LANDING, TABLE, f"{BASE}/ckpt")
print("micro-batches in the run        :", len(q.recentProgress))
print("rows after first run            :", rows(TABLE))
print("snapshots                       :", snapshots(TABLE))
print("data files in table             :", data_files(TABLE))

print("\n== 2. the same query again, no new files ==")
q = run_available_now(LANDING, TABLE, f"{BASE}/ckpt")
print("micro-batches in the run        :", len(q.recentProgress))
print("rows after re-run               :", rows(TABLE))
print("snapshots                       :", snapshots(TABLE))

print("\n== 3. a fourth file arrives ==")
stage(LANDING, 3)
q = run_available_now(LANDING, TABLE, f"{BASE}/ckpt")
p = q.lastProgress
print("rows after 4th file             :", rows(TABLE))
print("snapshots                       :", snapshots(TABLE))
print("data files in table             :", data_files(TABLE))
print("lastProgress batchId            :", p["batchId"])
print("lastProgress numInputRows       :", p["numInputRows"])
print("lastProgress keys               :", sorted(p.keys()))
print("lastProgress durationMs keys    :", sorted(p["durationMs"].keys()))
print("lastProgress source             :", p["sources"][0]["description"][:34])

print("\n== 4. what the checkpoint holds ==")
ckpt = f"{BASE}/ckpt"
print("checkpoint entries              :", ls(ckpt))
print("offsets log                     :", ls(f"{ckpt}/offsets"))
print("commits log                     :", ls(f"{ckpt}/commits"))
query_id = json.loads(read_text(f"{ckpt}/metadata")[0])["id"]
print("query id in checkpoint/metadata :", query_id)
latest = max(int(f) for f in ls(f"{ckpt}/offsets") if f.isdigit())
lines = read_text(f"{ckpt}/offsets/{latest}")
print(f"offsets/{latest} line 1            :", lines[0])
print(f"offsets/{latest} last line         :", lines[-1])

print("\n== 5. the sink's own record of progress ==")
sql(f"""SELECT snapshot_id, operation,
               summary['spark.sql.streaming.queryId'] = '{query_id}' AS same_query_id,
               summary['spark.sql.streaming.epochId'] AS epoch_id,
               summary['added-records'] AS added
        FROM {TABLE}.snapshots ORDER BY committed_at""").show(truncate=False)

print("== 6. the same query with a NEW checkpoint ==")
q = run_available_now(LANDING, TABLE, f"{BASE}/ckpt_new")
print("rows after new-checkpoint run   :", rows(TABLE))
print("snapshots                       :", snapshots(TABLE))

print("\n== 7. commits per hour, the dashboard query ==")
sql(f"""SELECT date_trunc('hour', committed_at) AS hour, count(*) AS commits,
               sum(CAST(summary['added-data-files'] AS INT)) AS files
        FROM {TABLE}.snapshots
        GROUP BY 1 ORDER BY 1 DESC LIMIT 12""").show(truncate=False)

print("== 8. maxFilesPerTrigger splits a backlog into batches ==")
BATCHED = "ice.blog_streaming.order_events_batched"
sql(f"DROP TABLE IF EXISTS {BATCHED} PURGE")
sql(f"CREATE TABLE {BATCHED} ({SCHEMA}) USING iceberg")
q = run_available_now(LANDING, BATCHED, f"{BASE}/ckpt_batched", maxFilesPerTrigger=2)
print("micro-batches in the run        :", len(q.recentProgress))
print("rows per batch                  :", [b["numInputRows"] for b in q.recentProgress])
print("rows                            :", rows(BATCHED))
print("snapshots                       :", snapshots(BATCHED))

print("\n== 9. files staged one directory deep ==")
NESTED = f"{BASE}/landing_nested"
NESTED_TABLE = "ice.blog_streaming.order_events_nested"
for n in range(3):
    stage(f"{NESTED}/batch{n}", 10 + n)
print("layout                          :", ls(NESTED))
sql(f"DROP TABLE IF EXISTS {NESTED_TABLE} PURGE")
sql(f"CREATE TABLE {NESTED_TABLE} ({SCHEMA}) USING iceberg")
q = run_available_now(NESTED, NESTED_TABLE, f"{BASE}/ckpt_nested")
print("micro-batches in the run        :", len(q.recentProgress))
print("rows read from landing_nested   :", rows(NESTED_TABLE))
print("snapshots                       :", snapshots(NESTED_TABLE))
print("batch read of the same path     :", spark.read.schema(SCHEMA).parquet(NESTED).count())

print("\n== 10. two ways to read nested input ==")
FIXED = "ice.blog_streaming.order_events_recursive"
sql(f"DROP TABLE IF EXISTS {FIXED} PURGE")
sql(f"CREATE TABLE {FIXED} ({SCHEMA}) USING iceberg")
q = run_available_now(NESTED, FIXED, f"{BASE}/ckpt_recursive", recursiveFileLookup="true")
print("recursiveFileLookup=true rows   :", rows(FIXED))
GLOBBED = "ice.blog_streaming.order_events_glob"
sql(f"DROP TABLE IF EXISTS {GLOBBED} PURGE")
sql(f"CREATE TABLE {GLOBBED} ({SCHEMA}) USING iceberg")
q = run_available_now(f"{NESTED}/*", GLOBBED, f"{BASE}/ckpt_glob")
print("glob landing_nested/* rows      :", rows(GLOBBED))

spark.stop()

streaming_upserts_and_maintenance.py covers fanout on a partitioned table, the foreachBatch upsert and its session, rollback, scoped compaction and metadata retention:

"""Partitioned streaming writes, foreachBatch upserts and the maintenance they need.

Run inside the datalake-playground stack:
  docker cp streaming_upserts_and_maintenance.py spark-master:/tmp/
  docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[2]" \
    --driver-memory 1g --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
    /tmp/streaming_upserts_and_maintenance.py'
"""
from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder.appName("iceberg-streaming-upserts")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.ice", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.ice.type", "hive")
    .config("spark.sql.catalog.ice.uri", "thrift://hive-metastore:9083")
    .config("spark.sql.catalog.ice.warehouse", "s3a://warehouse/")
    .config("spark.eventLog.enabled", "false")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
sql = spark.sql
jvm = spark._jvm
hconf = spark._jsc.hadoopConfiguration()

BASE = "s3a://warehouse/blog_streaming.db/_stream2"
SCHEMA = "order_id BIGINT, customer_id BIGINT, status STRING, amount DOUBLE, event_ts TIMESTAMP"


def rm(path):
    p = jvm.org.apache.hadoop.fs.Path(path)
    p.getFileSystem(hconf).delete(p, True)


def count_files(path, suffix):
    p = jvm.org.apache.hadoop.fs.Path(path)
    fs = p.getFileSystem(hconf)
    if not fs.exists(p):
        return 0
    return sum(1 for st in fs.listStatus(p) if st.getPath().getName().endswith(suffix))


def stage(directory, first_id, rows, status, day_spread=3, start="2026-09-26 12:00:00"):
    """One Parquet file; rows are deliberately NOT sorted by day."""
    (spark.range(rows)
          .select((F.lit(first_id) + F.col("id")).alias("order_id"),
                  (F.col("id") % 250 + 1).alias("customer_id"),
                  F.lit(status).alias("status"),
                  F.round(F.col("id") % 500 + 9.99, 2).alias("amount"),
                  (F.lit(start).cast("timestamp")
                   + F.make_interval(days=(F.col("id") % day_spread).cast("int"))).alias("event_ts"))
          .coalesce(1)
          .write.mode("append").parquet(directory))


def stream(directory):
    return spark.readStream.schema(SCHEMA).parquet(directory)


rm(BASE)
sql("CREATE NAMESPACE IF NOT EXISTS ice.blog_streaming")

print("== 1. unsorted input into a partitioned table ==")
stage(f"{BASE}/events", 0, 3000, "PLACED")
for label, options in (("default", {}), ("fanout-enabled=true", {"fanout-enabled": "true"}),
                       ("fanout-enabled=false", {"fanout-enabled": "false"})):
    table = "ice.blog_streaming.events_" + label.split("=")[-1].replace("-", "_")
    sql(f"DROP TABLE IF EXISTS {table} PURGE")
    sql(f"CREATE TABLE {table} ({SCHEMA}) USING iceberg PARTITIONED BY (days(event_ts))")
    writer = (stream(f"{BASE}/events").writeStream.format("iceberg").outputMode("append")
              .trigger(availableNow=True)
              .option("checkpointLocation", f"{BASE}/ckpt_{label}"))
    for key, value in options.items():
        writer = writer.option(key, value)
    try:
        query = writer.toTable(table)
        query.awaitTermination()
        plan = query._jsq.streamingQuery().lastExecution().executedPlan().toString()
        sorted_in_plan = "Sort [" in plan
        files = sql(f"SELECT count(*) FROM {table}.files").collect()[0][0]
        parts = sql(f"SELECT count(*) FROM {table}.partitions").collect()[0][0]
        print(f"{label:<22} rows={sql(f'SELECT count(*) FROM {table}').collect()[0][0]} "
              f"partitions={parts} data files={files} sort in plan={sorted_in_plan}")
    except Exception as e:
        msg = str(e).replace("\n", " ")
        cause = msg[msg.find("Incoming records"):][:150] if "Incoming records" in msg else msg[:150]
        print(f"{label:<22} FAILED: {cause}")

print("\n== 2. foreachBatch + MERGE for upserts ==")
TARGET = "ice.blog_streaming.orders_current"
sql(f"DROP TABLE IF EXISTS {TARGET} PURGE")
sql(f"CREATE TABLE {TARGET} ({SCHEMA}) USING iceberg")


SESSIONS = []


def upsert(batch_df, batch_id):
    SESSIONS.append(batch_df.sparkSession._jsparkSession.hashCode())
    batch_df.createOrReplaceTempView("changes")
    batch_df.sparkSession.sql(f"""
        MERGE INTO {TARGET} t USING changes s ON t.order_id = s.order_id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *""")


def run_upserts():
    (stream(f"{BASE}/changes").writeStream.foreachBatch(upsert)
         .trigger(availableNow=True)
         .option("checkpointLocation", f"{BASE}/ckpt_merge").start().awaitTermination())


def target_state():
    r = sql(f"SELECT count(*) AS n, count_if(status = 'SHIPPED') AS shipped "
            f"FROM {TARGET}").collect()[0]
    return r["n"], r["shipped"]


stage(f"{BASE}/changes", 0, 1000, "PLACED")
run_upserts()
print("foreachBatch session is spark   :", SESSIONS[-1] == spark._jsparkSession.hashCode())
print("driver reads, before REFRESH    : rows=%d shipped=%d" % target_state())
sql(f"REFRESH TABLE {TARGET}")
print("after 1,000 new orders          : rows=%d shipped=%d" % target_state())
stage(f"{BASE}/changes", 700, 500, "SHIPPED")      # 300 updates (700-999) + 200 new
run_upserts()
sql(f"REFRESH TABLE {TARGET}")
print("after 300 updates + 200 new     : rows=%d shipped=%d" % target_state())
replay = spark.read.schema(SCHEMA).parquet(f"{BASE}/changes").where("status = 'SHIPPED'")
upsert(replay, 1)                                    # the same batch applied twice
print("after replaying that batch      : rows=%d shipped=%d" % target_state())
sql(f"SELECT operation, summary['added-records'] AS added, "
    f"summary['deleted-records'] AS deleted FROM {TARGET}.snapshots ORDER BY committed_at").show()

print("== 3. rolling back a bad batch ==")
before_bad = sql(f"SELECT snapshot_id FROM {TARGET}.refs WHERE name = 'main'").collect()[0][0]
stage(f"{BASE}/changes", 5000, 100, "CANCELLED")
run_upserts()
sql(f"REFRESH TABLE {TARGET}")
print("after a bad batch               : rows=%d shipped=%d" % target_state())
sql(f"CALL ice.system.rollback_to_snapshot('blog_streaming.orders_current', {before_bad})")
print("after rollback_to_snapshot      : rows=%d shipped=%d" % target_state())

print("\n== 4. compaction scoped with current_date() ==")
# Events for the last three days, relative to today, so the where clause below
# selects some of them whatever day you run this.
COMPACT = "ice.blog_streaming.events_compact"
sql(f"DROP TABLE IF EXISTS {COMPACT} PURGE")
sql(f"CREATE TABLE {COMPACT} ({SCHEMA}) USING iceberg PARTITIONED BY (days(event_ts))")
recent = sql("SELECT CAST(date_sub(current_date(), 2) AS TIMESTAMP) + INTERVAL 12 HOURS").collect()[0][0]
for n in range(4):
    stage(f"{BASE}/recent", 10_000 + n * 500, 500, "PLACED", start=str(recent))
(spark.readStream.schema(SCHEMA).option("maxFilesPerTrigger", 1).parquet(f"{BASE}/recent")
     .writeStream.format("iceberg").outputMode("append")
     .trigger(availableNow=True)
     .option("checkpointLocation", f"{BASE}/ckpt_recent").toTable(COMPACT).awaitTermination())
sql(f"""SELECT partition.event_ts_day AS day, count(*) AS files
        FROM {COMPACT}.files GROUP BY 1 ORDER BY 1""").show()
sql("""CALL ice.system.rewrite_data_files(
         table   => 'blog_streaming.events_compact',
         where   => 'event_ts >= current_date() - INTERVAL 1 DAYS',
         options => map('min-input-files','2'))""").show(truncate=False)
sql(f"""SELECT partition.event_ts_day AS day, count(*) AS files
        FROM {COMPACT}.files GROUP BY 1 ORDER BY 1""").show()

print("== 5. metadata.json retention versus snapshot expiry ==")
for label, props in (("defaults", ""),
                     ("bounded", "TBLPROPERTIES ('write.metadata.delete-after-commit.enabled'='true', "
                                 "'write.metadata.previous-versions-max'='5')")):
    table = f"ice.blog_streaming.commits_{label}"
    sql(f"DROP TABLE IF EXISTS {table} PURGE")
    sql(f"CREATE TABLE {table} (id BIGINT) USING iceberg {props}")
    for i in range(20):
        sql(f"INSERT INTO {table} VALUES ({i})")
    loc = sql(f"DESCRIBE TABLE EXTENDED {table}").where("col_name = 'Location'").collect()[0][1]
    before = count_files(f"{loc}/metadata", ".metadata.json")
    sql(f"CALL ice.system.expire_snapshots(table => 'blog_streaming.commits_{label}', "
        f"older_than => TIMESTAMP '2099-01-01 00:00:00', retain_last => 1)")
    after = count_files(f"{loc}/metadata", ".metadata.json")
    snaps = sql(f"SELECT count(*) FROM {table}.snapshots").collect()[0][0]
    print(f"{label:<9} after 20 commits: metadata.json={before}; after expire_snapshots: "
          f"metadata.json={after}, snapshots={snaps}")

spark.stop()

Clean up everything both scripts created: drop the tables and the namespace first, then remove the prefix, which also holds the landing files and the checkpoints.

docker exec spark-master bash -c '$SPARK_HOME/bin/spark-sql --master "local[1]" --driver-memory 1g \
  --jars $(ls $ICEBERG_HOME/iceberg-spark-runtime*.jar) \
  --conf spark.sql.catalog.ice=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.ice.type=hive \
  --conf spark.sql.catalog.ice.uri=thrift://hive-metastore:9083 \
  -e "DROP TABLE IF EXISTS ice.blog_streaming.order_events PURGE; DROP TABLE IF EXISTS ice.blog_streaming.order_events_batched PURGE; DROP TABLE IF EXISTS ice.blog_streaming.order_events_nested PURGE; DROP TABLE IF EXISTS ice.blog_streaming.order_events_recursive PURGE; DROP TABLE IF EXISTS ice.blog_streaming.order_events_glob PURGE; DROP TABLE IF EXISTS ice.blog_streaming.events_default PURGE; DROP TABLE IF EXISTS ice.blog_streaming.events_true PURGE; DROP TABLE IF EXISTS ice.blog_streaming.events_false PURGE; DROP TABLE IF EXISTS ice.blog_streaming.orders_current PURGE; DROP TABLE IF EXISTS ice.blog_streaming.events_compact PURGE; DROP TABLE IF EXISTS ice.blog_streaming.commits_defaults PURGE; DROP TABLE IF EXISTS ice.blog_streaming.commits_bounded PURGE; DROP NAMESPACE ice.blog_streaming"' 2>/dev/null
docker exec mc /usr/bin/mc rm --recursive --force minio/warehouse/blog_streaming.db/

Common misconceptions

“Streaming writes row by row.” It writes micro-batches, each an atomic Iceberg commit.

“One snapshot per file.” One snapshot per micro-batch that produced data. Three files in one availableNow run gave one snapshot.

“Re-running will duplicate data.” Not with the same checkpoint: the checkpoint knows what was read, and the sink knows which epochs it committed. With a new checkpoint, it certainly will.

“Unsorted input needs fanout-enabled.” Fanout is already the default for a table without a sort order, and with it switched off Spark sorts each batch instead.

“availableNow is not really streaming.” It uses the streaming engine, offsets and checkpoint, and processes only what is new. It just stops when it catches up.

“Small files are the only streaming problem.” Metadata files accumulate on the same schedule and are bounded by different settings; snapshot expiry does not remove them.

A model worth keeping

A streaming write into Iceberg is a loop of small batch commits. The trigger sets how often you commit; the checkpoint and the sink’s epoch record make re-running safe; the snapshot count is the bill.

So choose availableNow unless latency requires otherwise, keep the checkpoint as carefully as you keep the table, and schedule compaction and expiry on the same day you create the table rather than the week queries get slow.

References

Trademarks

Apache Iceberg, Apache Spark, Apache Parquet, Apache and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.

Found this useful?

These posts and tools are free. If one saved you an afternoon, you can buy me a coffee.

Buy me a coffee