在使用Flink做数据处理时，一个典型的场景是对两个流的数据进行关联。Flink DataStream API中，相关方法有3个：join, coGroup和intervalJoin。

Join和CoGroup

调用示例

  1. stream.join(otherStream)
  2.     .where(<KeySelector>)
  3.     .equalTo(<KeySelector>)
  4.     .window(<WindowAssigner>)
  5.     .apply(<JoinFunction>)
  7. stream.coGroup(otherStream)
  8.     .where(<KeySelector>)
  9.     .equalTo(<KeySelector>)
  10.     .window(<WindowAssigner>)
  11.     .apply(<CoGroupFunction>)

介绍

这两个都属于WindowJoin，join为在窗口内做inner join，而coGroup为在窗口内做outer join。
以 join为例，根据指定Window类型的不同，能达到以下效果：

Tumbling Window Join image.png
Sliding Window Join image.png
Session Window Join image.png

join后元素的时间戳为该窗口的最大时间戳。

代码

  1. public interface JoinFunction<IN1, IN2, OUT> extends Function, Serializable {
  3.     /**
  4.      * The join method, called once per joined pair of elements.
  5.      */
  6.     OUT join(IN1 first, IN2 second) throws Exception;
  7. }
  9. public interface CoGroupFunction<IN1, IN2, O> extends Function, Serializable {
  11.     /**
  12.      * This method must be implemented to provide a user implementation of a
  13.      * coGroup. It is called for each pair of element groups where the elements share the
  14.      * same key.
  15.      */
  16.     void coGroup(Iterable<IN1> first, Iterable<IN2> second, Collector<O> out) throws Exception;
  17. }
  • JoinFunction作用于一个窗口内，参数为两个流中按key匹配上的每一对元素，为innerjoin
  • CoGroupFunction的参数为一个窗口内，两个流中key相同的所有元素，因此即可实现innerjoin，又可实现outerjoin。

先来看一下join的代码

  1. stream.join(otherStream) -> JoinedStreams
  2.     .where(<KeySelector>) -> JoinedStreams.Where
  3.     .equalTo(<KeySelector>) -> JoinedStreams.EqualTo
  4.     .window(<WindowAssigner>) -> JoinedStreams.WithWindow
  5.     .apply(<JoinFunction/FlatJoinFunction>)
  • 核心代码在JoinedStreams.WithWindow中，别的都是简单的check和wrap
  • Join被转换成CoGroup进行处理


  1. /**
  2.  * A streaming join operation is evaluated over elements in a window.
  3.  *
  4.  * <p>Note: Right now, the join is being evaluated in memory so you need to ensure that the number
  5.  * of elements per key does not get too high. Otherwise the JVM might crash.
  6.  */
  7. @Public
  8. public class JoinedStreams<T1, T2> {
  10.     @Public
  11.     public static class WithWindow<T1, T2, KEY, W extends Window> {
  12.         private final DataStream<T1> input1;
  13.         private final DataStream<T2> input2;
  14.         private CoGroupedStreams.WithWindow<T1, T2, KEY, W> coGroupedWindowedStream;
  16.         public <T> DataStream<T> apply(JoinFunction<T1, T2, T> function, TypeInformation<T> resultType) {
  17.             // clean the closure
  18.             function = input1.getExecutionEnvironment().clean(function);
  20.             // Join被转换成CoGroup操作
  21.             coGroupedWindowedStream = input1.coGroup(input2)
  22.                 .where(keySelector1)
  23.                 .equalTo(keySelector2)
  24.                 .window(windowAssigner)
  25.                 .trigger(trigger)
  26.                 .evictor(evictor)
  27.                 .allowedLateness(allowedLateness);
  29.             // 用户传入的JoinFunction被封装成JoinCoGroupFunction
  30.             return coGroupedWindowedStream
  31.                     .apply(new JoinCoGroupFunction<>(function), resultType);
  32.         }
  33.     }
  34. }
  1. /**
  2.  * A streaming join operation is evaluated over elements in a window.
  3.  *
  4.  * <p>Note: Right now, the join is being evaluated in memory so you need to ensure that the number
  5.  * of elements per key does not get too high. Otherwise the JVM might crash.
  6.  */
  7. @Public
  8. public class JoinedStreams<T1, T2> {
  10.     /**
  11.      * CoGroup function that does a nested-loop join to get the join result.
  12.      */
  13.     private static class JoinCoGroupFunction<T1, T2, T>
  14.             extends WrappingFunction<JoinFunction<T1, T2, T>>
  15.             implements CoGroupFunction<T1, T2, T> {
  17.         public JoinCoGroupFunction(JoinFunction<T1, T2, T> wrappedFunction) {
  18.             super(wrappedFunction);
  19.         }
  21.         @Override
  22.         public void coGroup(Iterable<T1> first, Iterable<T2> second, Collector<T> out) throws Exception {
  23.             for (T1 val1: first) {
  24.                 for (T2 val2: second) {
  25.                     out.collect(wrappedFunction.join(val1, val2));
  26.                 }
  27.             }
  28.         }
  29.     }
  30. }

