在睡梦中（和飞机上）运行 Apache Spark 实验

作为 Netflix 规模最大且最关键的数据管道之一，member-sessionizer 将每小时数百亿的客户端事件（包含用户点击、播放和互动信号）整合为每个个人资料对应的独立观看会话。

原本一个为期两周的快速调优工作，在标准方法碰壁后，拖成了两个月。由于故障只在完整生产规模下才会发生，每次测试都要花上数小时，让我每天只能尝试三到四次。转折点在于改变目标：不再试图加速单次运行，而是构建一种能够同时运行和监控多个测试并追踪每一项结果的方法。

这种思路催生了一套非常规的设置：一次性启动五个 Spark 作业，在整整一天回家的旅途中，通过时断时续的机上 Wi-Fi 监控它们，而实际的重活由云端服务器承担。这篇文章会介绍我是如何搭建这个循环的，以及在此过程中发现的 Spark 瓶颈和内存怪癖。

规模困境：小样本抓不住规模级 Bug

标准的工程智慧认为，你应该在小数据集上隔离 Bug，以便快速迭代。这对常规软件 Bug 有效，但对大数据性能瓶颈则完全行不通。

在将我们的会话化管道从旧式日志架构升级到现代平台时，新作业遭遇了严重故障，包括执行器内存溢出（OOM）错误、严重的 shuffle 溢出以及数千次 stage 重试。这些问题没有一个能用采样数据复现。故障只在完整生产规模下才会触发，原因在于数据倾斜：一个闲置的 Netflix 标签页彻夜流媒体会产生数万条遥测事件，而一个典型会话只会产生寥寥几条。标准的 1% 采样完全遗漏了这些极端长尾会话。

为了抓住真正的 Bug，每次测试运行都必须针对整整一小时未经采样的生产数据执行，还要附带随之而来的全部部署开销。

策略：分支驱动的隔离

最大的收获来自 Dataflow 中的一项功能，这是 Netflix 用于构建、测试和部署数据管道的命令行工具。它支持 分支驱动开发：我们的工作流运行在 Maestro（Netflix 的工作流编排器）之上，由它来调度和执行作业，而这些工作流已经在使用经过净化的分支名（即 ${dataflow.sanitizedbranch} 占位符）。所以，如果我推送一个像 pgn/cap-experiment 这样的分支，那么所部署的工作流、它的表以及它的运行命名空间都会被打上该分支名的标记。这就消除了与团队成员之间或我自己并发运行之间的冲突。

有了这种隔离，模型就很简单：一个分支就是一个独立实验。我把调度器参数化（cap 大小、shuffle 分区数、重设备阈值、功能开关），并给每个变体分配各自的分支和输出表。与其在 10 小时内排队执行五个串行测试，我针对同一时间窗口同时启动五个变体。总执行时间从所有运行时间之和降到了仅最慢那个作业的耗时。

管理按分支划分的表没有带来任何额外开销。Dataflow 会根据声明的 schema 自动创建和更新表，无需手动 DDL。每当我在实验之间修改事件包结构时，表 schema 都会无缝更新，不需要 CREATE TABLE 或 ALTER TABLE。

唯一真正的限制是集群争用。由于这些重型 Spark 作业与公司其他部分共享资源，高峰时段有时会把我限制在只能同时运行两个，而不是五个。但即便只是并行运行两三个作业，相比单线程调试也是巨大的改进。

关键要点：不要专注于让一次长达数小时的运行变得更快，而要构建一种设置，让你能同时运行许多实验，而它们不会互相干扰。

虽然上一张图表展示了节省的时间，但实现这种隔离的架构如下图所示：

AI 智能体如何发挥作用

同时运行五个并行实验很快就会变得混乱。跨变体追踪实例 ID、配置、内存溢出和 stage 失败，很容易让人跟丢。

这正是 AI 智能体介入的地方，不是为了写巧妙的代码，而是为了处理运维上的苦活。我用平实的英语描述变体，智能体就会更新调度器配置、部署到分支、触发作​​业并轮询指标。完成后，它会在汇总表中比较各变体的溢出和失败次数，让我可以问：“哪次运行在峰值负载下最接近无 OOM？”并立即得到答案。

