[jira] [Commented] (SPARK-48418) Spark structured streaming: Add microbatch timestamp to foreachBatch method

["Anil Dasari (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-48418?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17849465#comment-17849465 ] 

Anil Dasari commented on SPARK-48418:
-------------------------------------

Using the current time as the batch time in foreachBatch might not work because this information is not available in the query progress and cannot be mapped.

> Spark structured streaming: Add microbatch timestamp to foreachBatch method
> ---------------------------------------------------------------------------
>
>                 Key: SPARK-48418
>                 URL: https://issues.apache.org/jira/browse/SPARK-48418
>             Project: Spark
>          Issue Type: Improvement
>          Components: Structured Streaming
>    Affects Versions: 3.5.1
>            Reporter: Anil Dasari
>            Priority: Major
>
> We are on Spark 3.x and using Spark dstream + kafka and planning to use structured streaming + Kafka.
> Differences in Dstream microbatch and structured streaming microbatch metadata is making migration difficult. 
> Dstream#foreachRDD gives both microbatch RDD and start timestamp (in long). However, structured streaming Dataset#foreachBatch returns only microbatch dataset and batchID where BatchID is a numeric number. 
> micorbatch start time used across our data pipelines and final result. 
> Could you add microbatch start timestamp to  Dataset#foreachBatch method?
> Pseudo code :
>  
> {code:java}
> val inputStream = sparkSession.readStream.format("rate").load
> inputStream
>   .writeStream
>   .trigger(Trigger.ProcessingTime(10 * 1000))
>   .foreachBatch {
>     (ds: Dataset[Row], batchId: Long, batchTime: Long) => // batchTime is microbatch triggered/start timestamp
>       
>      // application logic.
>   }
>   .start()
>   .awaitTermination() {code}
>  
> Implementation approach when batchTime is trigger executor executed time:
> ( `currentTriggerStartTimestamp` can be used as well as batch time. Trigger executor time is source of microbatch and also can be easily added to query processor event as well)
> 1. Add trigger time to [TriggerExecutor|https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala#L31] 
> {code:java}
> trait TriggerExecutor {
>   // added batchTime (Long) argument 
>   def execute(batchRunner: (MicroBatchExecutionContext, Long) => Boolean): Unit 
>   ... // (other methods)
> }{code}
> 2. Update ProcessingTimeExecutor and other executors to pass trigger time.
> {code:java}
> override def execute(triggerHandler: (MicroBatchExecutionContext, Long) => Boolean): Unit = {
>   while (true) {
>     val triggerTimeMs = clock.getTimeMillis()
>     val nextTriggerTimeMs = nextBatchTime(triggerTimeMs)
>     
>     // pass triggerTimeMs to runOneBatch which invokes triggerHandler and is used in MicroBatchExecution#runActivatedStream method.
>     val terminated = !runOneBatch(triggerHandler, triggerTimeMs)
>     
>    ...
>   }
> } {code}
> 3. Add argument executionTime (long) argument to MicroBatchExecution#excuteOneBatch method [here|https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala#L330]
> 4. Pass execution time in [runBatch|https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala#L380C11-L380C19] and [here|https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala#L849]
> 5. Finally add following following method `foreachBatch` in `DataStreamWriter` and update existing `foreachBatch` methods for new argument.  and also add it to query processor event. 
> {code:java}
> def foreachBatch(function: (Dataset[T], Long, Long) => Unit): DataStreamWriter[T] = {
>   this.source = SOURCE_NAME_FOREACH_BATCH
>   if (function == null) throw new IllegalArgumentException("foreachBatch function cannot be null")
>   this.foreachBatchWriter = function
>   this
> }{code}
> Let me know your thoughts. 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org