Flink使用connect实现双流join全外连接

Posted 2021-06-08 Rango

一、背景说明

在Flink中可以使用Window join或者Interval Join实现双流join，不过使用join只能实现内连接，如果要实现左右连接或者外连接，则可以通过connect算子来实现。现有订单数据及支付数据如下方说明，基于数据时间实现订单及支付数据的关联，超时或者缺失则由侧输出流输出

//OrderLog.csv 订单数据，首列为订单id，付款成功则类型为pay（第二列），且生成支付id（第三列），最后列为时间
34729,create,,1558430842
34730,create,,1558430843
34729,pay,sd76f87d6,1558430844
34730,pay,3hu3k2432,1558430845
34731,pay,35jue34we,1558430849
...

//ReceiptLog.csv 支付数据，付款成功后生成，对于支付id/支付渠道/时间
sd76f87d6,wechat,1558430847
3hu3k2432,alipay,1558430848
...

//输出结果，限定时间内到达数据实现关联以Tuple2输出，否则侧输出流输出
(OrderEvent(orderId=34729, eventType=pay, txId=sd76f87d6, eventTime=1558430844),TxEvent(txId=sd76f87d6, payChannel=wechat, eventTime=1558430847))
(OrderEvent(orderId=34730, eventType=pay, txId=3hu3k2432, eventTime=1558430845),TxEvent(txId=3hu3k2432, payChannel=alipay, eventTime=1558430848))
No Receipt> 35jue34we 只有下单没有到账数据
...

二、实现过程

1. connect算子简单说明
   
   作用：两个不同来源的数据流进行连接，实现数据匹配。可以连接两个保持他们类型的数据流，两个数据流被connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

返回：DataStream[A], DataStream[B] -> ConnectedStreams[A,B]

//示例：
DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource<String> stringStream = env.fromElements("a", "b", "c");
// 把两个流连接在一起: 貌合神离
ConnectedStreams<Integer, String> cs = intStream.connect(stringStream);
cs.getFirstInput().print("first");
cs.getSecondInput().print("second");
env.execute();

2. 实现思路说明

①建立执行环境。

②读取数据映射到JavaBean，并指定WaterMark时间语义。
由于后续需要用到定时器来实现侧输出流，因此需要指定数据时间语义

③分组并使用connect关联两数据流。（关键步骤）

1.按支付id进行keyby，使相同id数据进入相同并行度处理。
2.由于使用到定时器，故使用process算子（支持定时器）分别对两条流进行处理。
3.process中使用状态编程（ValueState），对先到数据存入状态，并建立定时器，另外一条流在限定时间内到达则输出并删除定时器，否则触发定时器走侧输出流输出。在定时器中，对未到数据均输出到侧输出流则实现全外连接，对A到了B未到输出及A未到B到了不做输出则实现左连接，反之则是右连接效果。
ps：在connect中，无非对两条流进行单独的处理，在A流中处理B流未到时如何输出，在B流中处理A流未到时如何输入，建立定时器，能实现另一条流延迟到达的处理，如左连接、右连接或全外连接，如使用join算子，则只能实现内连接。

④打印并执行。

三、完整代码

package com.test.ordermatch;

import bean.OrderEvent;
import bean.TxEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author: Rango
 * @create: 2021-06-08 11:03
 * @description: 订单支付实时监控，两个数据源的关联
 **/