再看一下coGroup的代码：

Flink先将两个流union成一个流（元素类型为TaggedUnion，可标识属于左流还是右流），将DataStream变成KeyedDataStream，再套上window。

  1.     @Internal
  2.     public static class TaggedUnion<T1, T2> {
  3.         private final T1 one;
  4.         private final T2 two;
  5.     }


  1. @Public
  2. public class CoGroupedStreams<T1, T2> {
  3.     @Public
  4.     public static class WithWindow<T1, T2, KEY, W extends Window> {
  5.         private WindowedStream<TaggedUnion<T1, T2>, KEY, W> windowedStream;
  7.         public <T> DataStream<T> apply(CoGroupFunction<T1, T2, T> function, TypeInformation<T> resultType) {
  8.             //clean the closure
  9.             function = input1.getExecutionEnvironment().clean(function);
  10.             UnionTypeInfo<T1, T2> unionType = new UnionTypeInfo<>(input1.getType(), input2.getType());
  11.             UnionKeySelector<T1, T2, KEY> unionKeySelector = new UnionKeySelector<>(keySelector1, keySelector2);
  13.             // 给两个流分别加上标识
  14.             DataStream<TaggedUnion<T1, T2>> taggedInput1 = input1
  15.                     .map(new Input1Tagger<T1, T2>())  // 生成TaggedUnion(one, null)
  16.                     .setParallelism(input1.getParallelism())
  17.                     .returns(unionType);
  18.             DataStream<TaggedUnion<T1, T2>> taggedInput2 = input2
  19.                     .map(new Input2Tagger<T1, T2>()) // 生成TaggedUnion(null, two)
  20.                     .setParallelism(input2.getParallelism())
  21.                     .returns(unionType);
  23.             // 两个流合并成一个流
  24.             DataStream<TaggedUnion<T1, T2>> unionStream = taggedInput1.union(taggedInput2);
  26.             // we explicitly create the keyed stream to manually pass the key type information in
  27.             windowedStream =
  28.                     new KeyedStream<TaggedUnion<T1, T2>, KEY>(unionStream, unionKeySelector, keyType)
  29.                     .window(windowAssigner);
  31.             if (trigger != null) {
  32.                 windowedStream.trigger(trigger);
  33.             }
  34.             if (evictor != null) {
  35.                 windowedStream.evictor(evictor);
  36.             }
  37.             if (allowedLateness != null) {
  38.                 windowedStream.allowedLateness(allowedLateness);
  39.             }
  41.             return windowedStream.apply(new CoGroupWindowFunction<T1, T2, T, KEY, W>(function), resultType);
  42.         }
  43.     }

window被触发时，将其中的元素分开成左边一组和右边一组，再交给CoGroupFunction处理。

  1. @Public
  2. public class CoGroupedStreams<T1, T2> {    
  3.         private static class CoGroupWindowFunction<T1, T2, T, KEY, W extends Window>
  4.             extends WrappingFunction<CoGroupFunction<T1, T2, T>>
  5.             implements WindowFunction<TaggedUnion<T1, T2>, T, KEY, W> {
  7.         public CoGroupWindowFunction(CoGroupFunction<T1, T2, T> userFunction) {
  8.             super(userFunction);
  9.         }
  11.         @Override
  12.         public void apply(KEY key,
  13.                 W window,
  14.                 Iterable<TaggedUnion<T1, T2>> values,
  15.                 Collector<T> out) throws Exception {
  17.             List<T1> oneValues = new ArrayList<>();
  18.             List<T2> twoValues = new ArrayList<>();
  20.             // 窗口内的所有元素按左流和右流拆分开
  21.             for (TaggedUnion<T1, T2> val: values) {
  22.                 if (val.isOne()) {
  23.                     oneValues.add(val.getOne());
  24.                 } else {
  25.                     twoValues.add(val.getTwo());
  26.                 }
  27.             }
  28.             wrappedFunction.coGroup(oneValues, twoValues, out);
  29.         }
  30.     }
  31. }


IntervalJoin

调用示例

  1. import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
  2. import org.apache.flink.streaming.api.windowing.time.Time;
  4. ...
  6. val orangeStream: DataStream[Integer] = ...
  7. val greenStream: DataStream[Integer] = ...
  9. orangeStream
  10.     .keyBy(elem => /* select key */)
  11.     .intervalJoin(greenStream.keyBy(elem => /* select key */))
  12.     .between(Time.milliseconds(-2), Time.milliseconds(1))
  13.     .process(new ProcessJoinFunction[Integer, Integer, String] {
  14.         override def processElement(left: Integer, right: Integer, ctx: ProcessJoinFunction[Integer, Integer, String]#Context, out: Collector[String]): Unit = {
  15.          out.collect(left + "," + right); 
  16.         }
  17.       });
  18.     });

