Savepoint - 操作符在 Apache Flink 中无法匹配

Question

我正在尝试使用保存点停止我的工作，然后使用相同的保存点再次启动它。就我而言，我更新了我的工作，并用新的 jar 为它创建了新版本。这是我的代码示例；

class Reader(bla bla) {
   def read() = {
     val ds = readFromKafka()
     transform(ds)
   }
   def transform(ds: DataStream[]) = {
      ds.map()
   }
}

object MyJob {
  def run () = {
    val data = new Reader().read()
    data.keyBy(id).process(new MyStateFunc).uid("my-uid") // then write to kafka
  }
}

在这种情况下，我确实使用保存点停止了作业，然后使用具有相同 jar 的相同保存点启动它。然后，我像这样向我的阅读器添加一个过滤器；

class Reader(bla bla) {
   def read() = {
     val ds = readFromKafka()
     transform(ds)
   }
   def transform(ds: DataStream[]) = {
      ds.map().filter() // FILTER ADDED HERE
   }
}

我用保存点停止了我的工作，它有效。然后我尝试使用相同的保存点部署具有新版本（新过滤方法）的作业，它无法匹配操作员并且作业未部署。为什么？

Answer 1

除非您在获取保存点之前为所有有状态操作符明确提供 UID，否则在更改作业的拓扑之后，Flink 将不再能够确定保存点中的哪个状态属于哪个操作符。

我看到您的键控进程函数（“my-uid”）上有一个 UID。但是您还需要在 Kafka 源和接收器上拥有 UID，以及其他任何有状态的东西。这些 UID 需要附加到有状态操作符本身，并且需要在作业中是唯一的（但不是在所有作业中）。 （此外，每个状态描述符都需要为每个状态分配一个名称，使用在运算符中唯一的名称。）

通常一个人会做这样的事情

env
  .addSource(...)
  .name("KafkaSource")
  .uid("KafkaSource")

results.addSink(...)
    .name("KafkaSink")
    .uid("KafkaSink")

name() 方法用于提供出现在 Web UI 中的文本。