Nielsen Marketing Cloud 是一家领先的广告技术公司，在其某个管道中每天处理 25 TB 数据和 300 亿个事件。随着数据量的增长，高效扩展 Apache Spark 工作负载的挑战也随之增加。

Nielsen 的团队面临这样一种情况：当他们通过添加更多实例来扩展集群时，每个实例的性能反而下降。这种性能下降导致每个实例每小时完成的工作量减少，并推高了每 GB 数据处理成本。

此外，他们还偶尔遇到数据倾斜问题。数据倾斜是指数据在分区之间分布不均，会造成处理瓶颈并进一步降低集群效率。在极端情况下，这些因素叠加会导致集群故障。

在本文中，我们将跟随 Nielsen 的历程，了解他们如何构建稳健且可扩展的架构，同时实现线性扩展。我们首先审视 Nielsen 最初面临的挑战以及这些问题背后的根本原因。然后，我们探讨 Nielsen 的解决方案：在 Amazon Elastic Kubernetes（容器编排平台） Service（Amazon EKS）上运行 Spark，同时采用无服务器概念。

从 Spark 集群演进为 Amazon EKS 上的 Spark Pod

Nielsen 的 Marketing Cloud 架构最初是 Amazon EMR 上的典型 Spark 集群，接收源源不断的不同大小的文件进行处理。随着数据量和集群规模的增长，团队注意到每个实例的性能出现下降，如下图所示。除了处理速度变慢和成本上升之外，Nielsen 偶尔还会因数据倾斜而遭遇生产问题。

团队意识到问题在于随着集群增长，实例之间的远程 shuffle 越来越多。远程 shuffle 是 Spark 中将数据重新分布到各分区的过程，涉及大量网络数据传输，可能成为主要瓶颈。由于其场景中数据的流式特性，Nielsen 意识到他们可以改为以更小的批次处理数据。这意味着他们不必依赖 Spark 的分布式处理能力来使用大型 Spark 集群，而是可以选择小型集群。

为了解决性能下降问题，团队决定改变其扩展策略：不再纵向扩展单个 Spark 集群，而是通过 Amazon EKS 上运行的多个本地模式 Spark 集群（单节点集群）进行横向扩展。与 Spark 集群模式相比，本地模式为小型分析工作负载提供更好的性能。每个本地模式运行有限且更少量的数据，无需远程 shuffle，也无需与其他 Spark 实例交互。

此外，Amazon EKS 上运行的 Pod 可以根据待处理工作量进行扩展和缩减，这意味着 Nielsen 可以在不需要资源时停止使用它们。

新解决方案可线性扩展，成本降低 55%，数据处理速度更快，即使在大量突发条件下也是如此。

为什么 shuffle 很重要

当数据需要在 Spark 实例之间交换时，就会触发远程 shuffle。某些转换操作，如 join 或 repartition，需要对数据进行 shuffle。远程 shuffle 比内存计算慢一个数量级，因为它需要通过网络传输数据。它可能会显著降低处理速度，有时会使总处理时间增加 100–200%。

Nielsen 遇到的问题在于，随着集群规模的增长，shuffle 的数据量也按集群规模成比例增长。下图展示了这种情况发生的原因。它计算了随着集群规模增长，随机分布数据集所交换的数据量。

下图表明，这种相关性是与集群规模相关，而不是与数据规模相关。

解决 shuffle 问题

团队假设，最小化 shuffle 可以带来显著的性能提升。Nielsen 的工程师决定实施无服务器模式（serverless patterns）中的思路，将每个集群的规模大幅缩减至最小，同时增加更多这样的小型集群，以弥补每个集群容量较低的不足。这种方法有望为每个数据工作项完全消除远程 shuffle，如前面的图表所示。

尽管这一策略有望带来性能提升，但它也引入了一个限制：每个工作项的数据量上限。

基于无服务器模式设计新系统

Nielsen 的团队开发了一种新架构，使用了两个核心概念：

一个可供拉取的工作项队列
一组从队列中拉取工作项的本地模式 Spark 模块

他们有以下设计目标：

让 Spark 模块始终保持忙碌
在不需要时停止模块
确保所有工作项都被成功处理

下图展示了该工作流。

最终设计

最终设计包含以下组件：

文件元数据存储 – 一个 Amazon Relational Database Service (Amazon RDS) 集群运行 PostgreSQL（开源关系型数据库）引擎，用于存储和管理进入系统的每个文件的统计信息。
工作管理器 – 使用一个 AWS Lambda 函数定期从数据库中拉取等待处理的文件，准备由一个或多个文件组成的工作项，并将工作项发布到 Amazon Simple Queue Service (Amazon SQS) 消息队列。
工作队列 – 使用一个 SQS 消息队列存放等待被拉取处理的工作项。
处理单元 – 本地模式 Spark 实例作为 pod 运行在 EKS 集群上。它们从 SQS 队列中拉取工作项。只要队列中还有等待处理的工作项，这些 pod 就会持续忙碌。
指标适配器 – 一个适配器（Kubernetes-cloudwatch-adapter）为 Kubernetes Horizontal Pod Autoscaler 提供 Amazon CloudWatch 指标。
Kubernetes Horizontal Pod Autoscaler – Horizontal Pod Autoscaler (HPA) 使用扩缩容规则，根据 CloudWatch 的指标对 pod 进行扩容或缩容。它根据队列中可见的消息（工作项）数量进行扩缩容，该数量与等待处理的工作量成正比。在 Nielsen 的系统中，HPA 按照 targetValue = {SQS length/2} 对 pod 进行扩缩容。
工作完成队列 – 使用第二个 SQS 消息队列报告工作项的完成情况。完成信息由另一个 Lambda 函数拉取，并更新到 PostgreSQL 数据库中。

下图展示了最终系统的架构。