How do I use the availableNow trigger for incremental batch?

Summary

• Write the job as a streaming query with a durable checkpointLocation, add .trigger(availableNow=True) (Scala: Trigger.AvailableNow()), call awaitTermination(), and launch it on a schedule. Each run resumes from the checkpoint and exits once it has caught up.
• Size each run's micro-batches with source options: maxFilesPerTrigger or maxBytesPerTrigger for file sources (one or the other), the same pair for Delta Lake sources (files default to 1000), and maxOffsetsPerTrigger for Apache Kafka.
• A failed run needs no cleanup. The next run processes the uncommitted batches first, then everything else that's available, so you rerun the same job.
• For merges or writes to several tables, use foreachBatch. With Delta Lake, pass txnAppId and txnVersion so a retried batch isn't written twice.
• On Databricks, the usual pattern is Auto Loader plus a scheduled Lakeflow Jobs task, and serverless compute recommends this trigger.

How do I use the availableNow trigger for incremental batch?

Treat the job as a stream that you run on a schedule. Define a streaming read, write it with a durable checkpointLocation, set .trigger(availableNow=True), start it and block on awaitTermination(). Each time the scheduler launches it, Spark reads the checkpoint, processes every record that arrived since the last commit and stops. You keep batch-style compute (start, catch up, shut down) while the streaming engine does the bookkeeping, so there's no hand-maintained table of files or offsets already loaded. The trigger arrived in Apache Spark 3.3.0.

Step 1: write the query

This job lands a Kafka topic in a Delta table. With open-source Spark, add the spark-sql-kafka-0-10_2.13 and delta-spark_2.13 packages and the Delta session settings from the Delta Lake quick start.

orders = (spark.readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
          .option("subscribe", "orders")
          .option("startingOffsets", "earliest")
          .option("maxOffsetsPerTrigger", 50000)
          .load()
          .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)"))

query = (orders.writeStream
         .format("delta")
         .option("checkpointLocation", "/checkpoints/orders_raw")
         .trigger(availableNow=True)
         .toTable("orders_raw"))

query.awaitTermination()

startingOffsets only matters for the first run. After that, the query resumes from where the checkpoint left off. maxOffsetsPerTrigger splits a large backlog into several micro-batches, and Spark divides that total proportionally across topic partitions.

Step 2: size the micro-batches

Option	When to use it	Trade-off
maxFilesPerTrigger (file source)	Limit how many new files each micro-batch takes	No limit by default, and it can't be set together with maxBytesPerTrigger
maxBytesPerTrigger (file source)	Size micro-batches by input volume	A stream still reads at least one file, even one larger than the limit
maxFilesPerTrigger, maxBytesPerTrigger (Delta Lake source)	Limit work per micro-batch from a Delta table	Files default to 1000; the byte limit is a soft maximum
maxOffsetsPerTrigger (Kafka)	Cap the records in each micro-batch	Unset by default, so nothing splits a large backlog
startingOffsets (Kafka)	Pick where the first run begins	Ignored once the query resumes from a checkpoint

You can tune rate limits between runs. The Structured Streaming guide allows adding, removing or modifying them on restart from the same checkpoint, while changing the number or type of input sources isn't allowed.

Step 3: make custom writes idempotent

toTable() covers plain appends. For a MERGE, writes to several tables or a sink with only a batch writer, use foreachBatch, which hands your function each micro-batch and its ID. The Delta Lake docs point out that foreachBatch writes aren't idempotent on their own, because a retried batch doesn't know it already ran. Their fix is two writer options:

app_id = "orders_curated_loader"

def write_batch(batch_df, batch_id):
    (batch_df.write
        .format("delta")
        .mode("append")
        .option("txnAppId", app_id)
        .option("txnVersion", batch_id)
        .save("/lake/orders_curated"))

(spark.readStream.table("orders_raw")
    .writeStream
    .foreachBatch(write_batch)
    .option("checkpointLocation", "/checkpoints/orders_curated")
    .trigger(availableNow=True)
    .start()
    .awaitTermination())

Delta Lake ignores any write whose txnVersion isn't higher than the last one it recorded for that txnAppId. If you ever delete the checkpoint and start over, change app_id too, or the restarted query's writes will be ignored because batch IDs start again from 0.

Step 4: schedule it

Any scheduler that can launch a Spark application works, from cron to Apache Airflow. Keep runs from overlapping on the same checkpoint. Each run gives the documented guarantees: uncommitted batches from a failed run are processed first, the watermark advances per batch, and a no-data batch runs before termination if the last batch advanced the watermark. Spark falls back to a single one-time micro-batch when any of the query's sources lacks Trigger.AvailableNow support, so confirm it for every source. Kafka support also arrived in Spark 3.3.0, and Delta Lake added it for Delta sources in 2.1.0.

How Databricks approaches incremental batch

Databricks recommends Auto Loader, the cloudFiles Structured Streaming source, for ingesting files from cloud object storage. It records discovered files in a RocksDB store inside the checkpoint, which keeps processing exactly once, and it pairs with availableNow for scheduled runs. Its cloudFiles.maxFilesPerTrigger defaults to 1000, and in Databricks Runtime 18.0 and above the file and byte limits are configured dynamically. Lakeflow Jobs can start the task on a time-based schedule, when files arrive or when source tables update, and by default only one run of a job is active at a time. Serverless compute supports Trigger.AvailableNow() and recommends it. To skip checkpoint management entirely, run the ingestion in Lakeflow Pipelines: in triggered mode, a pipeline refreshes the available data and stops.

FAQs

How is this different from a plain batch job?

A batch job reads whatever its input holds unless you track progress yourself. An availableNow query saves the range of offsets processed in each trigger to its checkpoint, so every run sees only new data.

What if a run fails halfway?

Run it again. With replayable sources and idempotent sinks, Structured Streaming's checkpointing and write-ahead logs give end-to-end exactly-once semantics, and the retry processes the unfinished batches first.

How do I write it in Scala?

Import org.apache.spark.sql.streaming.Trigger and call .trigger(Trigger.AvailableNow()) on the stream writer.