如何在Apache Flink中连接两个流

Question

例如，我想在单个中组合1,2,3和4,5的流，因此结果应该是：1, 2, 3, 4, 5。换句话说：如果第一个源已耗尽-从第二个源获取元素。我最近的尝试是：

val a = streamEnvironment.fromElements(1, 2, 3)

val b = streamEnvironment.fromElements(4, 5)

val c = a.union(b)

c.map(x => println(s"X=$x")) // X=4, 5, 1, 2, 3 or something like that

也对datetime进行了类似的尝试，但结果相同。

Answer 1

您可以通过一个包含堆缓冲区的平面图大致实现这一点。但实际上，这取决于一些问题。例如，如果来自某个输入流的元素被延迟，则输出将不会严格排序。

def process(): StreamExecutionEnvironment = {
val env = StreamExecutionEnvironment.getExecutionEnvironment

implicit val typeInfo = TypeInformation.of(classOf[Int])
implicit val typeInfo2 = TypeInformation.of(classOf[Unit])

val BUF_SIZE = 3
val STREAM_NUM = 2

val a = env.fromElements(1, 2, 3, 3, 4, 5, 6, 7, Int.MaxValue)
val b = env.fromElements(4, 5, 9, 10 , 11, 13, Int.MaxValue)

val c = a.union(b).flatMap(new FlatMapFunction[Int, Int] {
  val heap = collection.mutable.PriorityQueue[Int]().reverse
  var endCount = 0

  override def flatMap(value: Int, out: Collector[Int]): Unit = {
    if (value == Int.MaxValue) {
      endCount += 1

      if (endCount == STREAM_NUM) {
        heap.foreach(out.collect)
      }
    }
    else {
      heap += value

      while (heap.size > BUF_SIZE) {
        val v = heap.dequeue()
        out.collect(v)
      }
    }
  }
}).setParallelism(1)

c.map(x => println(s"X=$x")).setParallelism(1)

env
}

Answer 2

如果您有N个源（不是流）要连续排序，那么您可以用外部源包装它们。类似于：

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

@SuppressWarnings("serial")
public class SequentialSources<T> implements SourceFunction<T>, ResultTypeQueryable<T> {

    private TypeInformation<T> type;
    private SourceFunction<T>[] sources;
    private volatile boolean isRunning = true;

    public SequentialSources(TypeInformation<T> type, SourceFunction<T>...sources) {
        this.type = type;
        this.sources = sources;
    }

    @Override
    public void run(SourceContext<T> context) throws Exception {
        int index = 0;

        while (isRunning) {
            sources[index++].run(context);
            isRunning = index < sources.length;
        }
    }

    @Override
    public void cancel() {
        isRunning = false;

        for (SourceFunction<T> source : sources) {
            source.cancel();
        }
    }

    @Override
    public TypeInformation<T> getProducedType() {
        return type;
    }
}

Answer 3

这在现在是不可能的，至少在高级DataStream API中是不可能的。

可以实现一个低级运算符，首先读取输入，然后读取另一个输入。但是，这将完全阻止一个输入，该输入与Flink处理水印和执行检查点的方式不一致。

在未来，这将有可能使用所谓的侧输入。