如何在Apache Flink中使用大文件丰富事件流？

Question

我有一个用于点击流收集和处理的Flink应用程序。该应用程序由Kafka作为事件源、一个html" target="_blank">map函数和一个接收器组成，如下图所示：

我想根据从Kafka摄取的原始事件中的userIp字段，使用用户的IP位置来丰富传入的点击流数据。

CSV文件的简化切片，如下所示

   start_ip,end_ip,country
   "1.1.1.1","100.100.100.100","United States of America"
   "100.100.100.101","200.200.200.200","China"

我做了一些研究，发现了一些潜在的解决方案：

1.解决方案：广播浓缩数据，用一些IP匹配逻辑连接事件流。

结果：它适用于几个示例IP位置数据，但不适用于整个CSV数据。JVM堆已达到3.5 GB，由于广播状态，无法将广播状态放入磁盘（对于RocksDb）

2、解决方案：在开始事件处理之前，将RichFlatMapFunction中的open（）方法中的CSV数据加载到状态（ValueState），并在flatMap方法中丰富事件数据。

结果：由于扩展数据太大，无法存储在JVM堆中，因此无法加载到ValueState。而且，对于键值性质的数据，通过ValueState进行反序列化也是一种不好的做法。

3.解决方案：为了避免处理JVM堆约束，我尝试将富集数据作为MapState的状态放入RocksDB（使用磁盘）中。

3、结果：试图在open（）方法中将CSV文件加载到MapState中，给了我一个错误，告诉我不能在open（）方法中放入MapState，因为我不在open（）方法中的键控上下文中，如以下问题：Flink键控流键为null

4.解决方案：由于需要MapState的键控上下文（放入RocksDB），我在将DataStream放入KeyedStream后，尝试将整个CSV文件加载到进程函数中的本地RocksDB实例（磁盘）中：

class KeyedIpProcess extends KeyedProcessFunction[Long, Event, Event] {

  var ipMapState: MapState[String, String] = _
  var csvFinishedFlag: ValueState[Boolean] = _

  override def processElement(event: Event,
                              ctx: KeyedProcessFunction[Long, Event, Event]#Context,
                              out: Collector[Event]): Unit = {

    val ipDescriptor = new MapStateDescriptor[String, String]("ipMapState", classOf[String], classOf[String])
    val csvFinishedDescriptor = new ValueStateDescriptor[Boolean]("csvFinished", classOf[Boolean])

    ipMapState = getRuntimeContext.getMapState(ipDescriptor)
    csvFinishedFlag = getRuntimeContext.getState(csvFinishedDescriptor)

    if (!csvFinishedFlag.value()) {
      val csv = new CSVParser(defaultCSVFormat)

      val fileSource = Source.fromFile("/tmp/ip.csv", "UTF-8")
      for (row <- fileSource.getLines()) {
        val Some(List(start, end, country)) = csv.parseLine(row)
        ipMapState.put(start, country)
      }
      fileSource.close()
      csvFinishedFlag.update(true)
    }

    out.collect {
      if (ipMapState.contains(event.userIp)) {
        val details = ipMapState.get(event.userIp)
        event.copy(data =
          event.data.copy(
            ipLocation = Some(details.country)
          ))
      } else {
        event
      }
    }
  }
}

4.结果：太hacky，由于阻止文件读取操作而阻止事件处理。

你能告诉我这种情况我能做些什么吗？

谢谢




Answer 1

您可以做的是实现一个自定义分区器，并将一部分丰富数据加载到每个分区中。这里有一个这种方法的示例；我将摘录一些关键部分：

工作是这样组织的：

java prettyprint-override">DataStream<SensorMeasurement> measurements = env.addSource(new SensorMeasurementSource(100_000));

DataStream<EnrichedMeasurements> enrichedMeasurements = measurements
  .partitionCustom(new SensorIdPartitioner(), measurement -> measurement.getSensorId())
  .flatMap(new EnrichmentFunctionWithPartitionedPreloading());

自定义分区器需要知道有多少个分区，并确定地将每个事件分配给特定的分区：

private static class SensorIdPartitioner implements Partitioner<Long> {
    @Override
    public int partition(final Long sensorMeasurement, final int numPartitions) {
        return Math.toIntExact(sensorMeasurement % numPartitions);
    }
}

然后，富集函数利用知道分区是如何完成的来仅将相关切片加载到每个实例中：

public class EnrichmentFunctionWithPartitionedPreloading extends RichFlatMapFunction<SensorMeasurement, EnrichedMeasurements> {

    private Map<Long, SensorReferenceData> referenceData;

    @Override
    public void open(final Configuration parameters) throws Exception {
        super.open(parameters);
        referenceData = loadReferenceData(getRuntimeContext().getIndexOfThisSubtask(), getRuntimeContext().getNumberOfParallelSubtasks());
    }


    @Override
    public void flatMap(
            final SensorMeasurement sensorMeasurement,
            final Collector<EnrichedMeasurements> collector) throws Exception {
        SensorReferenceData sensorReferenceData = referenceData.get(sensorMeasurement.getSensorId());
        collector.collect(new EnrichedMeasurements(sensorMeasurement, sensorReferenceData));
    }

    private Map<Long, SensorReferenceData> loadReferenceData(
            final int partition,
            final int numPartitions) {
        SensorReferenceDataClient client = new SensorReferenceDataClient();
        return client.getSensorReferenceDataForPartition(partition, numPartitions);
    }

}

请注意，富集不是在键控流上完成的，因此您不能在富集函数中使用键控状态或计时器。