Apache传输无界数据处理和计算聚合，如计数、每个窗口每个键的持续时间。

Question

我是一个新的波束管道，我有一个要求，以计算聚合统计(计数和持续时间为每个窗口非常30分钟左右)从事件收到的卡夫卡主题(无界源)。

Events

 {"id":"xxxxx", "state": "start", "timestamp": 1625718600000, "device": "device-1", ...}
 {"id":"xxxxx", "state": "end", "timestamp": 1625721300000, "device": "device-1",. ..}
 {"id":"yyyyy", "state": "start", "timestamp": 1625718600000, "device": "device-2", ...}
 {"id":"yyyyy", "state": "end", "timestamp": 1625719500000, "device": "device-2", ...}

事件"xxxxx“10:00开始，10:45结束

活动"yyyyy“10:00开始，10:15结束

预期来自管道的统计数据

Device    Interval       Count    Duration
device-1  10:00-10:30    1        30 min
device-2  10:00-10:30    1        15 min
device-1  10:30-11:00    0        15 min 

我使用固定窗口、触发器、groupByKey、CombineFn等，我成功地计算了聚集的计数器，如果事件状态是“开始”，则增加计数，但是，即使在状态完全处理的情况下，我也无法计算重叠窗口的持续时间。

注意:在对事件进行分组时使用事件标识符。

请在这个问题上给我建议。

Answer 1

听起来你需要做的是计算同一工人的开始和结束事件来计算差异，对吗？

我想不出有什么办法可以做到这一点。

difference.

• Stateful DoFn

• 使用GroupByKey --您可以将偶数ID设置为键，并执行GroupByKey来根据键对事件进行分组，并计算和输出

• --当您接收事件时，将它们存储在按ID键键的状态中，以及在存储检查另一个事件的可用性并计算差异时。

需要注意的一点是，在两个不同的Windows中，开始事件和结束事件都有可能发生。在这种情况下，上述任何一种解决方案都无法工作，因为不同的Windows计算方式不同。我认为你将不得不调整管道来解释这种情况(罕见吗?)发生的事。