java flink（二十七）实战之电商订单超时取消报警计算 CEP与ProcessFunction 对比

谁借莪１个温暖的怀抱￠ 2022-10-14 11:59 157阅读 0赞

**题目：根据订单文件，读取订单状态：创建-支付-取消，如果15分钟未支付，输出该订单报警。**

**1、首先利用CEP实现**

目录结构：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

文件内容：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1][]

文件内容转换成pojo包装类：

package Beans;
    
    public class OrderEvent {
        private Long orderId;
        private String eventType;
        private String txId;
        private Long timestamp;
    
        public OrderEvent() {
        }
    
        public OrderEvent(Long orderId, String eventType, String txId, Long timestamp) {
            this.orderId = orderId;
            this.eventType = eventType;
            this.txId = txId;
            this.timestamp = timestamp;
        }
    
        public Long getOrderId() {
            return orderId;
        }
    
        public void setOrderId(Long orderId) {
            this.orderId = orderId;
        }
    
        public String getEventType() {
            return eventType;
        }
    
        public void setEventType(String eventType) {
            this.eventType = eventType;
        }
    
        public String getTxId() {
            return txId;
        }
    
        public void setTxId(String txId) {
            this.txId = txId;
        }
    
        public Long getTimestamp() {
            return timestamp;
        }
    
        public void setTimestamp(Long timestamp) {
            this.timestamp = timestamp;
        }
    
        @Override
        public String toString() {
            return "OrderEvent{" +
                    "orderId=" + orderId +
                    ", eventType='" + eventType + '\'' +
                    ", txId='" + txId + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

报警信息包装类：

package Beans;
    
    public class OrderResult {
        private Long orderId;
        private String resultState;
    
        public OrderResult() {
        }
    
        public OrderResult(Long orderId, String resultState) {
            this.orderId = orderId;
            this.resultState = resultState;
        }
    
        public Long getOrderId() {
            return orderId;
        }
    
        public void setOrderId(Long orderId) {
            this.orderId = orderId;
        }
    
        public String getResultState() {
            return resultState;
        }
    
        public void setResultState(String resultState) {
            this.resultState = resultState;
        }
    
        @Override
        public String toString() {
            return "OrderResult{" +
                    "orderId=" + orderId +
                    ", resultState='" + resultState + '\'' +
                    '}';
        }
    }

CEP实现：

package Project;
    
    import Beans.OrderEvent;
    import Beans.OrderResult;
    import akka.stream.impl.QueueSink;
    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.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.OutputTag;
    
    import java.net.URL;
    import java.util.List;
    import java.util.Map;
    
    
    public class OrderpayTimeout {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
            //读取数据
            URL resource = OrderpayTimeout.class.getResource("/OrderLog.csv");
            DataStream<OrderEvent> orderEventStream = env.readTextFile(resource.getPath())
                    .map(line -> {
                        String[] fields = line.split(",");
                        return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                    })
                    .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
                        @Override
                        public long extractAscendingTimestamp(OrderEvent orderEvent) {
                            return orderEvent.getTimestamp()*1000L;
                        }
                    });
            //定义一个带时间限制的模式
            Pattern<OrderEvent, OrderEvent> orderPayPattern = Pattern.<OrderEvent>begin("create").where(new SimpleCondition<OrderEvent>() {
                @Override
                public boolean filter(OrderEvent orderEvent) throws Exception {
                    return "create".equals(orderEvent.getEventType());
                }
            }).followedBy("pay").where(new SimpleCondition<OrderEvent>() { //不用紧邻
                @Override
                public boolean filter(OrderEvent orderEvent) throws Exception {
                    return "pay".equals(orderEvent.getEventType());
                }
            }).within(Time.minutes(15));
            //定义侧输出流标签 用来表示超时时间
            OutputTag<OrderResult> orderResultOutputTag = new OutputTag<OrderResult>("order-timeout"){};
            //将pattern应用到输入输出流 得到patternStream
            PatternStream<OrderEvent> patternStream = CEP.pattern(orderEventStream.keyBy(OrderEvent::getOrderId), orderPayPattern);
            //调用select实现对匹配复杂事件和超时时间的提取和处理
            SingleOutputStreamOperator<OrderResult> resultStream = patternStream.select(orderResultOutputTag, new OrderTimeoutSelect(), new OrderPaySelect());
    
            resultStream.print("payed normally");
            resultStream.getSideOutput(orderResultOutputTag).print("payed timeout");
            env.execute("order timeout detect job");
        }
    
        //实现自定义的超时事件处理函数
        public static class  OrderTimeoutSelect implements PatternTimeoutFunction<OrderEvent, OrderResult>{
            @Override
            public OrderResult timeout(Map<String, List<OrderEvent>> map, long l) throws Exception {
                Long timeoutOrderId = map.get("create").get(0).getOrderId();
                return new OrderResult(timeoutOrderId,"timeout"+l);
            }
        }
        //实现自定义的正常匹配事件处理函数
        public static class OrderPaySelect implements PatternSelectFunction<OrderEvent, OrderResult>{
            @Override
            public OrderResult select(Map<String, List<OrderEvent>> map) throws Exception {
                Long payedOrderId = map.get("pay").get(0).getOrderId();
                return new OrderResult(payedOrderId,"payed success");
            }
        }
    }

