[SPARK-40849] Async log purge

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.4.0
• Fix Version/s: 3.4.0
• Component/s: Structured Streaming
• Assignee: Boyang Jerry Peng
• Reporter: Boyang Jerry Peng
• Created: 19/Oct/22 23:13
• Updated: 24/Oct/22 02:09
• Resolved: 24/Oct/22 02:09

Description

Purging old entries in both the offset log and commit log will be done asynchronously.

 

For every micro-batch, older entries in both offset log and commit log are deleted. This is done so that the offset log and commit log do not continually grow.  Please reference logic here

 

https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala#L539 

 

The time spent performing these log purges is grouped with the “walCommit” execution time in the StreamingProgressListener metrics.  Around two thirds of the “walCommit” execution time is performing these purge operations thus making these operations asynchronous will also reduce latency.  Also, we do not necessarily need to perform the purges every micro-batch.  When these purges are executed asynchronously, they do not need to block micro-batch execution and we don’t need to start another purge until the current one is finished.  The purges can happen essentially in the background.  We will just have to synchronize the purges with the offset WAL commits and completion commits so that we don’t have concurrent modifications of the offset log and commit log.

Issue Links

• Is contained by: SPARK-40025 Project Lightspeed: Faster and Simpler Stream Processing with Apache Spark
• links to: [Github] Pull Request #38313 (jerrypeng)