Flink系列 - 实时数仓之电商订单支付实时对账（六）

平时我们都是用过电商平台购买商品，当我们购买某个商品之后会有提示购买成功或者失败，那么这玩意在系统后台是如何处理订单的实时对账呢？？？接下来我们将使用两种方式 ( table api 和 process function) 进行这个对账的分析。
在实现代码之前我们先看下流数据的格式：

订单事件数据 ：
34729,create,,1558430842
34730,create,,1558430843
34729,pay,sd76f87d6,1558430844
34730,pay,3hu3k2432,1558430845
34731,create,,1558430846
34731,pay,35jue34we,1558430849

到账事件数据 ：
ewr342as4,wechat,1558430845
sd76f87d6,wechat,1558430847
3hu3k2432,alipay,1558430848
8fdsfae83,alipay,1558430850
32h3h4b4t,wechat,1558430852
766lk5nk4,wechat,1558430855

从数据格式我们可以知道：订单事件数据 -> 用户ID，订单状态，订单ID，时间戳；到账事件数据 -> 订单ID，支付平台类型，时间戳
鉴于以上数据格式类型我们将可以映射成如下两个实体类：

// 订单事件数据实体类
public class OrderEvent {
    private Long userId;
    private String action;
    private String orId;
    private Long timestamp;
    ......
}

// 到账事件数据实体类
public class ReceiptEvent {
    private String orId;
    private String payEquipment;
    private Long timestamp;
    ......
}

好了，数据类型和格式我们都准备好了，接下来我们将实现逻辑代码去对账。

一、TableAPI 实现双流合并对账