介绍

  • Window Join和CoGroup要求关联上的两个消息必须在同一个窗口内

  • 若想根据两个流中消息的相对时间来匹配，需要用Interval Join

    orangeElem.ts`` + ``lowerBound``<= ``greenElem.ts``<= ``orangeElem.ts`` + upperBound

  • Join后的元素的时间戳为两个消息时间戳中更大的一个

image.png

代码

Interval Join是基于ConnectedStreams实现的

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


  • ConnectedStreams可视为两个流的union，不要求两个流的数据类型一样
  • 通过流之间共享状态，可以在双流上实现一些复杂操作。


  1. @Public
  2. public class ConnectedStreams<IN1, IN2> {
  4.     protected final DataStream<IN1> inputStream1;
  5.     protected final DataStream<IN2> inputStream2;
  7.     public ConnectedStreams<IN1, IN2> keyBy(KeySelector<IN1, ?> keySelector1, KeySelector<IN2, ?> keySelector2) {
  8.         return new ConnectedStreams<>(environment, inputStream1.keyBy(keySelector1),
  9.             inputStream2.keyBy(keySelector2));
  10.     }
  12.     public <R> SingleOutputStreamOperator<R> map(CoMapFunction<IN1, IN2, R> coMapper) {
  13.         TypeInformation<R> outTypeInfo = TypeExtractor.getBinaryOperatorReturnType(
  14.             coMapper,
  15.             CoMapFunction.class,
  16.             0,
  17.             1,
  18.             2,
  19.             TypeExtractor.NO_INDEX,
  20.             getType1(),
  21.             getType2(),
  22.             Utils.getCallLocationName(),
  23.             true);
  24.         return map(coMapper, outTypeInfo);
  25.     }
  26. }
  28. @Public
  29. public interface CoMapFunction<IN1, IN2, OUT> extends Function, Serializable {
  30.     OUT map1(IN1 value) throws Exception;
  31.     OUT map2(IN2 value) throws Exception;
  32. }

