如何使用 flink cep 库检测 babbling 模式答案

【问题标题】：How to detect babbling patterns using the flink cep library如何使用 flink cep 库检测 babbling 模式
【发布时间】：2020-05-16 22:05:28
【问题描述】：

如何使用 flink cep 库检测 babbling 模式？

示例：假设设备有一些问题，它会不断地发布诸如开、关之类的值。如何使用 CEP 检测模式，如果问题存在 30 分钟。我在下面提到的一些示例数据。

OFF     16/08/18 11:38
ON      16/08/18 11:38
OFF     16/08/18 11:38
ON      16/08/18 11:37
OFF     16/08/18 11:37
ON      16/08/18 11:36
OFF     16/08/18 11:36
OFF     16/08/18 11:36
ON      16/08/18 11:36
OFF     16/08/18 11:35
ON      16/08/18 11:35
ON      16/08/18 11:34
OFF     16/08/18 11:34

【问题讨论】：

这是关于算法或工具的问题吗？
我需要为有故障的设备生成警报。正如您在上述情况下看到的，故障意味着每隔几秒的设备就会连续发送 ON 或 OFF。
那么，故障设备的特征究竟是什么？仅事件率就足够了吗，还是您还需要考虑它在 OFF 和 ON 之间快速切换？
我只需要考虑在特定的持续时间内快速打开和关闭设备。假设设备在 15 分钟内快速切换到 ON 和 OFF，则发出警报。

标签： apache-flink flink-streaming flink-cep

【解决方案1】：

如果您的流是按时间排序的（只需为每个单独的设备对流进行排序），那么您可以轻松地转换流以使分析更容易。像这样的RichFlatMapFunction 会将 ON OFF 事件序列转换为状态 CHANGE 事件序列：

static class DetectChanges extends RichFlatMapFunction<String, String> {
    private transient ValueState<String> previousState;

    @Override
    public void open(Configuration parameters) throws Exception {
        previousState = getRuntimeContext().getState(new ValueStateDescriptor<>("previousState", String.class));
    }

    @Override
    public void flatMap(String onOrOff, Collector<String> out) throws Exception {

        if (previousState.value() != onOrOff) {
            out.collect("CHANGE");
            previousState.update(onOrOff);
        }
    }
}

现在问题已简化为确定流在一段时间内是否有一定数量的 CHANGE 事件。这可以通过滑动窗口轻松完成，或者如果您愿意，也可以使用 CEP。

您也可以完全使用 CEP 来做到这一点。从概念上讲，您可以按如下方式处理：

定义一个匹配 ON+ OFF+ 的单独模式
然后定义一个模式组，只要它在某个时间间隔内出现 n 次，就匹配该 ON/OFF 模式

【讨论】：

是否只能使用 CEP createPattern 数据流？
是的，当然。但它更难（在我看来），这就是我没有那样做的原因。
OK.. 这很好，但在我们的例子中，如果我使用 CEP 创建模式会更容易一些。如果我会做任何其他可能影响主要领域的更改。感谢您分享 CEP 模式解决方案。这真的很有帮助。
我想到了一种使用 CEP 的方法，它比我最初想象的更容易，我已将其添加到我的答案中。如果您在编写代码时遇到困难，请分享您正在进行的工作。
顺便说一句，你也可以使用 Flink SQL 和 MATCH_RECOGNIZE 来做到这一点。