flink实践-电商用户行为数据分析-第6章-订单支付实时监控

在电商网站中，订单的支付作为直接与营销收入挂钩的一环，在业务流程中非常重要。对于订单而言，为了正确控制业务流程，也为了增加用户的支付意愿，网站一般会设置一个支付失效时间，超过一段时间不支付的订单就会被取消。另外，对于订单的支付，我们还应保证用户支付的正确性，这可以通过第三方支付平台的交易数据来做一个实时对账。在接下来的内容中，我们将实现这两个需求。

1、模块创建和数据准备

同样地，在 UserBehaviorAnalysis 下新建一个 maven module 作为子项目，命名为

OrderTimeoutDetect。在这个子模块中，我们同样将会用到 flink 的 CEP 库来实现事件流的模式匹配，所以需要

在 pom 文件中引入 CEP 的相关依赖：

<dependency>
 <groupId>org.apache.flink</groupId> 
 <artifactId>flink-cep _${scala.binary.version}</artifactId> 
 <version>${flink.version}</version>
</dependency>

2、代码实现

在电商平台中，最终创造收入和利润的是用户下单购买的环节；更具体一点，是用户真正完成支付动作的时

候。用户下单的行为可以表明用户对商品的需求，但在现实中，并不是每次下单都会被用户立刻支付。当拖延一段

时间后，用户支付的意愿会降低。所以为了让用户更有紧迫感从而提高支付转化率，同时也为了防范订单支付环节

的安全风险，电商网站往往会对订单状态进行监控，设置一个失效时间（比如 15 分钟），如果下单后一段时间仍

未支付，订单就会被取消。

2.1 使用 CEP 实现

我们首先还是利用 CEP 库来实现这个功能。我们先将事件流按照订单号 orderId分流，然后定义这样的一个

事件模式：在 15 分钟内，事件“create”与“pay”非严格紧邻：

Pattern<OrderEvent, OrderEvent> orderPayPattern =
    Pattern.<OrderEvent>begin("create").where(new SimpleCondition<OrderEvent>() {
        @Override
        public boolean filter(OrderEvent value) throws Exception {
            return "create".equals(value.getEventType());
        }
    })
    .followedBy("pay")
    .where(new SimpleCondition<OrderEvent>() {
        @Override
        public boolean filter(OrderEvent value) throws Exception {
            return "pay".equals(value.getEventType());
        }
    })
    .within(Time.minutes(15));

这样调用.select 方法时，就可以同时获取到匹配出的事件和超时未匹配的事件了。在 src/main/java 下继续

创建 OrderTimeout 类，定义 POJO 类 OrderEvent，这是输入的订单事件流；另外还有 OrderResult，这是输出

显示的订单状态结果。订单数据也本应该从 UserBehavior 日志里提取，由于 UserBehavior.csv 中没有做相关埋

点，我们从另一个文件 OrderLog.csv 中读取登录数据。

完整代码如下：

OrderTimeoutDetect/src/main/java/OrderTimeout.java

public class OrderTimeout {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        // 读取数据
        URL resource = OrderTimeout.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 element) {
                return element.getTimestamp() * 1000L;
            }
        });
        // 定义一个带匹配时间窗口的模式
        Pattern<OrderEvent, OrderEvent> orderPayPattern = Pattern.<OrderEvent>begin("create")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return "create".equals(value.getEventType());
                    }
                }).followedBy("pay").where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return "pay".equals(value.getEventType());
                    }
                }).within(Time.minutes(15));
        PatternStream<OrderEvent> patternStream = CEP.pattern(orderEventStream.keyBy(OrderEvent::getOrderId),
                orderPayPattern);
        // 定义一个输出标签
        OutputTag<OrderResult> orderTimeoutOutputTag = new OutputTag<OrderResult>("orderTimeout") {
        };
        SingleOutputStreamOperator<OrderResult> resultStream = patternStream.select(orderTimeoutOutputTag,
                new OrderTimeoutSelect(), new OrderPaySelect());
        resultStream.print("payed");
        resultStream.getSideOutput(orderTimeoutOutputTag).print("timeout");
        env.execute("Order Timeout Detect Job");
    }

    // 自定义超时拣选函数
    public static class OrderTimeoutSelect implements PatternTimeoutFunction<OrderEvent, OrderResult> {
        @Override
        public OrderResult timeout(Map<String, List<OrderEvent>> pattern, long timeoutTimestamp) throws Exception {
            Long timeoutOrderId = pattern.get("create").iterator().next().getOrderId();
            return new OrderResult(timeoutOrderId, "timeout");
        }
    }

    // 自定义匹配拣选函数
    public static class OrderPaySelect implements PatternSelectFunction<OrderEvent, OrderResult> {
        @Override
        public OrderResult select(Map<String, List<OrderEvent>> pattern) throws Exception {
            Long payedOrderId = pattern.get("pay").iterator().next().getOrderId();
            return new OrderResult(payedOrderId, "payed");
        }
    }
}