ProcessFunction实现：

package Project;
    
    import Beans.OrderEvent;
    import Beans.OrderResult;
    import org.apache.flink.api.common.state.ValueState;
    import org.apache.flink.api.common.state.ValueStateDescriptor;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
    import org.apache.flink.table.runtime.operators.window.TimeWindow;
    import org.apache.flink.util.Collector;
    import org.apache.flink.util.OutputTag;
    
    import java.net.URL;
    
    public class OrderTimeoutWithoutCep {
        //定义超时事件侧输出流标签
        private final static OutputTag<OrderResult> orderTimeoutTag = new OutputTag<OrderResult>("order-timeout") {
        };
    
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
            //读取数据
            URL resource = OrderTimeoutWithoutCep.class.getResource("/OrderLog.csv");
            DataStream<OrderEvent> orderEventStream = env.readTextFile(resource.getPath())
                    .map(line -> {
                        String[] fields = line.split(",");
                        return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                    })
                    .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
                        @Override
                        public long extractAscendingTimestamp(OrderEvent orderEvent) {
                            return orderEvent.getTimestamp() * 1000L;
                        }
                    });
            //自定义处理函数 主流输出正常匹配订单 侧输出流输出超时报警
            SingleOutputStreamOperator<OrderResult> resultStream = orderEventStream.keyBy(OrderEvent::getOrderId)
                    .process(new OrderPayMatchDetect());
            resultStream.print("payed normally");
            resultStream.getSideOutput(orderTimeoutTag).print("timeout order");
            env.execute("order timeout detect without cep job");
        }
    
        //实现自定义KeyedProcessFunction
        public static class OrderPayMatchDetect extends KeyedProcessFunction<Long, OrderEvent, OrderResult> {
            //定义状态 保存之前订单是否已经create pay
            ValueState<Boolean> isPayedState;
            ValueState<Boolean> isCreatedState;
            //保存定时器时间戳
            ValueState<Long> timerTsState;
    
            //注册状态
            @Override
            public void open(Configuration parameters) throws Exception {
                isPayedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-payed", Boolean.class, false));
                isCreatedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("is-created", Boolean.class, false));
                timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));
    
            }
    
            @Override
            public void processElement(OrderEvent orderEvent, Context context, Collector<OrderResult> collector) throws Exception {
                //取状态
                Boolean isPayed = isPayedState.value();
                Boolean isCreated = isCreatedState.value();
                Long timerTs = timerTsState.value();
    
                //判断当前时间类型
                if ("create".equals(orderEvent.getEventType())) {
                    //如果是create 判断是否支付过
                    if (isPayed) {
                        //如果已经正常支付 输出正常匹配结果
                        collector.collect(new OrderResult(orderEvent.getOrderId(), "payed successfully"));
                        //清空状态
                        isCreatedState.clear();
                        isPayedState.clear();
                        timerTsState.clear();
                        context.timerService().deleteEventTimeTimer(timerTs);
                    } else {
                        //如果没有支付过 注册15分钟之后的定时器 开始等待支付
                        Long ts = (orderEvent.getTimestamp() + 15 * 60) * 1000L;
                        context.timerService().registerEventTimeTimer(ts);
                        //更新状态
                        timerTsState.update(ts);
                        isCreatedState.update(true);
                    }
                } else if ("pay".equals(orderEvent.getEventType())) {
                    if (isCreated) {
                        if (orderEvent.getTimestamp() * 1000L < timerTs) {
                            collector.collect(new OrderResult(orderEvent.getOrderId(),
                                    "payed successfully"));
                        } else {
                            context.output(orderTimeoutTag, new
                                    OrderResult(orderEvent.getOrderId(), "payed but already timeout"));
                        }
                        isCreatedState.clear();
                        timerTsState.clear();
                        context.timerService().deleteEventTimeTimer(timerTs);
                    } else {
                        context.timerService().registerEventTimeTimer(orderEvent.getTimestamp() * 1000L);
                        isPayedState.update(true);
                        timerTsState.update(orderEvent.getTimestamp() * 1000L);
                    }
                }
            }
    
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx,
                                Collector<OrderResult> out) throws Exception {
                if (isPayedState.value()) {
                    ctx.output(orderTimeoutTag,
                            new OrderResult(ctx.getCurrentKey(), "already payed but not found created log"));
                } else {
                    ctx.output(orderTimeoutTag,
                            new OrderResult(ctx.getCurrentKey(), "order pay timeout"));
                }
                isPayedState.clear();
                isCreatedState.clear();
                timerTsState.clear();
            }
    
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221014/90d1ebe0ac2048109c445dcd8fa3ce58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1]: /images/20221014/2b8105b1688544dfb6cd7634f62c121d.png