Spark Structured streaming- 对不同的 GroupBy 键使用不同的 Windows

Question

目前我在通过 spark 结构化流阅读 Kafka 主题后关注 table

key,timestamp,value  
-----------------------------------
key1,2017-11-14 07:50:00+0000,10    
key1,2017-11-14 07:50:10+0000,10  
key1,2017-11-14 07:51:00+0000,10    
key1,2017-11-14 07:51:10+0000,10    
key1,2017-11-14 07:52:00+0000,10    
key1,2017-11-14 07:52:10+0000,10  

key2,2017-11-14 07:50:00+0000,10  
key2,2017-11-14 07:51:00+0000,10  
key2,2017-11-14 07:52:10+0000,10  
key2,2017-11-14 07:53:00+0000,10

我想为每个键使用不同的 windows 并执行聚合

例如
key1 将在 1 分钟的 window 上聚合以产生

key,window,sum
------------------------------------------
key1,[2017-11-14 07:50:00+0000,2017-11-14 07:51:00+0000],20  
key1,[2017-11-14 07:51:00+0000,2017-11-14 07:52:00+0000],20  
key1,[2017-11-14 07:52:00+0000,2017-11-14 07:53:00+0000],20

key2 将在 2 分钟的 window 上聚合以产生

key,window,sum
------------------------------------------
key2,[2017-11-14 07:50:00+0000,2017-11-14 07:52:00+0000],20  
key2,[2017-11-14 07:52:00+0000,2017-11-14 07:54:00+0000],20

目前我在做以下事情

var l1 = List(List(key1,"60 seconds"),List(key2,"120 seconds"))  
l1.foreach{list => 

    val filtered_df = df.filter($"key" === list(0))

    val windowedPlantSum = filtered_df
        .withWatermark("timestamp", "120 minutes")
        .groupBy(
          window($"timestamp", list(1)),
          $"key"
        )
        .agg(sum("value").alias("sum")

    //start the stream
}

上述方法启动了 2 个独立的流。在我的例子中，有 200 个这样的键启动 200 个流，但由于内存问题而失败。

有什么方法可以根据 Spark 结构化流中的键来指定 window 还是有其他方法？

Answer 1

我猜你必须使用 mapGroupsWithState 来只管理一个查询

来自第 28 张幻灯片：https://www.slideshare.net/databricks/arbitrary-stateful-aggregations-using-structured-streaming-in-apache-spark

还有：

Spark Structured streaming- 对不同的 GroupBy 键使用不同的 Windows

Spark Structured streaming- Using different Windows for different GroupBy Keys

scala

apache-spark

spark-streaming

apache-spark-sql

spark-dataframe