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

我有两条流：
测量
whomeasured（关于谁进行测量的元数据）
以下是它们的案例类：

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

这个 Measurement 流有很多数据。这个 WhoMeasured 小溪里几乎没有水。事实上，对于每个 who_measured_id 在 WhoMeasured 流中，只有一个名称是相关的，因此如果一个元素具有相同的名称，则可以丢弃旧元素 who_measured_id 到达。这本质上是一个由 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)

所以本质上这是一种查找表，当 WhoMeasured 小溪到了。
所以问题是：如何从一个 WindowedStream 变成另一个？
跟进：
按照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)] {

我想你要做的是一个窗口操作，然后是一个连接。
您可以使用有状态的 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)
  }
}

关于执行情况的说明：a CoFlatMapFunction 无法决定要处理的输入，即 flatmap1 以及 flatmap2 函数的调用取决于到达操作符的数据。它不能由函数控制。初始化状态时出现问题。一开始，州政府可能没有一个正确的名字 Measurement 对象，但返回默认值。您可以通过缓冲测量值并连接它们一次来避免这种情况，即从 who 小溪来了。你需要另一个州。

展开查看全部

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

1条答案

相关问题

热门标签

最新问答