平时我们都是用过电商平台购买商品，当我们购买某个商品之后会有提示购买成功或者失败，那么这玩意在系统后台是如何处理订单的实时对账呢？？？接下来我们将使用两种方式 ( 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，状态，测流，流合并 等；经过这个需求的实现，我相信同学们对以上的知识点有了进一步的理解了。感谢阅读，欢迎留言吐槽，共同进步，谢谢。
网友评论