scala - 使用流中的值动态创建平面映射函数(键控状态)

Question

我正在编写一个流式 Flink 程序来为我们的离线训练模型进行特征提取，并且想知道该程序的设计。我希望每个特征提取逻辑在其类中维护自己的状态，以便添加新的特征提取相当于添加新类。

粗略的高层设计如下:

#data is the stream of relative paths to the feature extraction logic in our code e.g. com.xxx.FeatureExtraction1
val data:DataStream[String] = ...

#based on the relative path, use reflection to initiate the class
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]) = {
// access the state value
}

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)

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

请大家多多指教我的幼稚想法。我很感谢您的指导。

Answer 1

Flink 无法动态添加功能。但我认为你可以做一些接近的事情。

我将使用广播流作为特征路径，并使用常规流来处理要处理的实际数据。连接它们以创建连接流，然后将其运行到 CoFlatMapFunction 。在此函数内，您将维护一个应用于传入数据的(动态生成的)特征提取函数列表。对于状态，请使用 Map<feature extraction function id, value> ，以便每个特征提取函数将其状态记录在同一个图中。

您确实遇到了一个典型问题，即希望在处理第一个数据元素之前清空广播流 - 请参阅邮件列表以获取有关如何执行此操作的讨论。