Flink 流：从一个窗口，在另一个窗口中查找状态

Question

我有两条流：

• 测量
• WhoMeasured（关于谁进行了测量的元数据）

这些是它们的案例类：

case class Measurement(var value: Int, var who_measured_id: Int)
case class WhoMeasured(var who_measured_id: Int, var name: String)

测量流包含大量数据。WhoMeasured流几乎没有任何可用性。事实上，对于＜code＞who_measured_id＜/code＞流中的每个＜code＞who_。这本质上是一个哈希表，由WhoMeasured流填充。

在我的自定义窗口函数中

class WFunc extends WindowFunction[Measurement, Long, Int, TimeWindow] {
  override def apply(key: Int, window: TimeWindow, input: Iterable[Measurement], out: Collector[Long]): Unit = {

    // Here I need access to the WhoMeasured stream to get the name of the person who took a measurement
    // The following two are equivalent since I keyed by who_measured_id
    val name_who_measured = magic(key)
    val name_who_measured = magic(input.head.who_measured_id)
  }
}

这是我的工作。现在你可能会看到，有一些东西不见了：两个流的结合。

val who_measured_stream = who_measured_source
  .keyBy(w => w.who_measured_id)
  .countWindow(1)

val measurement_stream = measurements_source
  .keyBy(m => m.who_measured_id)
  .timeWindow(Time.seconds(60), Time.seconds(5))
  .apply(new WFunc)

因此，从本质上讲，这是一种查找表，当Who量测流中的新元素到达时会被更新。

所以问题是：如何实现从一个窗口流到另一个窗口流的查找？

跟进:

按照 Fabian 建议的方式实现后，作业总是会因某种序列化问题而失败：

[info] Loading project definition from /home/jgroeger/Code/MeasurementJob/project
[info] Set current project to MeasurementJob (in build file:/home/jgroeger/Code/MeasurementJob/)
[info] Compiling 8 Scala sources to /home/jgroeger/Code/MeasurementJob/target/scala-2.11/classes...
[info] Running de.company.project.Main dev MeasurementJob
[error] Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: The implementation of the RichCoFlatMapFunction is not serializable. The object probably contains or references non serializable fields.
[error]     at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
[error]     at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1478)
[error]     at org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:161)
[error]     at org.apache.flink.streaming.api.datastream.ConnectedStreams.flatMap(ConnectedStreams.java:230)
[error]     at org.apache.flink.streaming.api.scala.ConnectedStreams.flatMap(ConnectedStreams.scala:127)
[error]     at de.company.project.jobs.MeasurementJob.run(MeasurementJob.scala:139)
[error]     at de.company.project.Main$.main(Main.scala:55)
[error]     at de.company.project.Main.main(Main.scala)
[error] Caused by: java.io.NotSerializableException: de.company.project.jobs.MeasurementJob
[error]     at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
[error]     at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
[error]     at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
[error]     at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
[error]     at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
[error]     at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
[error]     at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:301)
[error]     at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:81)
[error]     ... 7 more
java.lang.RuntimeException: Nonzero exit code returned from runner: 1
    at scala.sys.package$.error(package.scala:27)
[trace] Stack trace suppressed: run last MeasurementJob/compile:run for the full output.
[error] (MeasurementJob/compile:run) Nonzero exit code returned from runner: 1
[error] Total time: 9 s, completed Nov 15, 2016 2:28:46 PM

Process finished with exit code 1

错误消息：

The implementation of the RichCoFlatMapFunction is not serializable. The object probably contains or references non serializable fields.

但是，我的 JoiningCoFlatMap 唯一的字段是建议的 ValueState。

签名如下所示：

class JoiningCoFlatMap extends RichCoFlatMapFunction[Measurement, WhoMeasured, (Measurement, String)] {

Answer 1

我想你要做的是一个窗口操作，然后是一个连接。

您可以使用有状态的< code>CoFlatMapFunction来实现高容量流和低值按键更新流的联接，如下例所示:

val measures: DataStream[Measurement] = ???
val who: DataStream[WhoMeasured] = ???

val agg: DataStream[(Int, Long)] = measures
  .keyBy(_._2) // measured_by_id
  .timeWindow(Time.seconds(60), Time.seconds(5))
  .apply( (id: Int, w: TimeWindow, v: Iterable[(Int, Int, String)], out: Collector[(Int, Long)]) => {
    // do your aggregation
  })

val joined: DataStream[(Int, Long, String)] = agg
  .keyBy(_._1) // measured_by_id
  .connect(who.keyBy(_.who_measured_id))
  .flatMap(new JoiningCoFlatMap)

// CoFlatMapFunction
class JoiningCoFlatMap extends RichCoFlatMapFunction[(Int, Long), WhoMeasured, (Int, Long, String)] {

  var names: ValueState[String] = null

  override def open(conf: Configuration): Unit = {
    val stateDescrptr = new ValueStateDescriptor[String](
      "whoMeasuredName",
      classOf[String],
      ""                 // default value
    )
    names = getRuntimeContext.getState(stateDescrptr)
  }

  override def flatMap1(a: (Int, Long), out: Collector[(Int, Long, String)]): Unit = {
    // join with state
    out.collect( (a._1, a._2, names.value()) )
  }

  override def flatMap2(w: WhoMeasured, out: Collector[(Int, Long, String)]): Unit = {
    // update state
    names.update(w.name)
  }
}

关于实现的说明:< code>CoFlatMapFunction不能决定处理哪个输入，即< code>flatmap1和< code>flatmap2函数的调用取决于到达操作符的数据。它不能由函数控制。这是初始化状态时的一个问题。开始时，状态可能没有到达的< code>Measurement对象的正确名称，但会返回默认值。您可以通过缓冲测量值并连接它们一次来避免这种情况，来自< code>who流的键的第一个更新到达。为此你需要另一个州。