2.2 使用 Process Function 实现

我们同样可以利用 Process Function，自定义实现检测订单超时的功能。为了简化问题，我们只考虑超时报

警的情形，在 pay 事件超时未发生的情况下，输出超时报警信息。

一个简单的思路是，可以在订单的 create 事件到来后注册定时器，15 分钟后触发；然后再用一个布尔类型的

Value 状态来作为标识位，表明 pay 事件是否发生过。如果 pay 事件已经发生，状态被置为 true，那么就不再需

要做什么操作；而如果 pay 事件一直没来，状态一直为 false，到定时器触发时，就应该输出超时报警信息。

具体代码实现如下：

OrderTimeoutDetect/src/main/java/OrderTimeoutWithoutCep.java

public class OrderTimeoutWithoutCep { 

    private final static OutputTag<OrderResult> orderTimeoutOutputTag = new  OutputTag<OrderResult>("orderTimeout") {}; 

    public static void main(String[] args) throws Exception{ 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
        env.setParallelism(1); 
        // 读取数据 
        URL resource = OrderTimeout.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 element) {
                    return element.getTimestamp()  1000L;
                }
            });
        // 自定义处理函数
        SingleOutputStreamOperator<OrderResult> resultStream = orderEventStream.keyBy(OrderEvent::getOrderId).process(new OrderPayMatchDetect());
        resultStream.print("payed");
        resultStream.getSideOutput(orderTimeoutOutputTag).print("timeout");
        env.execute("Order Timeout Detect without CEP Job");
    }

    //实现自定义 KeyedProcessFunction
    public static class OrderPayMatchDetect extends KeyedProcessFunction<Long,OrderEvent, OrderResult>{ 
    
        // 定义状态
        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 value, Context ctx, Collector<OrderResult> out) throws Exception { 
            Boolean isPayed = isPayedState.value(); 
            Boolean isCreated = isCreatedState.value();
            Long timerTs = timerTsState.value();
            if( "create".equals(value.getEventType()) ){
                if( isPayed ){
                    out.collect( new OrderResult( value.getOrderId(), "payed successfully" ) );
                    isPayedState.clear();
                    timerTsState.clear();
                    ctx.timerService().deleteEventTimeTimer(timerTs); 
                }else {
                    Long ts = ( value.getTimestamp() + 15  60 )  1000L;
                    ctx.timerService().registerEventTimeTimer(ts); 
                    isCreatedState.update(true); 
                    timerTsState.update(ts); 
                }
            }else if( "pay".equals(value.getEventType()) ){
                if( isCreated ){
                    if( value.getTimestamp()  1000L < timerTs ){
                        out.collect( new OrderResult( value.getOrderId(),"payed successfully" )); 
                    }else{
                        ctx.output(orderTimeoutOutputTag, new OrderResult( value.getOrderId(), "payed but already timeout" )); 
                    } 
                    isCreatedState.clear(); 
                    timerTsState.clear(); 
                    ctx.timerService().deleteEventTimeTimer(timerTs);
                }else{
                    ctx.timerService().registerEventTimeTimer( value.getTimestamp()* 1000L ); 
                    isPayedState.update(true); 
                    timerTsState.update( value.getTimestamp()* 1000L ); 
                }
            } 
        } 

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx,Collector<OrderResult> out) throws Exception {
            if( isPayedState.value() ){
                ctx.output(orderTimeoutOutputTag,new OrderResult(ctx.getCurrentKey(), "already payed but not found created log")); 
            }else{
                ctx.output(orderTimeoutOutputTag,new OrderResult(ctx.getCurrentKey(), "order pay timeout")); 
            }
            isPayedState.clear();
            isCreatedState.clear();
            timerTsState.clear();
        }
    }
}

