脚本：
我有一个来自传感器的事件流。事件可能是t型或j型。
t型事件具有事件发生时间戳。
j-type事件具有开始和结束时间戳。
根据j-type事件的开始和结束时间戳，对时间范围内的所有t-type事件应用聚合逻辑，并将结果写入数据库。
为此，我创建了一个自定义触发器，当接收到j-type事件时触发它。在我的自定义processwindowfunction中，我正在执行聚合逻辑和时间检查。
但是，可能有一种情况，t型事件不在当前j型事件的时间范围内。在这种情况下，在清除当前窗口之前，应该将t-type事件推送到下一个窗口。
解决方案思路：
在自定义窗口处理函数中，将未处理的t型事件推送到kinesis流（源）中(最坏情况解决方案）
不要使用fire和purge，而是使用fire在整个运行时保持状态。使用元素迭代器删除已处理的元素。（不推荐，以保持无限窗口）
我想知道，是否有任何方法可以直接将未处理的事件推回到输入流（不带kinesis）(重新排队）
或者
有没有办法在keyby上下文中保持状态，以便我们对这些未处理的数据（在或之前）以及窗口元素执行计算。
1条答案
按热度按时间xlpyo6sf1#
这里有两个解决方案。它们在基本行为上或多或少是等价的，但是您可能会发现其中一个更容易理解、维护或测试。
至于你的问题，不，没有办法循环（重新排队）未消费的事件，而不把它们推回到动觉。但只要坚持到需要的时候就可以了。
解决方案1：使用richflatmapfunction
当t型事件到达时，将它们附加到
ListState对象。当j-type事件到达时，将列表中所有匹配的t-type事件收集到输出中，并更新列表以仅保留那些将属于以后j-type事件的t-type事件。解决方案2：使用带有自定义触发器和逐出器的globalwindows
除了已经完成的工作之外，实现
Evictor在触发窗口之后，只从窗口中删除j-type事件和所有匹配的t-type事件。更新：清除过时密钥/失效传感器的状态
在解决方案1中，可以使用statettl来安排清除与死键相关联的任何非活动状态。或者你可以用一个
KeyedProcessFunction而不是一个RichFlatMapFunction，并使用计时器来完成相同的任务。使用windowapi管理过时密钥的状态可能不那么简单，但是对于解决方案2，我相信您可以扩展您的自定义触发器以包含一个将清除窗口的超时。如果你在
ProcessWindowFunction，您将需要依赖于状态ttl来清理它。