Flink 双流 Join 的3种操作示例

在数据库中的静态表上做 OLAP 分析时，两表 join 是非常常见的操作。同理，在流式处理作业中，有时也需要在两条流上做 join 以获得更丰富的信息。Flink DataStream API 为用户提供了3个算子来实现双流 join，分别是：

join()
coGroup()
intervalJoin()

本文举例说明它们的使用方法，顺便聊聊比较特殊的 interval join 的原理。

准备数据

从 Kafka 分别接入点击流和订单流，并转化为 POJO。

1DataStream<String> clickSourceStream = env
2  .addSource(new FlinkKafkaConsumer011<>(
3    "ods_analytics_access_log",
4    new SimpleStringSchema(),
5    kafkaProps
6  ).setStartFromLatest());
7DataStream<String> orderSourceStream = env
8  .addSource(new FlinkKafkaConsumer011<>(
9    "ods_ms_order_done",
10    new SimpleStringSchema(),
11    kafkaProps
12  ).setStartFromLatest());
13
14DataStream<AnalyticsAccessLogRecord> clickRecordStream = clickSourceStream
15  .map(message -> JSON.parseObject(message, AnalyticsAccessLogRecord.class));
16DataStream<OrderDoneLogRecord> orderRecordStream = orderSourceStream
17  .map(message -> JSON.parseObject(message, OrderDoneLogRecord.class));

join()

join() 算子提供的语义为"Window join"，即按照指定字段和（滚动/滑动/会话）窗口进行 inner join，支持处理时间和事件时间两种时间特征。以下示例以10秒滚动窗口，将两个流通过商品 ID 关联，取得订单流中的售价相关字段。

1clickRecordStream
2  .join(orderRecordStream)
3  .where(record -> record.getMerchandiseId())
4  .equalTo(record -> record.getMerchandiseId())
5  .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
6  .apply(new JoinFunction<AnalyticsAccessLogRecord, OrderDoneLogRecord, String>() {
7    @Override
8    public String join(AnalyticsAccessLogRecord accessRecord, OrderDoneLogRecord orderRecord) throws Exception {
9      return StringUtils.join(Arrays.asList(
10        accessRecord.getMerchandiseId(),
11        orderRecord.getPrice(),
12        orderRecord.getCouponMoney(),
13        orderRecord.getRebateAmount()
14      ), '\t');
15    }
16  })
17  .print().setParallelism(1);

简单易用。

coGroup()

只有 inner join 肯定还不够，如何实现 left/right outer join 呢？答案就是利用 coGroup() 算子。它的调用方式类似于 join() 算子，也需要开窗，但是 CoGroupFunction 比 JoinFunction 更加灵活，可以按照用户指定的逻辑匹配左流和/或右流的数据并输出。

以下的例子就实现了点击流 left join 订单流的功能，是很朴素的 nested loop join 思想（二重循环）。

1clickRecordStream
2  .coGroup(orderRecordStream)
3  .where(record -> record.getMerchandiseId())
4  .equalTo(record -> record.getMerchandiseId())
5  .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
6  .apply(new CoGroupFunction<AnalyticsAccessLogRecord, OrderDoneLogRecord, Tuple2<String, Long>>() {
7    @Override
8    public void coGroup(Iterable<AnalyticsAccessLogRecord> accessRecords, Iterable<OrderDoneLogRecord> orderRecords, Collector<Tuple2<String, Long>> collector) throws Exception {
9      for (AnalyticsAccessLogRecord accessRecord : accessRecords) {
10        boolean isMatched = false;
11        for (OrderDoneLogRecord orderRecord : orderRecords) {
12          // 右流中有对应的记录
13          collector.collect(new Tuple2<>(accessRecord.getMerchandiseName(), orderRecord.getPrice()));
14          isMatched = true;
15        }
16        if (!isMatched) {
17          // 右流中没有对应的记录
18          collector.collect(new Tuple2<>(accessRecord.getMerchandiseName(), null));
19        }
20      }
21    }
22  })
23  .print().setParallelism(1);

intervalJoin()

join() 和 coGroup() 都是基于窗口做关联的。但是在某些情况下，两条流的数据步调未必一致。例如，订单流的数据有可能在点击流的购买动作发生之后很久才被写入，如果用窗口来圈定，很容易 join 不上。所以 Flink 又提供了"Interval join"的语义，按照指定字段以及右流相对左流偏移的时间区间进行关联，即：