3、来自两条流的订单交易匹配

对于订单支付事件，用户支付完成其实并不算完，我们还得确认平台账户上是否到账了。而往往这会来自不

同的日志信息，所以我们要同时读入两条流的数据来做合并处理。这里我们利用 connect 将两条流进行

连接，然后用自定义的CoProcessFunction 进行处理。

具体代码如下：

TxMatchDetect/src/main/java/TxMatch.java

public class TxPayMatch {
    private final static OutputTag<OrderEvent> unmatchedPays = new OutputTag<OrderEvent>("unmatchedPays") {
    };
    private final static OutputTag<ReceiptEvent> unmatchedReceipts = new OutputTag<ReceiptEvent>("unmatchedReceipts") {
    };

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        URL orderRes = OrderTimeout.class.getResource("/OrderLog.csv");
        DataStream<OrderEvent> orderEventStream = env.readTextFile(orderRes.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 element) {
                return element.getTimestamp() * 1000L;
            }
        }).filter(data -> !"".equals(data.getTxId()));
        URL receiptRes = OrderTimeout.class.getResource("/ReceiptLog.csv");
        DataStream<ReceiptEvent> receiptEventStream = env.readTextFile(receiptRes.getPath()).map(line -> {
            String[] fields = line.split(",");
            return new ReceiptEvent(fields[0], fields[1], new Long(fields[2]));
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ReceiptEvent>() {
            @Override
            public long extractAscendingTimestamp(ReceiptEvent element) {
                return element.getTimestamp() * 1000L;
            }
        });
        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> resultStream = orderEventStream
                .keyBy(OrderEvent::getTxId).connect(receiptEventStream.keyBy(ReceiptEvent::getTxId))
                .process(new TxPayMatchDetect());
        resultStream.print("matched");
        resultStream.getSideOutput(unmatchedPays).print("unmatchedPays");
        resultStream.getSideOutput(unmatchedReceipts).print("unmatchedReceipts");
        env.execute("tx pay match job");
    }

    // 实现自定义的 CoProcessFunction
    public static class TxPayMatchDetect
            extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {
        ValueState<OrderEvent> payState;
        ValueState<ReceiptEvent> receiptState;

        @Override
        public void open(Configuration parameters) throws Exception {
            payState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("pay", OrderEvent.class));
            receiptState = getRuntimeContext()
                    .getState(new ValueStateDescriptor<ReceiptEvent>("receipt", ReceiptEvent.class));
        }

        @Override
        public void processElement1(OrderEvent pay, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out)
                throws Exception {
            ReceiptEvent receipt = receiptState.value();
            if (receipt != null) {
                out.collect(new Tuple2<>(pay, receipt));
                receiptState.clear();
            } else {
                payState.update(pay);
                ctx.timerService().registerEventTimeTimer(pay.getTimestamp() * 1000L + 5000L);
            }
        }

    @Override
 public void processElement2(ReceiptEvent receipt, Context ctx,
Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
 OrderEvent pay = payState.value();
 if (pay != null) {
 out.collect(new Tuple2<>(pay, receipt));
 payState.clear();
 } else {
 receiptState.update(receipt);
ctx.timerService().registerEventTimeTimer(receipt.getTimesta
mp() * 1000L + 3000L);
 }
 }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out)
                throws Exception {
            if (payState.value() != null) {
                ctx.output(unmatchedPays, payState.value());
            }
            if (receiptState.value() != null) {
                ctx.output(unmatchedReceipts, receiptState.value());
            }
            payState.clear();
            receiptState.clear();
        }
    }
}

原文链接：https://blog.csdn.net/q287573145/article/details/124523285