在IntervalJoinOprater中，用两个MapState(leftBuffer, rightBuffer)分别保存两个流中到达的消息，map的key是消息的时间戳。

  1. @Internal
  2. public class IntervalJoinOperator<K, T1, T2, OUT>
  3.         extends AbstractUdfStreamOperator<OUT, ProcessJoinFunction<T1, T2, OUT>>
  4.         implements TwoInputStreamOperator<T1, T2, OUT>, Triggerable<K, String> {
  6.     private final long lowerBound;
  7.     private final long upperBound;
  9.     private transient MapState<Long, List<BufferEntry<T1>>> leftBuffer;
  10.     private transient MapState<Long, List<BufferEntry<T2>>> rightBuffer;
  12.     private transient InternalTimerService<String> internalTimerService;
  14.     @Override
  15.     public void processElement1(StreamRecord<T1> record) throws Exception {
  16.         processElement(record, leftBuffer, rightBuffer, lowerBound, upperBound, true);
  17.     }
  19.     @Override
  20.     public void processElement2(StreamRecord<T2> record) throws Exception {
  21.         processElement(record, rightBuffer, leftBuffer, -upperBound, -lowerBound, false);
  22.     }
  23. }


  • 当左流的消息到达时，会将其加入leftBuffer中，然后到rightBuffer中找是否有符合时间范围的消息。如果有，匹配的元素就被传入用户自定义的ProcessJoinFunction中。反之亦然。
  • join上的消息对的时间戳为两个消息中更大的一个。
  • 为了避免两个state无限膨胀，每条消息会注册一个timer，在时间越过该消息的有效范围后，便将其移出state。


  1. @Internal
  2. public class IntervalJoinOperator<K, T1, T2, OUT> {
  3.     @SuppressWarnings("unchecked")
  4.     private <THIS, OTHER> void processElement(
  5.             final StreamRecord<THIS> record,
  6.             final MapState<Long, List<IntervalJoinOperator.BufferEntry<THIS>>> ourBuffer,
  7.             final MapState<Long, List<IntervalJoinOperator.BufferEntry<OTHER>>> otherBuffer,
  8.             final long relativeLowerBound,
  9.             final long relativeUpperBound,
  10.             final boolean isLeft) throws Exception {
  11.         final THIS ourValue = record.getValue();
  12.         final long ourTimestamp = record.getTimestamp();
  14.         if (ourTimestamp == Long.MIN_VALUE) {
  15.             throw new FlinkException("Long.MIN_VALUE timestamp: Elements used in " +
  16.                     "interval stream joins need to have timestamps meaningful timestamps.");
  17.         }
  18.         if (isLate(ourTimestamp)) { // 若消息时间戳小于internalTimerService.currentWatermark()，丢弃
  19.             return;
  20.         }
  21.         // 将消息加入状态
  22.         addToBuffer(ourBuffer, ourValue, ourTimestamp);
  24.         // 从另一个数据流的状态中查找匹配的消息
  25.         for (Map.Entry<Long, List<BufferEntry<OTHER>>> bucket: otherBuffer.entries()) {
  26.             final long timestamp  = bucket.getKey();
  27.             if (timestamp < ourTimestamp + relativeLowerBound ||
  28.                     timestamp > ourTimestamp + relativeUpperBound) {
  29.                 continue;
  30.             }
  31.             for (BufferEntry<OTHER> entry: bucket.getValue()) {
  32.                 if (isLeft) {
  33.                     collect((T1) ourValue, (T2) entry.element, ourTimestamp, timestamp);
  34.                 } else {
  35.                     collect((T1) entry.element, (T2) ourValue, timestamp, ourTimestamp);
  36.                 }
  37.             }
  38.         }
  39.         // 注册清理消息的timer
  40.         long cleanupTime = (relativeUpperBound > 0L) ? ourTimestamp + relativeUpperBound : ourTimestamp;
  41.         if (isLeft) {
  42.             internalTimerService.registerEventTimeTimer(CLEANUP_NAMESPACE_LEFT, cleanupTime);
  43.         } else {
  44.             internalTimerService.registerEventTimeTimer(CLEANUP_NAMESPACE_RIGHT, cleanupTime);
  45.         }
  46.     }
  47. }