java - FlinkCEP的执行时间

Question

我正在使用 Flink 和 FlinkCEP 来检测数据流上的复杂事件。出于研究目的，我只需要测量识别时间。

我正在使用Flink / FlinkCEP - 1.7.1 。我正在使用 env.fromCollection() 在 Flink 环境中创建流功能。之后，我使用FlinkCEP:CEP.pattern(....)与其他一起select和print功能。

我只找到这个帖子:Measure job execution time in flink这很有帮助。它提出了一种返回流环境进程的执行时间的解决方案。这并不是我正在寻找的。

我注意到返回值包括其他运算符的时间，例如 .assignAscendingTimestamps(x => x.TimeStamp())因此我无法使用它。

有没有办法只测量CEP.pattern的时间过程？在这种情况下，我也找不到对我有帮助的指标，除非我错过了一些东西......

Answer 1

您可以向每条记录添加一个时间戳字段，并在 CEP 之前使用 mapFunction 将当前时间放入该字段中。然后使用它在 RichMapFunction 中立即计算 CEP 内耗时 - 然后您可以通过自定义指标进行报告，或发送到接收器。这会增加一点开销，但不会太多。只要您可以避免这两个函数之间的任何 keyBy 或重新平衡调用，所涉及的所有内容都将通过函数调用链接在一起，而无需任何序列化或网络开销。