Flink系列 - 实时数仓之CEP预警实战

大数据真好玩

发布于 2021-01-26 16:15:32

1.6K00

代码可运行

文章被收录于专栏：暴走大数据暴走大数据

运行总次数：0

代码可运行

CEP 即Complex Event Processing - 复杂事件，Flink CEP 是在 Flink 中实现的复杂时间处理(CEP)库。处理事件的规则，被叫做“模式”(Pattern)，Flink CEP 提供了 Pattern API，用于对输入流数据进行复杂事件规则定义，用来提取符合规则的事件序列。

Pattern API 大致分为三种：个体模式，组合模式，模式组。

Flink CEP 应用场景

CEP 在互联网各个行业都有应用，例如金融、物流、电商、智能交通、物联网行业等行业：

实时监控：我们需要在大量的订单交易中发现那些虚假交易，在网站的访问日志中寻找那些使用脚本或者工具“爆破”登录的用户，或者在快递运输中发现那些滞留很久没有签收的包裹等。
风险控制：比如金融行业可以用来进行风险控制和欺诈识别，从交易信息中寻找那些可能存在危险交易和非法交易。
营销广告：跟踪用户的实时行为，指定对应的推广策略进行推送，提高广告的转化率。

Flink CEP 开发流程

DataSource 中的数据转换为 DataStream；
定义 Pattern，并将 DataStream 和 Pattern 组合转换为 PatternStream；
PatternStream 经过 select、process 等算子转换为 DataStraem；
再次转换的 DataStream 经过处理后，sink 到目标库。

接下来我们讲对超时未支付、连续登录、交易活跃用户这三个场景进行实操。

一、超时未支付

需求：找出那些下单后 10 分钟内没有支付的订单。创建 TimeOutPayCEPMain.java 类并编写整体代码：

public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<PayEvent> source = env.fromElements(
                new PayEvent(1L, "create", 1597905234000L),
                new PayEvent(1L, "pay", 1597905235000L),
                new PayEvent(2L, "create", 1597905236000L),
                new PayEvent(2L, "pay", 1597905237000L),
                new PayEvent(3L, "create", 1597905239000L)

        ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<PayEvent>(Time.milliseconds(500L)) {
            @Override
            public long extractTimestamp(PayEvent payEvent) {
                return payEvent.getTimestamp();
            }
        }).keyBy(new KeySelector<PayEvent, Object>() {
            @Override
            public Object getKey(PayEvent value) throws Exception {
                return value.getUserId();
            }
        });

        // 逻辑处理代码

        env.execute("execute cep");
}

逻辑处理代码如下：

OutputTag<PayEvent> orderTimeoutOutput = new OutputTag<PayEvent>("orderTimeout") {};
        Pattern<PayEvent, PayEvent> pattern = Pattern.<PayEvent>
                begin("begin")
                .where(new IterativeCondition<PayEvent>() {
                    @Override
                    public boolean filter(PayEvent payEvent, Context context) throws Exception {
                        return payEvent.getActive().equals("create");
                    }
                })
                .followedBy("pay")
                .where(new IterativeCondition<PayEvent>() {
                    @Override
                    public boolean filter(PayEvent payEvent, Context context) throws Exception {
                        return payEvent.getActive().equals("pay");
                    }
                })
                .within(Time.seconds(600));

        PatternStream<PayEvent> patternStream = CEP.pattern(source, pattern);
        SingleOutputStreamOperator<PayEvent> result = patternStream.select(orderTimeoutOutput, new PatternTimeoutFunction<PayEvent, PayEvent>() {
            @Override
            public PayEvent timeout(Map<String, List<PayEvent>> map, long l) throws Exception {
                return map.get("begin").get(0);
            }
        }, new PatternSelectFunction<PayEvent, PayEvent>() {
            @Override
            public PayEvent select(Map<String, List<PayEvent>> map) throws Exception {
                return map.get("pay").get(0);
            }
        });

        //result.print();
        DataStream<PayEvent> sideOutput = result.getSideOutput(orderTimeoutOutput);
        sideOutput.print();

运行结果：

二、连续登录失败

