This feature is in Public Preview.
Asynchronous progress tracking allows Structured Streaming pipelines to checkpoint progress asynchronously and in parallel to the actual data processing within a micro-batch, reducing latency associated with maintaining the
Asynchronous progress tracking does not work with
Trigger.availableNow triggers. Attempting to enable this feature with these triggers results in query failure.
Structured Streaming relies on persisting and managing offsets as progress indicators for query processing. Offset management operation directly impacts processing latency, because no data processing can occur until these operations are complete. Asynchronous progress tracking enables Structured Streaming pipelines to checkpoint progress without being impacted by these offset management operations.
Users can configure the frequency at which progress is checkpointed. The default settings for checkpoint frequency provide good throughput for most queries. Configuring the frequency is helpful for scenarios in which offset management operations occur at a higher rate than they can be processed, which creates an ever increasing backlog of offset management operations. To stem this growing backlog, data processing is blocked or slowed, essentially reverting the processing behavior to eliminate the benefits of asynchronous progress tracking.
Failure recovery time increases with the increase in checkpoint interval time. In case of failure, a pipeline has to reprocess all the data before the previous successful checkpoint. Users can consider this trade-off between lower latency during regular processing and recovery time in case of failure.
enable or disable asynchronous progress tracking
the interval in which we commit offsets and completion commits
Users can use code similar to the code below to enable this feature:
val stream = spark.readStream
val query = stream.writeStream
When async progress tracking is enabled, the framework does not checkpoint progress for every batch. To address this, before you disable asynchronous progress tracking, process at least two micro-batches with the following settings:
Stop the query after at least two micro-batches have finished processing. Now you can safely disable the async progress tracking and restart the query.
If you have disabled asynchronous progress tracking without completing this step, you may encounter the following error:
java.lang.IllegalStateException: batch x doesn't exist
In the driver logs, you might see the following error:
The offset log for batch x doesn't exist, which is required to restart the query from the latest batch x from the offset log. Please ensure there are two subsequent offset logs available for the latest batch via manually deleting the offset file(s). Please also ensure the latest batch for commit log is equal or one batch earlier than the latest batch for offset log.
Following the instructions in this section to disable asynchronous progress tracking allows you to address these errors and repair your streaming workload.
This feature has the following limitations:
Asynchronous progress tracking is only supported in stateless pipelines when using Kafka as a sink.
Exactly once end-to-end processing is not guaranteed with asynchronous progress tracking because offset ranges for batch can be changed in case of failure. Some sinks, such as Kafka, never provide exactly-once guarantees.