Apache Beam高基数按key计数场景下worker内存占用相关问题

现有基数超10亿唯一key的无界数据集，需按固定窗口执行按key计数操作，相关疑问解答如下：

疑问1解答 #

你对Combine函数的认知是正确的。默认配置下，Combine会在每个worker的内存中，为每个窗口内的每个key维护对应的累加器做本地预聚合。
是否支持刷写磁盘取决于你使用的Beam运行器：

• Flink、Spark等生产级运行器自带spill-to-disk（内存溢写）能力，当内存占用达到设定阈值时，会自动将未触发窗口计算的key和对应累加器序列化后写入本地磁盘，等需要处理对应key的新数据或窗口触发计算时再读回内存，不会直接出现内存溢出。
• DirectRunner这类测试用轻量运行器不支持磁盘溢写，内存不足时会直接抛出OOM异常。

疑问2解答 #

如果数据源没有做天然分区，且使用不支持内存溢写的运行器，确实存在worker内存耗尽的风险。
哪怕运行器支持溢写，也会因为全量key的序列化、落盘、读回操作产生极高的IO开销，导致作业性能严重下降。这种场景建议在Combine操作前增加一次shuffle分区步骤，将相同key的数据路由到同一个worker处理，避免每个worker都维护全量key的累加器。

疑问3解答 #

这种场景下worker的内存占用存在差异，差异完全由Kafka分区的key分布均匀度决定：
如果所有Kafka分区的唯一key基数接近，各worker需要维护的累加器数量差异很小，内存占用基本持平。如果存在数据倾斜，某一个Kafka分区内的唯一key基数远高于其他分区，对应worker的内存占用就会明显更高。
另外该场景因为省去了跨worker的shuffle开销，且每个worker只需维护自身分区内的key累加器，整体内存压力会远小于非分区数据源的场景。

内容的提问来源于stack exchange，提问作者Luke De Feo