Apache Flink - 实现具有潜在非常大状态的流处理器

Question

我希望从事件流中投射出一个潜在的非常大的状态。这就是我以命令式方式实现它的方式：

class ImperativeFooProcessor {
  val state: mutable.Map[UUID, BarState] = mutable.HashMap.empty[UUID, BarState]
  def handle(event: InputEvent) = {
    event match {
      case FooAdded(fooId, barId) => {
        // retrieve relevant state and do some work on it
        val barState = state(barId)
        // let the world know about what may have happened
        publish(BarOccured(fooId, barId))
        // or maybe rather
        publish(BazOccured(fooId, barId))
      }
      case FooRemoved(fooId, barId) => {
        // retrieve relevant state and do some work on it
        val barState = state(barId)
        // let the world know about what may have happened
        publish(BarOccured(fooId, barId))
        // or maybe rather
        publish(BazOccured(fooId, barId))
      }
    }
  }
  private def publish(event: OutputEvent): Unit = {
    // push event to downstream sink
  }
}

在最坏的情况下，BarState的规模将随着FooAdded提及的次数而增长

相对于每个 barId 的事件总数，唯一 barId 的数量非常小。

我将如何开始在 Flink 中表示这种处理结构？

我如何处理每个 BarState 可能会变得非常大的事实？

Answer 1

Flink 在所谓的状态后端维护状态。有状态后端(MemoryStateBackend和FsStateBackend(在工作进程的JVM堆上运行。这些后端不适合处理大型状态。

Flink 还具有基于 RocksDB 的 RocksDBStateBackend。RocksDB 在每个工作节点上用作本地数据库(无需将其设置为外部服务(，并将状态数据写入磁盘。因此，它可以处理超过内存的非常大的状态。

Flink 提供了一个KeyedStream，它是一个在某个属性上分区的流。在您的情况下，您可能希望对同一 id 的所有访问都转到同一状态实例，因此您将使用 barId 作为键。然后，根据barId在所有并行工作线程之间对状态进行分区。这基本上是一个分布式键值存储或映射。所以你不需要将状态表示为地图，因为它是由 Flink 自动分发的。