Flink CEP

package com.cnic.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

//数据流：web/app -> Nginx -> 日志服务器(log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> Kafka(DWD)
//程  序：  Mock -> f1.sh -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> DwdTrafficUserJumpDetail -> Kafka(ZK)
public class DwdTrafficUserJumpDetail {

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

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生成环境设置为Kafka主题的分区数

//        env.setStateBackend(new HashMapStateBackend());
//        env.enableCheckpointing(5000L);
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs:xxx:8020//xxx/xx");

        //TODO 2.读取 Kafka dwd_traffic_page_log 主题数据创建流
        String sourceTopic = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_user_jump_detail_211027";
        DataStreamSource kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //TODO 3.将数据转换为JSON对象
        SingleOutputStreamOperator jsonObjDS = kafkaDS.map(JSON::parseObject);

        //TODO 4.提取事件时间生成WaterMark
        SingleOutputStreamOperator jsonObjWithWmDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        }));

        //TODO 5.按照Mid进行分组
        KeyedStream keyedByMidStream = jsonObjWithWmDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //TODO 6.定义模式序列
        Pattern pattern = Pattern.begin("first").where(new SimpleCondition() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                return value.getJSONObject("page").getString("last_page_id") == null;
            }
        }).next("second").where(new SimpleCondition() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                return value.getJSONObject("page").getString("last_page_id") == null;
            }
        }).within(Time.seconds(10));

//        Pattern pattern = Pattern.begin("first").where(new SimpleCondition() {
//            @Override
//            public boolean filter(JSONObject value) throws Exception {
//                return value.getJSONObject("page").getString("last_page_id") == null;
//            }
//        })
//                .times(2)      //默认为宽松近邻
//                .consecutive() //指定为严格近邻
//                .within(Time.seconds(10));

        //TODO 7.将模式序列作用到流上
        PatternStream patternStream = CEP.pattern(keyedByMidStream, pattern);

        //TODO 8.提取匹配上的事件以及超时事件
        OutputTag timeOutTag = new OutputTag("time-out") {
        };
        SingleOutputStreamOperator selectDS = patternStream.select(timeOutTag,
                new PatternTimeoutFunction() {
                    @Override
                    public JSONObject timeout(Map> map, long l) throws Exception {
                        return map.get("first").get(0);
                    }
                }, new PatternSelectFunction() {
                    @Override
                    public JSONObject select(Map> map) throws Exception {
                        return map.get("first").get(0);
                    }
                });
        DataStream timeOutDS = selectDS.getSideOutput(timeOutTag);

        //TODO 9.合并两个事件流
        selectDS.print("Select>>>>>>>>>");
        timeOutDS.print("TimeOut>>>>>>>>>");
        DataStream unionDS = selectDS.union(timeOutDS);

        //TODO 10.将数据写出到Kafka
        String targetTopic = "dwd_traffic_user_jump_detail";
        unionDS.map(JSONAware::toJSONString)
                .addSink(MyKafkaUtil.getKafkaProducer(targetTopic));

        //TODO 11.启动任务
        env.execute("DwdTrafficUserJumpDetail");

    }

}