分组事件

您需要解决的第一个问题是如何将开始/停止事件分组在一起？
很可能，你会想把他们按某种方式分组 USER_ID 或其他唯一标识正在启动/停止流的广播者的属性。
很有可能，你也会希望通过某种 STREAM_ID 或其他唯一标识正在播放的流的属性。
这可能就足够了，只要您只需要每个广播公司、每个视频的总播放时间。但是，您可能还需要考虑时间。例如，如果我今天看一个视频，明天再看一次，那是两个观看时段，两个独立的观看时间总和，还是你不在乎？
对事件进行时间分组的一种方法是使用会话窗口。在对数据进行会话化之前，需要定义定义会话的参数。下面是一个在ksqldb中使用会话窗口的好例子。
另一种将事件按时间分组的方法是使用滚动窗口。下面是一个使用滚动窗口的好例子。

计算播放时间

一旦你将你的事件分组，你可能需要计算游戏时间。例如，如果我在时间5开始播放，在时间8停止播放，那么我观看视频的时间量是 5 - 8 = 3 .
这需要捕获播放事件并等待停止事件，然后输出时间差。以容错的方式做一些事情。
在编写本文时，这将需要一个自定义的udaf（自定义用户定义的聚合函数）。
自定义udaf可以捕获开始事件，将其存储以供将来参考，并输出播放时间的“0”，然后当它看到相应的停止事件时，可以将开始事件从其状态中移除，计算播放时间并返回它。
这是一个在ksqldb中编写自定义udf的好例子，不过您需要一个自定义udaf，本文将介绍这一点。
目前有一个公关开放的一个增强到最新的\u由\u抵消方法，可能很好地服务于您的目的。这增强了方法，使其能够捕获最后的n值，而不仅仅是最后的1值。很可能，这将在ksqldbv0.13中发布，如果您有任何开发经验，您可以随时在本地提取代码并编译它。如果它不符合你的目的，那么你可以用它作为发展自己的起点。
当然，这些解决方案需要对源事件流进行正确排序，以便停止事件永远不会出现在其关联的播放事件之前。

聚合

一旦计算了一对开始/停止事件之间的播放时间，就需要对它们进行聚合。下面是如何在ksqldb中聚合的一个很好的示例。