小编Sea*_*n O的帖子

我正在尝试使用在 Kubernetes 上运行的 Apache Spark 2.3 Scala API 来扩展结构化流管道。工作的基本流程是这样的：

• 读取包含约 1,000,000 条记录的静态数据集，这些记录将各个源 ID 映射到输出聚合
• 从 Kafka 读取流数据集，其中包含要聚合的时间序列指标并映射到其源 ID
• 根据源 id 对每个 Dataset 重新分区
• 加入源 id 上的 2 个数据集（这将指标映射到正确的输出聚合，同时还从 kafka 中过滤掉不应聚合的数据）
• 应用水印
• 删除重复项
• 聚合数据
• 写入 Kafka 输出接收器

我在 Kubernetes 上运行并配置了一个集群，其中包含 30 个执行程序，每个执行程序有 3 个内核。Kafka 目前每个源 id 每秒流式传输 600000 个指标，并配置了 600 个分区。我正在尝试将它们全部聚合为 10 个不同的输出（即，每个输出聚合包含 60000 个不同的源 ID）。我每 10 秒就有一次管道触发器来处理来自 Kafka 的约 6,000,000 条记录。我的聚合窗口是 1 分钟不重叠，我的水印设置为 30 秒。理想情况下，我想要更长的水印来解释迟到的数据，但删除重复项/水印阶段似乎是一个瓶颈，尤其是在调用垃圾收集器时。以下是我最近运行的管道的一些数据：

每秒处理和输入行数

该图显示管道每秒跟上输入行大约 8-9 分钟，但随后橙色线下降到绿线以下（时间轴上的~10:01）并且管道很难跟上与输入数据速率。我查看了 Spark UI 以寻找有关为什么会发生减速的线索，并发现一个执行程序在删除重复项/水印阶段需要 55 秒来执行 GC。以下是舞台的汇总统计数据和事件时间线的放大图：

• 汇总统计
• 单 Executor GC 变慢 …