flink：如何在flink中处理外部应用程序配置更改

Answer 1

更新正在运行的流式应用程序的配置是一个常见的要求。在Flink的DataStream API中，可以使用所谓的CoflatMapFunction来实现这一点，它处理两个输入流。其中一个流可以是数据流，而另一个流可以是控制流。

下面的示例演示如何动态适配一个筛选出超过一定长度的字符串的用户函数。

val data: DataStream[String] = ???
val control: DataStream[Int] = ???

val filtered: DataStream[String] = data
  // broadcast all control messages to the following CoFlatMap subtasks
  .connect(control.broadcast)
  // process data and control messages
  .flatMap(new DynLengthFilter)


class DynLengthFilter extends CoFlatMapFunction[String, Int, String] with Checkpointed[Int] {

  var length = 0

  // filter strings by length
  override def flatMap1(value: String, out: Collector[String]): Unit = {
    if (value.length < length) {
      out.collect(value)
    }
  }

  // receive new filter length
  override def flatMap2(value: Int, out: Collector[String]): Unit = {
    length = value
  }

  override def snapshotState(checkpointId: Long, checkpointTimestamp: Long): Int = length

  override def restoreState(state: Int): Unit = {
    length = state
  }
}

dynlengthfilter用户函数实现筛选器长度的checkpoint接口。如果发生故障，该信息将自动恢复。