public class TestOrderMatch {
    public static void main(String[] args) throws Exception {

        //1.建立环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        //2.读取数据，映射数据源到javabean,指定watermark时间语义
        WatermarkStrategy<OrderEvent> orderWMS = WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
            @Override
            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                return element.getEventTime() * 1000L;
            }});
        WatermarkStrategy<TxEvent> receiptWMS = WatermarkStrategy.<TxEvent>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<TxEvent>() {
                    @Override
                    public long extractTimestamp(TxEvent element, long recordTimestamp) {
                        return element.getEventTime() * 1000L;
                    }});

        //使用flatmap能实现筛选，map+filter代码量更多，实现类型为pay的数据
        SingleOutputStreamOperator<OrderEvent> OrderDS = env.readTextFile("input/OrderLog.csv")
                .flatMap(new FlatMapFunction<String, OrderEvent>() {
                    @Override
                    public void flatMap(String value, Collector<OrderEvent> out) throws Exception {
                        String[] split = value.split(",");
                        OrderEvent orderEvent = new OrderEvent(Long.parseLong(split[0]), split[1], split[2], Long.parseLong(split[3]));
                        if (orderEvent.getEventType().equals("pay")){
                            out.collect(orderEvent);
                        }}})
                .assignTimestampsAndWatermarks(orderWMS);

        SingleOutputStreamOperator<TxEvent> ReceiptDS = env.readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new TxEvent(split[0], split[1], Long.parseLong(split[2]));
                    }
        }).assignTimestampsAndWatermarks(receiptWMS);

        //3.数据处理
        //使用keyby，使同一id数据进入同一并行度，以Tuple2输出
        SingleOutputStreamOperator<Tuple2<OrderEvent, TxEvent>> processDS = OrderDS.connect(ReceiptDS)
                .keyBy("txId", "txId")
                .process(new orderReceiptKeyProcessFunc());

        //4.打印执行
        processDS.print();
        processDS.getSideOutput(new OutputTag<String>("PayWithoutReceipt"){}).print("No Receipt");
        processDS.getSideOutput(new OutputTag<String>("ReceiptWithoutPay"){}).print("No Order");
        env.execute();
    }

    public static class orderReceiptKeyProcessFunc extends KeyedCoProcessFunction<String,OrderEvent,TxEvent, Tuple2<OrderEvent,TxEvent>>{

        private ValueState<OrderEvent> orderState;
        private ValueState<TxEvent> receiptState;
        private ValueState<Long> timeState;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("order-state",OrderEvent.class));
            receiptState = getRuntimeContext().getState(new ValueStateDescriptor<TxEvent>("receipt-state",TxEvent.class));
            timeState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("time-state",Long.class));
        }

        @Override
        public void processElement1(OrderEvent value, Context ctx, Collector<Tuple2<OrderEvent, TxEvent>> out) throws Exception {
            //在订单流中，判断支付流是否有数据，没有则启用定时器10秒后触发输出到侧输出流
            if (receiptState.value() == null){
                //支付数据未到，先把订单数据放入状态
                orderState.update(value);
                //建立定时器，10秒后触发
                Long ts = (value.getEventTime() + 10) * 1000L;
                ctx.timerService().registerEventTimeTimer(ts);
                timeState.update(ts);
            }else{
                //支付数据已到，直接输出到主流
                out.collect(new Tuple2<>(value,receiptState.value()));
                //删除定时器，如支付流先到，支付流建立了的定时器，在这里删除
                ctx.timerService().deleteEventTimeTimer(timeState.value());
                //清空状态，注意清空的是支付状态
                receiptState.clear();
                timeState.clear();
            }}
        @Override
        public void processElement2(TxEvent value, Context ctx, Collector<Tuple2<OrderEvent, TxEvent>> out) throws Exception {
            //在支付流中，判断订单流是否有数据，没有则启用定时器5秒后触发输出到侧输出流
            if (orderState.value() == null){
                //订单数据未到，先把支付数据放入状态
                receiptState.update(value);
                //建立定时器，5秒后再关联
                Long ts = (value.getEventTime() + 5) * 1000L;
                ctx.timerService().registerEventTimeTimer(ts);
                timeState.update(ts);
            }else{
                //订单数据已到，直接输出到主流
                out.collect(new Tuple2<>(orderState.value(),value));
                //删除定时器，如支付流先到，支付流建立了的定时器，在这里删除
                ctx.timerService().deleteEventTimeTimer(timeState.value());
                //清空状态，注意清空的是订单状态
                orderState.clear();
                timeState.clear();
            }}
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, TxEvent>> out) throws Exception {
            //这样为全外连接输出，删除else则是实现左连接，右连接则只输出else部分即可
            if (orderState.value() != null){
                ctx.output(new OutputTag<String>("PayWithoutReceipt") {}
                        , orderState.value().getTxId() + " 只有下单没有到账数据");
            }else{
                ctx.output(new OutputTag<String>("ReceiptWithoutPay") {}
                        , receiptState.value().getTxId() + " 只有到账无下单数据");
            }
            orderState.clear();
            receiptState.clear();
            timeState.clear();
        }
    }
}

---

学习交流，有任何问题还请随时评论指出交流。