right.timestamp ∈ [left.timestamp + lowerBound; left.timestamp + upperBound]

interval join 也是 inner join，虽然不需要开窗，但是需要用户指定偏移区间的上下界，并且只支持事件时间。

示例代码如下。注意在运行之前，需要分别在两个流上应用 assignTimestampsAndWatermarks() 方法获取事件时间戳和水印。

1clickRecordStream
2  .keyBy(record -> record.getMerchandiseId())
3  .intervalJoin(orderRecordStream.keyBy(record -> record.getMerchandiseId()))
4  .between(Time.seconds(-30), Time.seconds(30))
5  .process(new ProcessJoinFunction<AnalyticsAccessLogRecord, OrderDoneLogRecord, String>() {
6    @Override
7    public void processElement(AnalyticsAccessLogRecord accessRecord, OrderDoneLogRecord orderRecord, Context context, Collector<String> collector) throws Exception {
8      collector.collect(StringUtils.join(Arrays.asList(
9        accessRecord.getMerchandiseId(),
10        orderRecord.getPrice(),
11        orderRecord.getCouponMoney(),
12        orderRecord.getRebateAmount()
13      ), '\t'));
14    }
15  })
16  .print().setParallelism(1);

由上可见，interval join 与 window join 不同，是两个 KeyedStream 之上的操作，并且需要调用 between() 方法指定偏移区间的上下界。如果想令上下界是开区间，可以调用 upperBoundExclusive()/lowerBoundExclusive() 方法。

interval join 的实现原理

以下是 KeyedStream.process(ProcessJoinFunction) 方法调用的重载方法的逻辑。

1public <OUT> SingleOutputStreamOperator<OUT> process(
2        ProcessJoinFunction<IN1, IN2, OUT> processJoinFunction,
3        TypeInformation<OUT> outputType) {
4    Preconditions.checkNotNull(processJoinFunction);
5    Preconditions.checkNotNull(outputType);
6    final ProcessJoinFunction<IN1, IN2, OUT> cleanedUdf = left.getExecutionEnvironment().clean(processJoinFunction);
7    final IntervalJoinOperator<KEY, IN1, IN2, OUT> operator =
8        new IntervalJoinOperator<>(
9            lowerBound,
10            upperBound,
11            lowerBoundInclusive,
12            upperBoundInclusive,
13            left.getType().createSerializer(left.getExecutionConfig()),
14            right.getType().createSerializer(right.getExecutionConfig()),
15            cleanedUdf
16        );
17    return left
18        .connect(right)
19        .keyBy(keySelector1, keySelector2)
20        .transform("Interval Join", outputType, operator);
21}

可见是先对两条流执行 connect() 和 keyBy() 操作，然后利用 IntervalJoinOperator 算子进行转换。在 IntervalJoinOperator 中，会利用两个 MapState 分别缓存左流和右流的数据。

1private transient MapState<Long, List<BufferEntry<T1>>> leftBuffer;
2private transient MapState<Long, List<BufferEntry<T2>>> rightBuffer;
3
4@Override
5public void initializeState(StateInitializationContext context) throws Exception {
6    super.initializeState(context);
7    this.leftBuffer = context.getKeyedStateStore().getMapState(new MapStateDescriptor<>(
8        LEFT_BUFFER,
9        LongSerializer.INSTANCE,
10        new ListSerializer<>(new BufferEntrySerializer<>(leftTypeSerializer))
11    ));
12    this.rightBuffer = context.getKeyedStateStore().getMapState(new MapStateDescriptor<>(
13        RIGHT_BUFFER,
14        LongSerializer.INSTANCE,
15        new ListSerializer<>(new BufferEntrySerializer<>(rightTypeSerializer))
16    ));
17}

其中 Long 表示事件时间戳，List> 表示该时刻到来的数据记录。当左流和右流有数据到达时，会分别调用 processElement1() 和 processElement2() 方法，它们都调用了 processElement() 方法，代码如下。

