Spark 结构化流 - 对不同的 GroupBy 键使用不同的窗口

Question

目前我在通过 Spark 结构化流阅读 Kafka 主题后有下表

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  

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

例如
key1 将在 1 分钟的窗口内聚合以产生

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 分钟的窗口内聚合以产生

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 结构化流中的密钥指定窗口，或者还有其他方法吗？

Answer 1

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

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

还有：

• Apache Spark 结构化流中的任意状态处理
• 深入了解有状态流处理
• 官方文档