其价值不在于深厚的 Spark 专业知识，而在于运维记忆。它把繁琐的 Spark UI 交叉核对变成了一次简单的提示。

最终，我把这个工作流打包成一个可复用的智能体技能（.agents/skills/sandbox-deploy-and-run/），通过 /sandbox-deploy-and-run <module> <scheduler-file> [PARAM=value] 来运行。它处理完整的生命周期——预检、部署、后台轮询、自动诊断以及可重入的恢复 <instance_id> 支持——把耗时数小时的实验变成一条命令。

为什么 tmux 很重要

所有这些都运行在 Workbench 上，这是 Netflix 的云开发环境，我的笔记本电脑只充当瘦终端。如果没有会话持久性，一次 SSH 断开或笔记本休眠就会在飞行途中杀死耗时数小时的 Spark 运行。

直接在 Workbench 上运行 tmux 解决了这个问题。每个实验都存在于一个命名的云会话中：我可以合上笔记本电脑，数小时后再重新连接，从上次中断处继续。结合分支隔离，这使得“启动五个然后走开”成为可能。

现在这是我的默认设置：本地终端（Ghostty）、每个分支专用的 git worktree，以及 Workbench 上的持久 tmux 会话。本地笔记本是一次性的——合上它或断网都不会中断作业。

这种设置把漫长旅途的一天变成了活跃的迭代时间。我在登机口启动批次，通过机上时断时续的 Wi-Fi 用 AI 智能体查看结果，在单次 24 小时的飞行中挤进了 4–5 个完整实验周期。

所有实验揭示的内容

根本原因归结为单一算子：collect_list。由于 Spark 会在内存中缓存每个完整的事件包，长会话 OOM 是由单个包的大小驱动的，而不是数据集总量。旧管道处理相同的行数却没有崩溃，这证明新包中的每个事件都变得重得多。

两个因素放大了事件重量：

• 字段类型与 Option 装箱： 会话 ID 从 8 字节的 Long 变成了 UUID 字符串。更糟的是，将字段包装在 Scala Option 中迫使 Spark 将类型化的 Dataset 反序列化为 JVM 对象（Some 包装器），从而在每个会话数万个事件中推高对象分配和 GC 压力。
• 膨胀的字符串属性： 低价值字段（例如 referrer URL、测试分配）为每个事件增加了数百字节。即使下游任务很少使用它们，这些字符串在 collect_list 中相乘也会每小时浪费数 TB 的空间。

修复措施：

1. 拆箱为原始类型： 从非空字段中移除 Option 包装器，将执行器峰值内存从 槽位的 112% 降至 58%，且零磁盘溢写。
2. 精简事件包： 删除未使用的逐事件属性，并将会话不变的元数据（应用版本、国家）提升到会话头部。

要点： 降低单个事件的重量，比向问题投入更多集群内存能带来好得多的稳定性。

两个改变我想法的实验

在数十次运行中，有两个发现彻底颠覆了我对 Spark 内存管理的假设：

• 发现 1：缩窄行可能触发更快的 OOM。 精简我们的表后，其大小降至 Spark 的自动广播阈值以下，促使查询规划器从 shuffle join 切换到 broadcast join。由于 broadcast join 会将构建侧的表直接固定在执行器内存中，更窄的行实际上 增加了 总内存压力。
• 发现 2：Broadcast hash join 在悄悄消耗内存。 自适应查询执行（AQE）在运行时将九个 sort-merge join 重写为 broadcast hash join。虽然每个 join 单独都低于我们 400 MB 的广播限制，但 AQE 同时将 约 3.1 GB 的累积数据固定在单个执行器中，将容器推向崩溃边缘。

精准修复： 我没有全局禁用 broadcast join，而是仅将 AQE 的运行时阈值（spark.sql.adaptive.autoBroadcastJoinThreshold）降低到 10 MB，同时保持编译时阈值不变。这一项更改释放了约 3.1 GB 的执行器内存，且不影响初始查询规划。

要点： Broadcast join 能加速查询，但固定的内存并非免费。要关注执行计划中累积的广播，而不仅仅是数据集大小。

维护单一实验日志

在进行数十个并行测试时，最大的风险不是作业失败，而是忘记你已经尝试过什么，或者混淆了哪个 JAR 和配置产生了某个结果。