1@Override
2public void processElement1(StreamRecord<T1> record) throws Exception {
3    processElement(record, leftBuffer, rightBuffer, lowerBound, upperBound, true);
4}
5
6@Override
7public void processElement2(StreamRecord<T2> record) throws Exception {
8    processElement(record, rightBuffer, leftBuffer, -upperBound, -lowerBound, false);
9}
10
11@SuppressWarnings("unchecked")
12private <THIS, OTHER> void processElement(
13        final StreamRecord<THIS> record,
14        final MapState<Long, List<IntervalJoinOperator.BufferEntry<THIS>>> ourBuffer,
15        final MapState<Long, List<IntervalJoinOperator.BufferEntry<OTHER>>> otherBuffer,
16        final long relativeLowerBound,
17        final long relativeUpperBound,
18        final boolean isLeft) throws Exception {
19    final THIS ourValue = record.getValue();
20    final long ourTimestamp = record.getTimestamp();
21    if (ourTimestamp == Long.MIN_VALUE) {
22        throw new FlinkException("Long.MIN_VALUE timestamp: Elements used in " +
23                "interval stream joins need to have timestamps meaningful timestamps.");
24    }
25    if (isLate(ourTimestamp)) {
26        return;
27    }
28    addToBuffer(ourBuffer, ourValue, ourTimestamp);
29    for (Map.Entry<Long, List<BufferEntry<OTHER>>> bucket: otherBuffer.entries()) {
30        final long timestamp  = bucket.getKey();
31        if (timestamp < ourTimestamp + relativeLowerBound ||
32                timestamp > ourTimestamp + relativeUpperBound) {
33            continue;
34        }
35        for (BufferEntry<OTHER> entry: bucket.getValue()) {
36            if (isLeft) {
37                collect((T1) ourValue, (T2) entry.element, ourTimestamp, timestamp);
38            } else {
39                collect((T1) entry.element, (T2) ourValue, timestamp, ourTimestamp);
40            }
41        }
42    }
43    long cleanupTime = (relativeUpperBound > 0L) ? ourTimestamp + relativeUpperBound : ourTimestamp;
44    if (isLeft) {
45        internalTimerService.registerEventTimeTimer(CLEANUP_NAMESPACE_LEFT, cleanupTime);
46    } else {
47        internalTimerService.registerEventTimeTimer(CLEANUP_NAMESPACE_RIGHT, cleanupTime);
48    }
49}

这段代码的思路是：

取得当前流 StreamRecord 的时间戳，调用 isLate() 方法判断它是否是迟到数据（即时间戳小于当前水印值），如是则丢弃。
调用 addToBuffer() 方法，将时间戳和数据一起插入当前流对应的 MapState。
遍历另外一个流的 MapState，如果数据满足前述的时间区间条件，则调用 collect() 方法将该条数据投递给用户定义的 ProcessJoinFunction 进行处理。collect() 方法的代码如下，注意结果对应的时间戳是左右流时间戳里较大的那个。

private void collect(T1 left, T2 right, long leftTimestamp, long rightTimestamp) throws Exception { final long resultTimestamp = Math.max(leftTimestamp, rightTimestamp); collector.setAbsoluteTimestamp(resultTimestamp); context.updateTimestamps(leftTimestamp, rightTimestamp, resultTimestamp); userFunction.processElement(left, right, context, collector); }
调用 TimerService.registerEventTimeTimer() 注册时间戳为 timestamp + relativeUpperBound 的定时器，该定时器负责在水印超过区间的上界时执行状态的清理逻辑，防止数据堆积。注意左右流的定时器所属的 namespace 是不同的，具体逻辑则位于 onEventTime() 方法中。

@Override public void onEventTime(InternalTimer<K, String> timer) throws Exception { long timerTimestamp = timer.getTimestamp(); String namespace = timer.getNamespace(); logger.trace("onEventTime @ {}", timerTimestamp); switch (namespace) { case CLEANUP_NAMESPACE_LEFT: { long timestamp = (upperBound <= 0L) ? timerTimestamp : timerTimestamp - upperBound; logger.trace("Removing from left buffer @ {}", timestamp); leftBuffer.remove(timestamp); break; } case CLEANUP_NAMESPACE_RIGHT: { long timestamp = (lowerBound <= 0L) ? timerTimestamp + lowerBound : timerTimestamp; logger.trace("Removing from right buffer @ {}", timestamp); rightBuffer.remove(timestamp); break; } default: throw new RuntimeException("Invalid namespace " + namespace); } }

原文链接

本文为阿里云原创内容，未经允许不得转载。