我可以使用 Flink 状态执行加入吗？

Question

我正在评估Apache Flink的流处理，作为Apache Spark的替代品/补充。我们通常使用Spark解决的任务之一是数据扩充。

也就是说，我有来自物联网传感器的带有传感器ID的数据流，并且我有一组传感器元数据。我想将输入流转换为传感器测量传感器元数据流。

在星火中，我可以和RDD一起加入数据流。

case calss SensorValue(sensorId: Long, ...)
case class SensorMetadata(sensorId: Long, ...)
val sensorInput: DStream[SensorValue] = readEventsFromKafka()
val staticMetadata: RDD[(Long, SensorMetadata)] =
  spark.read.json(...).as[SensorMetadata]
 .map {s => (s.sensorId, s)}.rdd
val joined: DStream[(SensorValue, SensorMetadata)] = 
  sensorInput.map{s => (s.sensorId, s)}.transform { rdd: RDD[SensorValue] => 
  rdd.join(staticMetadata)
     .map { case (_, (s, m)) => (s, m) } // Get rid of nested tuple
}

我可以用Apache Flink做同样的技巧吗？我在这方面没有看到直接的API。我唯一的想法是使用有状态转换——我可以在单个流中合并元数据和传感器事件，并使用Flink状态存储来存储元数据（伪代码）：

val sensorInput: DataStream[SensorValue] = readEventsFromKafka()
val statisMetadata: DataStream[SensorMetadata] = readMetadataFromJson()
val result: DataStream[(SensorValue, SensorMetadata)] =
  sensorInput.keyBy("sensorId")
 .connect(staticMetadata.keyBy("sensorId"))
 .flatMap {new RichCoFlatMapFunction() {
   private val ValueState<SensorMetadata> md = _;
   override def open = ??? // initiate value state
   def flatMap1(s: SensorEvent, s: Collector(SensorEvent, SensorMetadata)) = 
      collector.collect(s, md.value) 
   def flatMap2(s: SensorMetadata, s: Collector[(SensorEvent, SensorMetadata)]) = 
   md.update(s)  
 }}

这是正确的方法吗？当元数据不适合一台机器时，我可以在更大的范围内使用它吗？

谢啦

Answer 1

使用CoFlatMapFunction连接是一种常见的方法。但是，它有一个明显的缺点。每当任何一个输入的元组到达时都会调用该函数，并且您无法控制首先使用哪个输入。因此，在开始时，您必须在元数据尚未完全读取时处理传感器事件。一种方法是缓冲一个输入的所有事件，直到另一个输入被消耗。另一方面，CoFlatMapFunction方法的好处是您可以动态更新元数据。在您的代码示例中，两个输入都被键入连接键。这意味着输入是分区的，每个任务槽都在处理不同的键集。因此，您的元数据可能比机器可以处理的要大（如果您配置RocksDB状态后端，状态可以持久化到磁盘，因此您甚至不受内存大小的限制）。

如果您要求作业开始时所有元数据都必须存在，并且如果元数据是静态的（它不会改变）并且足够小以适合一台机器，您还可以使用常规的FlatMapFunction并从文件中加载open（）方法中的元数据。与您的方法相反，这将是一个广播连接，其中每个任务槽都在内存中拥有完整的元数据。除了在事件数据被消耗时所有元数据可用之外，该方法的好处是您不需要洗牌事件数据，因为它可以在任何机器上连接。