我正在编写一个流式 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)
如果我能做到这一点,添加一个新功能将添加一个具有自己的状态跟踪的新功能提取类
请大家多多指教我的幼稚想法。我很感谢您的指导。
最佳答案
Flink 无法动态添加功能。但我认为你可以做一些接近的事情。
我将使用广播流作为特征路径,并使用常规流来处理要处理的实际数据。连接它们以创建连接流,然后将其运行到 CoFlatMapFunction
。在此函数内,您将维护一个应用于传入数据的(动态生成的)特征提取函数列表。对于状态,请使用 Map<feature extraction function id, value>
,以便每个特征提取函数将其状态记录在同一个图中。
您确实遇到了一个典型问题,即希望在处理第一个数据元素之前清空广播流 - 请参阅邮件列表以获取有关如何执行此操作的讨论。
关于scala - 使用流中的值动态创建平面映射函数(键控状态),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/56026116/