为解决这个问题，我搭建了 AI 代理，使其在每次运行后向仓库的 Markdown 表格追加一行：

• 运行 # | Jar 版本 | 与上次运行的变化 | 配置 | 关键指标 （溢写、失败任务、峰值内存、耗时）

如果某次运行产生了关键发现，代理会添加一条简短说明，解释该结果证明了或否定了什么。

关键要点：

• 单一事实来源： 它为整个调查创建了可审计的历史记录。
• 强制纪律： 要求填写 “与上次运行的变化” 迫使我隔离变量。每当我捆绑多个更改时，我就无法隔离根本原因，不得不重新运行测试——这个错误会被日志格式立即暴露出来。

一份次要文档（events-join-spark-tuning.md）将提炼后的发现记录为快速的“症状到旋钮”查找表，这样队友就不必翻阅完整的按时间顺序的历史记录。

我从中吸取的一个错误：只在非高峰时段有效的修复。

为了限制会话大小，我添加了 row_number().over(Window...) 并过滤前 N 个事件。它在非高峰测试中顺利通过，但在高峰负载时因 OOM 崩溃。

问题在于结构：Spark 的 Window 算子在评估行号和过滤之前会将整个会话缓存在内存中。这个上限截断了下游输出，但并未限制缓冲期间的内存使用，因此单个倾斜的会话就会压垮执行器内存。

修复与教训： 我们将截断移到 Window 阶段之前，以便在缓冲发生之前限制事件数量。

要点： 针对最坏情况的输入倾斜测试修复。非高峰运行会掩盖高峰规模会立即触发的结构性瓶颈。

最终修复

综合起来，共发布了四项更改，每项在上线前都通过单个实验隔离：

• 将永不为空的字段拆箱 从 Option 改为普通原始类型。执行器峰值内存从槽位的约 112% 降至约 58%，且零溢写。
• 精简事件包。 删除无人使用的逐事件字段，并将会话不变的字段（版本、国家、连接类型）提升到会话级别，使其按会话存储一次，而不是按事件存储一次。
• 在 Window 阶段之前限制会话大小。 在缓冲之前丢弃过大的会话，并将包截断到上限，而不是丢弃整个会话。
• 降低 AQE 的运行时广播阈值。 将 spark.sql.adaptive.autoBroadcastJoinThreshold 设置为 10 MB，保持编译时阈值不变。

未解决的瓶颈

虽然这些修复让管道顺利运行，但它们并未解决底层的扩展问题：将整个会话的事件包聚合到单行中有着硬性上限。随着我们上游遥测源的发展并添加新属性，包中的每个事件都会变得更重，这意味着即使我们这边不改一行代码，这种内存压力最终也会再次出现。

精简和拆箱为我们赢得了空间，但它们并未改变基本的数据形态。这项工作中最大的收获是 调优买来的是时间，而不是无限扩展。 如果你的底层数据模型允许行无限增长，仅靠调优救不了你——最终，架构本身必须改变。

总结

• 如果 bug 只在规模化时出现，就不要再去尝试抽样了。相反，要构建一种方法来并行运行许多全规模运行。
• 如果你的调度器可以使用分支名，那么一个分支就变成一个隔离的实验，同时运行许多实验就变得很容易。
• 使用 AI agent 来执行操作，而不仅仅是编写代码。部署、轮询、拉取指标、比较变体：大部分时间都花在这里。
• 一旦某个循环重复出现，就把它打包成一个 skill。部署-运行-观察-诊断的循环就变成一条命令。
• 使用 tmux 进行长时间运行。运行必须能在你合上笔记本电脑后继续存活。
• 保留单一实验日志，并让每一条记录都说明相对于上一次运行的那一处改动。这种格式本身就会迫使你一次只改变一个变量。
• 对于内存问题，首先要看每行的宽度，而不仅仅是总数据量。更大的字段类型、Option 装箱以及未被使用的臃肿字符串，都会在一个 collect_list 袋中累积起来。
• 调优换来的是余量，而不是新的上限。如果数据形态可以无界增长，那么设计最终必须改变，仅靠调优救不了你。

在我睡觉时（以及在飞机上）运行 Apache Spark 实验 最初发表于 Netflix TechBlog 在 Medium 上，人们通过高亮和回应这个故事来继续对话。