如何在Flink键控状态下实现支持嵌套地图的快速插入，查找和迭代的地图？

如何解决如何在Flink键控状态下实现支持嵌套地图的快速插入，查找和迭代的地图？

我想写一个Flink流运算符，每个键维护1500-2000个映射，每个映射可能包含100,000个〜100B的元素。大多数记录都会触发插入和读取，但是我也想支持整个嵌套地图的偶尔快速迭代。

我编写了一个KeyedProcessFunction，每个键可创建1500个RocksDb支持的MapState，并通过使用单个键生成记录流来对其进行测试，但是我发现性能很差。仅初始化它们就需要几分钟的时间，一旦数据开始异步传输，增量检查点就会由于超时而频繁失败。这是一个合理的方法吗？如果没有，我应该考虑哪些替代方案？

谢谢！

在功能上，我的代码遵循以下原则：

val stream = env.fromCollection(new Iterator[(Int,String)] with Serializable {
  override def hasNext: Boolean = true

  override def next(): (Int,String) = {
    (1,randomString())
  }
})

stream
  .keyBy(_._1)
  .process(new KPF())
  .writeUsingOutputFormat(...)

class KFP extends KeyedProcessFunction[Int,(Int,String),String] {

  var states: Array[MapState[Int,String]] = _

  override def processElement(
    value: (Int,ctx: KeyedProcessFunction[Int,String]#Context,out: Collector[String]
  ): Unit = {
    if (states(0).isEmpty) {
      // insert 0-300,000 random strings <= 100B
    }

    val state = states(random.nextInt(1500))
    // Read from R random keys in state
    // Write to W random keys state
    // With probability 0.01 iterate entire contents of state
    if (random.nextInt(100) == 0) {
      state.iterator().forEachRemaining {
        // do something trivial
      }
    }
  }

  override def open(parameters: Configuration): Unit = {
    states = (0 until 1500).map { stateId =>
      getRuntimeContext.getMapState(new MapStateDescriptor[Int,String](stateId.toString,classOf[Int],classOf[String]))
    }.toArray
  }
}

解决方法

在您所描述的内容中，没有任何东西可以清楚地说明性能不佳。您已经在做最重要的事情，那就是使用MapState<K,V>而不是ValueState<Map<K,V>>。这样，地图中的每个键/值对都是一个单独的RocksDB对象，而不是整个地图是一个RocksDB对象，对于任何条目的每次访问/更新都必须通过ser / de。

要更好地了解性能，下一步可能是enable the RocksDB native metrics，并研究它们的线索。 RocksDB是非常可调的，并且可以实现更好的性能。例如，您可以调整预期的读写组合，如果尝试访问不存在的密钥，则应启用Bloom过滤器（默认情况下处于关闭状态）。

对于每个状态访问/更新，RocksDB状态后端必须通过ser / de，这肯定是昂贵的。您应该考虑是否可以优化序列化程序。一些串行器的速度可能是其他串行器的2-5倍。 （Some benchmarks。）

此外，您可能想研究正在开发的新的可溢出堆状态后端。请参见https://flink-packages.org/packages/spillable-state-backend-for-flink，https://cwiki.apache.org/confluence/display/FLINK/FLIP-50%3A+Spill-able+Heap+Keyed+State+Backend和https://issues.apache.org/jira/browse/FLINK-12692。早期的基准测试表明，此状态后端比RocksDB快得多，因为它可以将其工作状态保持为堆上的对象，并将冷对象洒到磁盘上。 （这有多大帮助可能取决于您必须迭代的频率。）

如果您不需要溢出到磁盘，则FsStateBackend仍会更快。