如何在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

这是现在不可能的，至少没有高级数据流API。

可以实现首先在输入上读取的低级操作员，然后再读取其他输入。但是，这将完全阻止一个输入，该输入与缩小处理方式并执行检查点的方式无法很好地工作。

将来，使用所谓的侧输入将可能。

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

您可以通过其中的flatMap大约实现此目标。但是实际上，这取决于一些问题。如果，例如来自某些输入流的元素被延迟，不会严格订购输出。

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
}