需求：找出那些 5 秒钟内连续登录失败的账号，然后禁止用户再次尝试登录需要等待 1 分钟。创建 LoginCEPMain.java 类并实现前提代码：

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<LogInEvent> source = env.fromElements(
                new LogInEvent(1L, "fail", 1597905234000L),
                new LogInEvent(1L, "success", 1597905235000L),
                new LogInEvent(2L, "fail", 1597905236000L),
                new LogInEvent(2L, "fail", 1597905237000L),
                new LogInEvent(2L, "fail", 1597905238000L),
                new LogInEvent(3L, "fail", 1597905239000L),
                new LogInEvent(3L, "success", 1597905240000L)
        ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LogInEvent>(Time.milliseconds(500L)) {
            @Override
            public long extractTimestamp(LogInEvent logInEvent) {
                return logInEvent.getTimestamp();
            }
        }).keyBy(new KeySelector<LogInEvent, Object>() {
            @Override
            public Object getKey(LogInEvent value) throws Exception {
                return value.getuId();
            }
        });

        // 关键逻辑代码

        env.execute("execute cep");

关键逻辑代码如下：

        Pattern pattern = Pattern.<LogInEvent>begin("start").where(new IterativeCondition<LogInEvent>() {

            @Override
            public boolean filter(LogInEvent logInEvent, Context<LogInEvent> context) throws Exception {
                return logInEvent.getAction().equals("fail");
            }

        }).next("next").where(new IterativeCondition<LogInEvent>() {

            @Override
            public boolean filter(LogInEvent logInEvent, Context<LogInEvent> context) throws Exception {
                return logInEvent.getAction().equals("fail");
            }

        }).within(Time.seconds(5));


        PatternStream<LogInEvent> patternStream = CEP.pattern(source, pattern);
        SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<LogInEvent, AlertEvent>() {
            @Override
            public void processMatch(Map<String, List<LogInEvent>> match, Context ctx, Collector<AlertEvent> out) throws Exception {
                List<LogInEvent> start = match.get("start");
                List<LogInEvent> next = match.get("next");
                System.out.println("start:" + start + ",next:" + next);

                out.collect(new AlertEvent(String.valueOf(start.get(0).getuId()), "Login fail ~ "));
            }
        });

        process.print();

运行结果：

三、交易活跃用户

需求：找出那些 24 小时内至少 5 次有效交易的账户。创建 ActiveCEPMain.java 类并实现前提代码：

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<TransactionEvent> source = env.fromElements(
                new TransactionEvent("100XX", 0.0D, 1597905234000L),
                new TransactionEvent("100XX", 100.0D, 1597905235000L),
                new TransactionEvent("100XX", 200.0D, 1597905236000L),
                new TransactionEvent("100XX", 300.0D, 1597905237000L),
                new TransactionEvent("100XX", 400.0D, 1597905238000L),
                new TransactionEvent("100XX", 500.0D, 1597905239000L),
                new TransactionEvent("101XX", 0.0D, 1597905240000L),
                new TransactionEvent("101XX", 100.0D, 1597905241000L)
        ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<TransactionEvent>(Time.milliseconds(500L)) {
            @Override
            public long extractTimestamp(TransactionEvent transactionEvent) {
                return transactionEvent.getTimestamp();
            }
        }).keyBy(new KeySelector<TransactionEvent, Object>() {
            @Override
            public Object getKey(TransactionEvent value) throws Exception {
                return value.getAccountId();
            }
        });

        // 关键逻辑处理代码    

        env.execute("execute cep");

关键逻辑处理代码：

        Pattern pattern = Pattern.<TransactionEvent>begin("start").where(
                new SimpleCondition<TransactionEvent>() {
                    @Override
                    public boolean filter(TransactionEvent transactionEvent) {
                        return transactionEvent.getAmount() > 0;
                    }
                }
        ).timesOrMore(5)
         .within(Time.hours(24));

        PatternStream<TransactionEvent> patternStream = CEP.pattern(source, pattern);
        SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<TransactionEvent, AlertEvent>() {

            @Override
            public void processMatch(Map<String, List<TransactionEvent>> map, Context context, Collector<AlertEvent> collector) throws Exception {
                List<TransactionEvent> start = map.get("start");
                List<TransactionEvent> next = map.get("next");
                System.out.println("start:" + start + ",next:" + next);
                collector.collect(new AlertEvent(start.get(0).getAccountId(), "lian xu 有效交易！"));
            }

        });

        process.print();

运行结果：