Multiple aggregations in Spark Structured Streaming

Question

I would like to do multiple aggregations in Spark Structured Streaming.

Something like this:

• Read a stream of input files (from a folder)
• Perform aggregation 1 (with some transformations)
• Perform aggregation 2 (and more transformations)

When I run this in Structured Streaming, it gives me an error "Multiple streaming aggregations are not supported with streaming DataFrames/Datasets".

Is there a way to do such multiple aggregations in Structured Streaming?

Answer 1

This is not supported, but there are other ways also. Like performing single aggregation and saving it to kafka. Read it from kafka and apply aggregation again. This has worked for me.

Answer 2

As in Spark 2.4.4 (latest for now) is NOT support the Multiple streaming aggregations you can use the .foreachBatch() method

A dummy example:

query =  spark
        .readStream
        .format('kafka')
        .option(..)
        .load()

       .writeStream
       .trigger(processingTime='x seconds')
       .outputMode('append')
       .foreachBatch(foreach_batch_function)
       .start()

query.awaitTermination()        


def foreach_batch_function(df, epoch_id):
     # Transformations (many aggregations)
     pass