title: How to aggregate the microbatch data into a dataframe for spark structured streaming?
body: <p>The use case is as following. Based on the spark structured streaming, the kafka data is ingested. We hope that the data for each microbatch, every 10 seconds, can be processed and aggregated into a single dataframe, which keeps on monitoring the sum of some values for each id. Is the following way correct? It seems the old data from the previous microbatches are still in the 'monitoring_table' for each query, which causes the unwanted aggregation. What is the best way to solve this issue? Thanks!</p>
  <pre><code>    val monitoring_stream = df.writeStream
                                .outputMode(&quot;append&quot;)
                                .format(&quot;memory&quot;)
                                .queryName(&quot;monitoring_table&quot;)
                                .start()
  
        var rounds = 0
        
        while(monitoring_stream.isActive) {
            Thread.sleep(10000)
    
            spark.sql(&quot;SELECT * from monitoring_table&quot;).show()   
            var tempDF = spark.sql(&quot;SELECT * from monitoring_table&quot;)
            var batchDF_group = tempDF.withWatermark(&quot;timestamp&quot;, &quot;10 seconds&quot;).groupBy(&quot;id&quot;).sum(&quot;download_volume&quot;, &quot;upload_volume&quot;).withColumnRenamed(&quot;sum(download_volume)&quot;,&quot;total_download_volume_batch&quot;).withColumnRenamed(&quot;sum(upload_volume)&quot;,&quot;total_upload_volume_batch&quot;)
            monitoring_df = monitoring_df.join(batchDF_group, monitoring_df(&quot;id&quot;) === batchDF_group(&quot;id&quot;), &quot;left&quot;).select(monitoring_df(&quot;id&quot;), monitoring_df(&quot;total_download_volume&quot;), monitoring_df(&quot;upload_volume&quot;), monitoring_df(&quot;total_volume&quot;), batchDF_group(&quot;total_download_volume_batch&quot;), batchDF_group(&quot;total_upload_volume_batch&quot;)).na.fill(0)
            monitoring_df = monitoring_df.withColumn(&quot;total_upload_volume&quot;, monitoring_df(&quot;total_upload_volume&quot;)+monitoring_df(&quot;total_upload_volume_batch&quot;))
            monitoring_df = monitoring_df.withColumn(&quot;total_download_volume&quot;, monitoring_df(&quot;total_download_volume&quot;)+monitoring_df(&quot;total_download_volume_batch&quot;))
            monitoring_df = monitoring_df.withColumn(&quot;total_volume&quot;, monitoring_df(&quot;total_download_volume&quot;)+monitoring_df(&quot;total_upload_volume&quot;))                                    
  
            
            monitoring_df.show()    
      }```
  </code></pre>