使用Flink同步处理2个流

Question

我有两个流A和B。

我开始同时吃A和B。

流A仅在每分钟的第59秒获得记录。

流B在每分钟的任何一秒都有记录。

我希望处理使两个流同步。

示例：在10:01:59之后从流A中，我将在10:02:59收到一条记录，直到10:02:59，我也不想从流B中读取任何内容。

这可以在Flink中实现吗？

Answer 1

Flink使用基于推的模型（当源和汇被重构为基于拉的模型时，这种模型应该很快就会改变）来处理下游的元素。这意味着您不能“等到事件到达后再拉入更多数据”，同时您必须在某些操作符状态下缓冲这些数据。Flink提供各种状态后端供您使用。

为了对Kkrugler的答案进行可视化，给定两个流，我们将以逻辑方式连接它们，然后在另一个元素到达时使用ListState对其中一个进行检索：

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.JavaConverters._ 

object Test {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironment()
    val streamA = env.fromCollection(List(1, 2, 3))
    val streamB = env.fromCollection(List("a", "b", "c"))

    streamA
      .connect(streamB)
      .process {
        new CoProcessFunction[Int, String, (Int, String)] {
          var myStateA: ListState[Int] = _

          override def open(parameters: Configuration): Unit = {
            myStateA = getRuntimeContext.getListState[Int](
              new ListStateDescriptor[Int]("my_state", classOf[Int])
            )
          }

          override def processElement1(
              value: Int,
              ctx: CoProcessFunction[Int, String, (Int, String)]#Context,
              out: Collector[(Int, String)]
          ): Unit = {
            myStateA.add(value)
          }

          override def processElement2(
              value: String,
              ctx: CoProcessFunction[Int, String, (Int, String)]#Context,
              out: Collector[(Int, String)]
          ): Unit = {
            val list = myStateA.get().iterator().asScala.toList
            val intFromState = list.headOption
            intFromState match {
              case Some(myInt) =>
                out.collect((myInt, value))
              case None => ()
            }

            myStateA.update(list.tail.asJava)
          }
        }
      }
  }
}

注意：简化了此实现。这里无法保证元素的到达顺序，您需要将其添加到您的状态和实现中。您还可以使用计时器，从而为每个进入流的事件注册一个计时器，作为新数据何时到达的指示。

Answer 2

您不能在Flink中从流中读取记录，但可以从流中删除（或保存）记录。因此，您可以连接这两个流，并使用共平面图进行处理。当您从流a中获取记录时，请将其保存在状态。当您从流B获得记录时，根据流a的状态决定如何处理它。