我正在编写一个流式 Flink 程序，用于为我们的离线训练模型进行特征提取，并对程序的设计感到疑惑。我希望每个特征提取逻辑都能在其类中维护自己的状态，这样添加新的特征提取就相当于添加一个新类。

大致的高层次设计如下：

#data 是代码中特征提取逻辑的相对路径流，例如 com.xxx.FeatureExtraction1
val data:DataStream[String] = ...
#基于相对路径，使用反射来初始化类
featureExtraction1 = method.getReflect("com.xxx.FeatureExtraction1")
data.keyBy(_).flatmap(featureExtraction1)

其中每个特征提取逻辑都有自己的内部状态跟踪

class FeatureExtraction1 extends RichFlatMapFunction[String, Double)] {
  private var mystate: MapState = _
  override def flatMap(input: String, out: Collector[Double]) = {
    // 访问状态值
  }
  override def open(parameters: Configuration): Unit = {
    mystate = xxx
  }
}

我可以这样做，只要我添加了一个新的特征提取类，例如 com.xxx.FeatureExtraction2，我就将它附加到数据流中，如下所示：

data.keyBy(_).flatmap(featureExtraction1).flatmap(featureExtraction2)...flatmap(featureExtractionN)

然而，我对 Flink 的了解不足以确定如果像这样链接，featureExtraction1 到 featureExtractionN 是否会并发执行（在我看来应该是这样的）。其次，我想编写代码，让它自动创建新的特征提取逻辑，而无需我手动添加到流中。在我的设想中，它可能是这样的：

data.keyBy(_).foreachValueIntheStream.flatmap(new FeatureExtractionX based on the Value)

如果我能做到这一点，添加新的特征将是添加一个带有自己状态跟踪的新特征提取类

请指导我的天真想法。我对任何指导表示感谢。

回答：

Flink 无法动态添加函数。但我想你可以做一些类似的事情。

我会使用一个广播流来处理特征路径，使用常规流来处理实际要处理的数据。将它们连接起来创建一个连接流，然后将其输入到一个 CoFlatMapFunction 中。在这个函数内部，你可以维护一个（动态生成的）特征提取函数列表，并将这些函数应用于传入的数据。对于状态，可以使用一个 Map<特征提取函数ID, 值>，这样每个特征提取函数都会在同一个映射中记录其状态。

你确实会遇到典型的问题，即希望在处理第一个数据元素之前清空广播流 – 请查看邮件列表，了解如何做到这一点的讨论。