这里为了方便我们的数据事先是放在excel里边去的，生产环境一般都是解析 kafka 过来的 json 数据然后再对其进行逻辑操作的哦。
1.创建关键代码 PayJoinReceMain.java：

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

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

        // 1. 读取订单事件数据
        DataStream<String> inputOrderStream = env.readTextFile("C:\\Users\\Administrator\\Desktop\\my-gitlib\\shishi-daping\\dip\\shishi-daping\\NFDWSYYBigScreen\\TestJsonDmon\\src\\main\\resources\\OrderLog.csv");
        KeyedStream<OrderEvent,String> orderDataStream = inputOrderStream.map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String s) throws Exception {
                String[] dataArray = s.split(",");
                return new OrderEvent(Long.parseLong(dataArray[0]),dataArray[1],dataArray[2],Long.parseLong(dataArray[3]));
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<OrderEvent>(Time.seconds(1)) {
            @Override
            public long extractTimestamp(OrderEvent element) {
                return element.getTimestamp()*1000L;
            }
        }).filter(order -> order.getAction().equals("pay"))
                .keyBy(order -> order.getOrId());

        // 2. 读取到账事件数据
        DataStream<String> inputReceipStream = env.readTextFile("C:\\Users\\Administrator\\Desktop\\my-gitlib\\shishi-daping\\dip\\shishi-daping\\NFDWSYYBigScreen\\TestJsonDmon\\src\\main\\resources\\ReceiptLog.csv");
        KeyedStream<ReceiptEvent,String> receipDataStream = inputReceipStream.map(new MapFunction<String, ReceiptEvent>() {
            @Override
            public ReceiptEvent map(String s) throws Exception {
                String[] dataArray = s.split(",");
                return new ReceiptEvent(dataArray[0],dataArray[1],Long.parseLong(dataArray[2]));
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<ReceiptEvent>(Time.seconds(1)) {
            @Override
            public long extractTimestamp(ReceiptEvent element) {
                return element.getTimestamp()*1000L;
            }
        }).keyBy(order -> order.getOrId());

        // -------------------------------关联处理-------------------------------------------------
       
         DataStream resultStream = orderDataStream.intervalJoin(receipDataStream)  //这里使用相对关联
                .between(Time.seconds(-3), Time.seconds(5))  // 订单数据等待到账数据时间前三秒到后三秒区间
                .process(new OrderMatchWithJoinFunction());  // 自定义类输出服务上边条件的数据

        // ---------------------------------------------------------------------------------------  

        resultStream.print();
        env.execute("tx match with join job");
    }

实现下自定义类 OrderMatchWithJoinFunction.class ：

public static class OrderMatchWithJoinFunction extends ProcessJoinFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent,ReceiptEvent>> {

        @Override
        public void processElement(OrderEvent orderEvent, ReceiptEvent receiptEvent, Context context, Collector<Tuple2<OrderEvent, ReceiptEvent>> collector) throws Exception {
            collector.collect(new Tuple2<>(orderEvent, receiptEvent));
        }
    }

运行结果如下：

image.png

从整体来看，这个代码很简单，但是也有缺点：1. 由于是相对关联，因此匹配度不是很高；2. TableAPI 只能实现符合需求的数据输出，不能输出不符合的数据。
为了避免以上的缺陷，我们接下来使用 process function 来实现对账功能。

二、process function 方式的实现

1.改造虚线部分的代码：

        //合并两条流，进行处理
        SingleOutputStreamOperator resultStream = resultStream = orderDataStream.connect(receipDataStream)
                .process(new OrderMatchFunction());

        resultStream.print("matched");
        resultStream.getSideOutput(unmatchedPayEventOutputTag).print("unmatched pays");
        resultStream.getSideOutput(unmatchedReceiptEventOutputTag).print("unmatched receipts");

由于要测输出不符合的数据，因此我们需要在 main 方法前边实例化 OutputTag :

private static final OutputTag unmatchedPayEventOutputTag = new OutputTag<OrderEvent>("unmatched-pay"){};
private static final OutputTag unmatchedReceiptEventOutputTag = new OutputTag<ReceiptEvent>("unmatched-receipt"){};

我们继承 CoProcessFunction 去创建 OrderMatchFunction ，整体代码如下:

public static class OrderMatchFunction extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>{

        // 定义状态，保存当前交易对应的订单支付事件和到账事件
        transient ValueState<OrderEvent> payEventState = null;
        transient ValueState<ReceiptEvent> receiptEventState = null;

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

        @Override
        public void processElement1(OrderEvent orderEvent, Context context, Collector<Tuple2<OrderEvent, ReceiptEvent>> collector) throws Exception {
            // 订单支付来了，要判断之前是否有到账事件
            ReceiptEvent receipt = receiptEventState.value();
            if( receipt != null ){
                // 如果已经有receipt，正常输出匹配，清空状态
                collector.collect(new Tuple2(orderEvent, receipt));
                receiptEventState.clear();
                payEventState.clear();
            } else{
                // 如果还没来，注册定时器开始等待5秒
                context.timerService().registerEventTimeTimer(orderEvent.getTimestamp() * 1000L + 5000L);
                // 更新状态
                payEventState.update(orderEvent);
            }

        }

        @Override
        public void processElement2(ReceiptEvent receiptEvent, Context context, Collector<Tuple2<OrderEvent, ReceiptEvent>> collector) throws Exception {
            // 到账事件来了，要判断之前是否有pay事件
            OrderEvent pay = payEventState.value();
            if( pay != null ){
                // 如果已经有pay，正常输出匹配，清空状态
                collector.collect(new Tuple2(pay, receiptEvent));
                receiptEventState.clear();
                payEventState.clear();
            } else{
                // 如果还没来，注册定时器开始等待3秒
                context.timerService().registerEventTimeTimer(receiptEvent.getTimestamp() * 1000L + 3000L);
                // 更新状态
                receiptEventState.update(receiptEvent);
            }

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            super.onTimer(timestamp, ctx, out);

            // 定时器触发，判断状态中哪个还存在，就代表另一个没来，输出到侧输出流
            if( payEventState.value() != null ){
                ctx.output(unmatchedPayEventOutputTag, payEventState.value());
            }
            if( receiptEventState.value() != null ){
                ctx.output(unmatchedReceiptEventOutputTag, receiptEventState.value());
            }
            // 清空状态
            receiptEventState.clear();
            payEventState.clear();

        }

    }

运行结果如下：

image.png

到目前为止，我们用了两种方式实现多流对账功能，整体来看也是挺简单的，主要用到的知识点是 Watermark，状态，测流，流合并等；经过这个需求的实现，我相信同学们对以上的知识点有了进一步的理解了。感谢阅读，欢迎留言吐槽，共同进步，谢谢。