??平時(shí)我們都是用過電商平臺(tái)購買商品，當(dāng)我們購買某個(gè)商品之后會(huì)有提示購買成功或者失敗，那么這玩意在系統(tǒng)后臺(tái)是如何處理訂單的實(shí)時(shí)對(duì)賬呢？？？接下來我們將使用兩種方式 ( table api 和 process function) 進(jìn)行這個(gè)對(duì)賬的分析。
??在實(shí)現(xiàn)代碼之前我們先看下流數(shù)據(jù)的格式：
訂單事件數(shù)據(jù) ：
34729,create,,1558430842
34730,create,,1558430843
34729,pay,sd76f87d6,1558430844
34730,pay,3hu3k2432,1558430845
34731,create,,1558430846
34731,pay,35jue34we,1558430849
到賬事件數(shù)據(jù) ：
ewr342as4,wechat,1558430845
sd76f87d6,wechat,1558430847
3hu3k2432,alipay,1558430848
8fdsfae83,alipay,1558430850
32h3h4b4t,wechat,1558430852
766lk5nk4,wechat,1558430855
??從數(shù)據(jù)格式我們可以知道：訂單事件數(shù)據(jù) -> 用戶ID，訂單狀態(tài)，訂單ID，時(shí)間戳；到賬事件數(shù)據(jù) -> 訂單ID，支付平臺(tái)類型，時(shí)間戳
??鑒于以上數(shù)據(jù)格式類型我們將可以映射成如下兩個(gè)實(shí)體類：
// 訂單事件數(shù)據(jù)實(shí)體類
public class OrderEvent {
private Long userId;
private String action;
private String orId;
private Long timestamp;
......
}
// 到賬事件數(shù)據(jù)實(shí)體類
public class ReceiptEvent {
private String orId;
private String payEquipment;
private Long timestamp;
......
}
??好了，數(shù)據(jù)類型和格式我們都準(zhǔn)備好了，接下來我們將實(shí)現(xiàn)邏輯代碼去對(duì)賬。
一、TableAPI 實(shí)現(xiàn)雙流合并對(duì)賬
??這里為了方便我們的數(shù)據(jù)事先是放在excel里邊去的，生產(chǎn)環(huán)境一般都是解析 kafka 過來的 json 數(shù)據(jù)然后再對(duì)其進(jìn)行邏輯操作的哦。
1.創(chuàng)建關(guān)鍵代碼 PayJoinReceMain.java：
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.setParallelism(1);
// 1. 讀取訂單事件數(shù)據(jù)
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. 讀取到賬事件數(shù)據(jù)
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());
// -------------------------------關(guān)聯(lián)處理-------------------------------------------------
DataStream resultStream = orderDataStream.intervalJoin(receipDataStream) //這里使用相對(duì)關(guān)聯(lián)
.between(Time.seconds(-3), Time.seconds(5)) // 訂單數(shù)據(jù)等待到賬數(shù)據(jù)時(shí)間前三秒到后三秒?yún)^(qū)間
.process(new OrderMatchWithJoinFunction()); // 自定義類輸出服務(wù)上邊條件的數(shù)據(jù)
// ---------------------------------------------------------------------------------------
resultStream.print();
env.execute("tx match with join job");
}
- 實(shí)現(xiàn)下自定義類 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));
}
}
-
運(yùn)行結(jié)果如下：image.png
??從整體來看，這個(gè)代碼很簡單，但是也有缺點(diǎn)：1. 由于是相對(duì)關(guān)聯(lián)，因此匹配度不是很高；2. TableAPI 只能實(shí)現(xiàn)符合需求的數(shù)據(jù)輸出，不能輸出不符合的數(shù)據(jù)。
??為了避免以上的缺陷，我們接下來使用 process function 來實(shí)現(xiàn)對(duì)賬功能。
二、process function 方式的實(shí)現(xiàn)
1.改造虛線部分的代碼：
//合并兩條流，進(jìn)行處理
SingleOutputStreamOperator resultStream = resultStream = orderDataStream.connect(receipDataStream)
.process(new OrderMatchFunction());
resultStream.print("matched");
resultStream.getSideOutput(unmatchedPayEventOutputTag).print("unmatched pays");
resultStream.getSideOutput(unmatchedReceiptEventOutputTag).print("unmatched receipts");
??由于要測(cè)輸出不符合的數(shù)據(jù)，因此我們需要在 main 方法前邊實(shí)例化 OutputTag :
private static final OutputTag unmatchedPayEventOutputTag = new OutputTag<OrderEvent>("unmatched-pay"){};
private static final OutputTag unmatchedReceiptEventOutputTag = new OutputTag<ReceiptEvent>("unmatched-receipt"){};
- 我們繼承 CoProcessFunction 去創(chuàng)建 OrderMatchFunction ，整體代碼如下:
public static class OrderMatchFunction extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>{
// 定義狀態(tài)，保存當(dāng)前交易對(duì)應(yīng)的訂單支付事件和到賬事件
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 ){
// 如果已經(jīng)有receipt，正常輸出匹配，清空狀態(tài)
collector.collect(new Tuple2(orderEvent, receipt));
receiptEventState.clear();
payEventState.clear();
} else{
// 如果還沒來，注冊(cè)定時(shí)器開始等待5秒
context.timerService().registerEventTimeTimer(orderEvent.getTimestamp() * 1000L + 5000L);
// 更新狀態(tài)
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 ){
// 如果已經(jīng)有pay，正常輸出匹配，清空狀態(tài)
collector.collect(new Tuple2(pay, receiptEvent));
receiptEventState.clear();
payEventState.clear();
} else{
// 如果還沒來，注冊(cè)定時(shí)器開始等待3秒
context.timerService().registerEventTimeTimer(receiptEvent.getTimestamp() * 1000L + 3000L);
// 更新狀態(tài)
receiptEventState.update(receiptEvent);
}
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
super.onTimer(timestamp, ctx, out);
// 定時(shí)器觸發(fā)，判斷狀態(tài)中哪個(gè)還存在，就代表另一個(gè)沒來，輸出到側(cè)輸出流
if( payEventState.value() != null ){
ctx.output(unmatchedPayEventOutputTag, payEventState.value());
}
if( receiptEventState.value() != null ){
ctx.output(unmatchedReceiptEventOutputTag, receiptEventState.value());
}
// 清空狀態(tài)
receiptEventState.clear();
payEventState.clear();
}
}
-
運(yùn)行結(jié)果如下：image.png
??到目前為止，我們用了兩種方式實(shí)現(xiàn)多流對(duì)賬功能，整體來看也是挺簡單的，主要用到的知識(shí)點(diǎn)是 Watermark，狀態(tài)，測(cè)流，流合并 等；經(jīng)過這個(gè)需求的實(shí)現(xiàn)，我相信同學(xué)們對(duì)以上的知識(shí)點(diǎn)有了進(jìn)一步的理解了。感謝閱讀，歡迎留言吐槽，共同進(jìn)步，謝謝。
