flink CEP 介绍和实例

大数据开发技术博客 / 2023-05-18 / 原文

1，Flink介绍

Flink 是一个分布式的基于状态计算的流处理计算引擎，或者说框架，可以处理有边界流数据和无边界流数据，在内存中执行计算，而且具有任意扩展计算能力。最初由柏林工业大学的xxx 小组研发，后被阿里巴巴收购。初略看起来，和spark 功能类似，但是某些特征优于spark。

Flink 提供了对于复杂事件模式识别的功能，并以提供一个专用库（flink-cep）,方便用户开发CEP 应用。

2, CEP 定义和场景

CEP 英文全称是“complex event processing” , 即“复杂事件处理” ，它可以探查无限流数据中的复杂模式，使得抓取其中的重要事件。

通过这种能力，我们可以对大型应用，海量用户行为，海量数据进行实时监控，并提取所需信息，尤其是重要的，危急，欺诈的信息。

适用的场景包括，银行应用的信用欺诈，大型系统的系统失效恢复，用户行为监控（直播平台用户弹幕合规），舆情监控，等等。

3，Flink cep 开发流程：

1，定义flink 环境

2，输入流

3，定义复杂模式

4，应用模式，

5，提取输出数据

4，CEP 代码实例：

4.1，功能：

输入流为socket 字符模拟应用日志流。

自定义需要捕捉的规则为：在1分钟内，如果某个实体依次发送 warn 和 danger级别日志，则捕捉到该数据，并输出到控制板。

4.2 ，代码：

public class FraudDetectionJob {
    public static void main(String[] args) throws Exception {
        System.out.println("Hello World!");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

// 输入流定义，从socket 读入模拟的流数据
        DataStream<Event> input = env
                .socketTextStream("localhost", 9000)
                .map(str -> new Event(str.split(",")[0], str.split(",")[1], str.split(",")[2]))
                /*.keyBy(event -> event.getId())*/;

        // 定义需要捕捉的事件规则
        Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
                .where(SimpleCondition.of(event -> event.getLevel().equals("warn")))
                .next("end")
                .where(SimpleCondition.of(event -> event.getLevel().equals("danger")))
                .within(Time.minutes(1));

　　　　//对输入流适配规则
        PatternStream<Event> patternStream = CEP.pattern(input.keyBy(e -> e.getId()), pattern);

        //输出结果流事件
        DataStream<Alert> result = patternStream.process(
                new PatternProcessFunction<Event, Alert>() {
                    @Override
                    public void processMatch(
                            Map<String, List<Event>> match,
                            Context ctx,
                            Collector<Alert> out) throws Exception {

                        Event warn = match.get("middle").get(0);
                        Event danger = match.get("danger").get(0);
                        Alert alter = new Alert(warn, danger);
                        out.collect(alter);
                    }
                });

        result.print();

     }    
}