题干与适用场景
一个 Spark SQL 日任务把 4.8 TB 事件事实表与 180 GB 商品维表按 product_id 做左连接。 上游发布后，32% 的事件被统一映射为 product_id='UNKNOWN'。作业配置了 2,000 个 shuffle 分区；Spark UI 显示该连接 Stage 的任务 shuffle read 中位数为 1.1 GiB，一个任务却读取 720 GiB，持续发生磁盘 spill，并在重试后 OOM。作业耗时从 24 分钟增长到 96 分钟，业务要求 45 分钟内完成，同时不能丢弃未知商品事件或改变左连接结果。
表大小、key 占比、分区指标、耗时和 SLA 都是面试假设。题目的核心是用分区级证据区分数据倾斜、 资源不足和连接结果膨胀，再依据数据语义选择修复。它归入 data，因为主要考察 Spark 执行计划、 shuffle 分区、数据分布与批处理正确性。现有 Kafka 热分区题关注消息分区键、顺序和消费位点； 本题关注 SQL 运行时分区、join 策略、AQE 与结果守恒，故障层和验证方法都不同。
这类题也适用于 groupBy、distinct、窗口函数或其他宽依赖：只要同一 key 的大量记录在 shuffle 后集中到少数任务，就可能形成长尾、spill、GC 压力或 OOM。答案不能从“加内存”开始， 因为必须先证明最慢任务是否承担了不成比例的数据与计算。
面试官考察点
首先看证据粒度。强回答会从 Job 进入 SQL 查询和具体 Stage，再比较每个 task 的 duration、 shuffle read records/bytes、spill、peak execution memory 与 GC time。一个 task 比中位数读取几百倍 数据，并且重试到其他 executor 后仍处理同一巨量分区，才支持确定性数据倾斜。只看 executor 总内存或作业总耗时无法完成归因。
执行计划理解同样关键。EXPLAIN FORMATTED 用来确认 Exchange、连接类型和物理 join； EXPLAIN COST 与 SQL UI 的运行时统计用来检查估算和实际数据量。Spark 的 AQE 会利用运行时统计 调整计划。Spark 4.2.0 文档中，skew join 优化会拆分倾斜的 sort-merge join 分区，并在需要时复制 较小一侧。默认判定同时要求分区大于中位数的 5 倍和 256 MiB；实际回答应检查运行环境的有效配置， 不能把文档默认值当成所有集群的固定事实。
数据语义是分水岭。UNKNOWN 可能代表合法的“无法归因”事件，也可能由上游错误填充。 删除、随机分散或改写这些记录都会影响结果。加盐 join 必须只复制维表中的热点 key，并给事实表 热点记录分配确定性 salt；给所有维表记录做笛卡尔复制会扩大数据量，给两侧独立随机加盐会漏掉连接。
方案选择也能体现回答深度。增加 spark.sql.shuffle.partitions 会产生更多 hash 桶，但同一个热点 key 仍进入同一桶。广播只适合经过列裁剪、过滤和统计后确实能安全放入每个 executor 的一侧；180 GB 维表不应靠提示强行广播。AQE 是低侵入首选，显式加盐用于热点 key 稳定、自动优化未生效或仍达不到 SLA 的情况；聚合类作业则常用热点 key 分片预聚合后再合并。
最终要看验证完整性。性能修复必须同时证明行数、业务金额、未知 key、未匹配率和重复率没有 变化。只展示作业从 96 分钟降到 40 分钟，仍无法证明结果正确，也无法解释下一次 key 分布变化时 方案是否继续有效。
回答前需要澄清的问题
- 慢点位于 scan、shuffle write 还是 shuffle read 后？ scan task 不均衡可能来自超大文件或不可切分输入；本题的异常位于 join 后的 shuffle read，才进入 key 倾斜路径。
- 32% 指记录数、压缩字节还是处理成本？ 宽行、复杂 UDF 或结果扇出会造成成本倾斜。需要同时比较记录、字节、耗时和输出行数。
UNKNOWN的业务语义是什么？ 若未知商品无需维度属性，可以把它们从 join 主路径分离后按原语义补齐空值；若必须匹配维表中的 sentinel 行，则应保留连接并拆分该热点。- 商品维表在
product_id上是否唯一？ 若UNKNOWN在维表中有多行，事实表热点与维表重复会形成多对多结果膨胀。修复前必须区分分区倾斜和错误基数。 - 当前 Spark 版本和 AQE 有效配置是什么？ 需要从 Environment 页和最终 adaptive plan 核对开关、阈值、join 类型与是否真的执行 skew split。
- 180 GB 是原始大小还是裁剪后的连接输入？ 如果只保留连接键和两个属性后变为可安全广播的小表，广播可能优于双边 shuffle；必须用运行时统计和 executor 内存预算证明。
- 结果的守恒条件有哪些？ 至少明确总行数、唯一事件数、金额等可加总指标、未知 key 行数、未匹配行数和允许的重复语义。
- 热点 key 是否稳定且可枚举？ 稳定的少数热点适合定向加盐；每天变化的长尾分布更适合 AQE、动态热点检测或上游语义修复。
30 秒回答框架
“我先在 SQL UI 比较 join task 的 shuffle read、spill、GC 和重试位置。720 GiB 对 1.1 GiB 中位数， 加上 UNKNOWN 占 32%，支持 key 倾斜；我还会检查维表唯一性，排除结果膨胀。先验证 AQE skew join 是否拆分大分区；仍超过 45 分钟时，用稳定 event_id 对 UNKNOWN 定向加盐，只复制 sentinel 行。 最后对照基线核验行数与金额，再比较最大/中位 task 输入、Stage 时间和成本。”
分步骤深入解答
第一步：把 96 分钟拆到具体 Stage 和 task
先保留 event log，使用 Spark History Server 查看同一数据日期的正常版和异常版。沿 SQL 查询进入 Stage 详情，记录 task duration 的分位数和最大值，并对齐 shuffle read records/bytes、shuffle spill、peak execution memory、GC time、失败原因和 executor。SQL 计划中确认左连接前是否有 Exchange hashpartitioning(product_id, 2000)，最终物理算子是否为 sort-merge join，以及 adaptive plan 是否完成。
本题中，同一 task 在不同 executor 上重试仍读取约 720 GiB，而大多数 task 约 1.1 GiB。该证据 指向输入分区本身。若慢任务读取量普通、只在一台 executor 上出现高 GC 或磁盘等待，应先排查节点； 若所有 task 都均匀 spill，重点是整体分区大小和资源预算；若输出行数突然倍增，先检查维表重复和 连接条件。
第二步：用 key 分布与连接基数证明原因
在与生产相同的过滤和 key 规范化之后统计热点。不能只查原始列，因为 trim、大小写转换、 coalesce 或 UDF 可能把多个值合并成同一 key。大表上应利用已有统计、抽样或受控聚合，避免为 诊断再制造一个无边界作业。下面的查询假设事实表已有或预先计算了 payload_bytes，用于同时检查 记录和字节倾斜；没有该列时应从存储统计或受控序列化估算字节。查询表达了需要核对的口径：
SELECT
COALESCE(product_id, '<NULL>') AS join_key,
COUNT(*) AS row_count,
SUM(payload_bytes) AS payload_bytes
FROM fact_events
WHERE event_date = DATE '2026-07-17'
GROUP BY COALESCE(product_id, '<NULL>')
ORDER BY row_count DESC
LIMIT 20;同时验证维表中每个 product_id 最多一行，并对比 join 前后的事实事件数。对于本题的左连接，维表 key 唯一时，每个事实事件应恰好输出一行；未匹配事件仍保留。若 UNKNOWN 有 32% 的事实记录且 维表只有一条 sentinel 记录，输出没有倍增，720 GiB 的单分区就能由 hash 分区规则解释。
第三步：先处理语义根因，再选执行层手段
检查上游发布为何把 32% 映射成 UNKNOWN。若它是回归，应回滚或修复映射，并重跑受影响分区； 这同时恢复数据质量和性能。若它是合法业务值，执行层仍要承载该分布。
合法但无需商品属性的未知事件可以分流：冷 key 正常连接，未知 key 按原契约直接补齐维度空值， 最后 unionByName。这消除了没有信息增益的 shuffle。若业务要求 UNKNOWN 匹配 sentinel 属性， 则保留 join，使用 AQE 或定向加盐拆分工作。选择由结果语义决定，不能为了均匀分区随意删除记录。
第四步：按成本从低到高选择修复
先核对 AQE。Spark 4.2.0 中 spark.sql.adaptive.enabled 和 spark.sql.adaptive.skewJoin.enabled 默认开启，但集群、作业或平台可能覆盖。skew join 判定的倍数 阈值与绝对字节阈值必须同时满足。查看最终 adaptive plan 是否出现 skew 处理，并确认 Stage 指标确实 拆开大分区。AQE 对符合条件的 sort-merge join 可拆分大侧、复制较小侧；它能跟随每天变化的分布， 但可能增加 shuffle 与复制成本，也不能修复错误的多对多连接。
如果裁剪后的维表有可靠统计且足够小，可以评估 broadcast hash join，以免事实表按 join key shuffle。原始 180 GB 明显不符合常规广播预算；强制 hint 可能让每个 executor 内存不足。需要用 投影后的字节数、并发 task、executor heap 与广播超时共同判断。
AQE 未生效或仍不达标时，对稳定热点做手工加盐。下面的代码假设 event_id 稳定且唯一、维表 product_id 唯一，并且只有 UNKNOWN 需要拆分。事实表热点记录被确定性分到 32 个 salt；维表只把 对应的一条热点记录复制 32 份。冷 key 保持 salt 0，因此没有全表复制。
from pyspark.sql import functions as F
SALT_BUCKETS = 32
HOT_KEYS = ["UNKNOWN"]
events_salted = events.withColumn(
"salt",
F.when(
F.col("product_id").isin(*HOT_KEYS),
F.pmod(F.xxhash64("event_id"), F.lit(SALT_BUCKETS)).cast("int"),
).otherwise(F.lit(0)),
)
salt_values = spark.range(SALT_BUCKETS).select(
F.col("id").cast("int").alias("salt")
)
products_hot = (
products.filter(F.col("product_id").isin(*HOT_KEYS))
.crossJoin(salt_values)
)
products_cold = (
products.filter(~F.col("product_id").isin(*HOT_KEYS))
.withColumn("salt", F.lit(0))
)
products_salted = products_cold.unionByName(products_hot)
result = (
events_salted.join(products_salted, ["product_id", "salt"], "left")
.drop("salt")
)32 只是面试假设下的起始候选。桶数应由热点分区字节、目标 task 大小、可用并行度与小侧复制成本 推导，再用代表性数据测试。桶过少仍有长尾，桶过多会增加调度、文件和复制开销。单纯执行 repartition(4000, "product_id") 仍把 UNKNOWN 放入一个分区，无法获得同样效果。
对于 groupBy(productid)，通常无需复制维表：先按 (productid, salt) 做局部聚合，再按 product_id 合并部分结果。可结合交换律与结合律安全拆分的 sum、count、min、max 较适合； 精确 median、有顺序要求的聚合或不可结合 UDF 需要不同算法，不能机械套用。
第五步：把正确性、性能和成本放进同一验收
在固定输入快照上并行运行基线和候选方案。正确性先行：比较总输出行数、唯一 event_id、 UNKNOWN 行数、未匹配行数、按业务维度聚合的金额与计数，并抽取热点、冷 key、空值和重复维表 key 样本做逐行 diff。左连接每个事实事件输出一行的前提是维表 key 唯一，这项约束要单独报警。
性能侧比较 join Stage 的 p50、p95 与最大 task duration，最大/中位 shuffle read，比对 spill、 GC、OOM、task retry、Stage 时长与总作业时长。成本侧记录 executor-hours、shuffle bytes 和输出文件 数量。达到 45 分钟只是一项门槛；若时长达标但 shuffle 翻倍、结果不守恒或换一天数据又出现新热点， 方案仍不合格。
发布时先对一个历史日期回放，再对新数据日期 shadow 运行并比对结果。热点 key 集合、最大/中位 task 输入比和未知 key 占比应进入监控。上游发布后 UNKNOWN 从基线上升到 32% 的变化需要数据质量告警， 让语义回归在拖慢作业前被发现。
高质量示范回答
“我先把这个问题定位到 SQL 查询中的具体 join Stage。Spark UI 已经给出很强的倾斜信号：2,000 个 task 的 shuffle read 中位数为 1.1 GiB，单个 task 是 720 GiB，并且重试到其他 executor 后仍然处理 同一大分区。接下来我会查最终 adaptive plan、task spill 和 GC，再按连接前的规范化逻辑统计 key 分布。维表也要做唯一性检查；如果 UNKNOWN 有多条维度记录，当前症状还包含 join 结果膨胀。
假设维表 key 唯一、UNKNOWN 占事实表 32%，那么同 key 经 hash shuffle 集中到一个分区足以解释 长尾。增加分区数只增加其他 hash 桶，无法拆开这个 key；提高 executor 内存最多延后 OOM。我的第一 个动作是检查上游映射是否回归。如果 UNKNOWN 不需要维度属性，我会把它从 join 主路径分离并按 原左连接语义补空值。若它必须匹配 sentinel 行，我会先确认 AQE 的 skew join 在最终计划中实际生效， 因为它可以按运行时统计拆分倾斜 sort-merge join 分区并复制小侧。
如果 AQE 后仍超过 45 分钟，我会做定向加盐。用稳定 event_id 把 UNKNOWN 事实记录确定性分到例如 32 个 salt，只复制维表的 UNKNOWN 行 32 份；其他 key 的 salt 都为 0。这样保持每个事件只匹配一条 维度记录，同时把热点工作拆到多个 task。桶数由热点字节与目标 task 大小推导，不固定写死。
验证时我会在同一输入上比较基线：总行数、唯一事件数、未知与未匹配记录、金额聚合都必须一致； 然后看最大/中位 shuffle read、task 长尾、spill、OOM、Stage 时间、executor-hours 和输出文件数。 最后用一个历史日期回放和一个新日期 shadow 运行，并对 UNKNOWN 占比和新热点告警。这样能同时证明 作业在 45 分钟内完成、结果未变，并且上游分布再次变化时可以及时发现。”
常见错误
- 直接把 shuffle 分区从 2,000 调到 8,000 → 同一热点 key 仍散列到一个分区，其他 task 反而更碎 → 先确认 key 分布，再用 AQE、语义分流或热点加盐拆开该 key。
- 只提高 executor memory → 它扩大单 task 的容忍量，却没有减少 720 GiB 输入和长尾 → 先降低最大分区工作量，再根据测得的 task 大小配置资源。
- 看到一个慢 task 就宣布数据倾斜 → 节点故障、GC、远端 fetch 或慢 UDF 也会制造长尾 → 比较 task 输入、重试位置、spill、GC 和执行计划。
- 给事实表和维表各自随机加盐 → 两侧 salt 对不上会漏连接，重复尝试还可能制造不确定结果 → 事实侧使用稳定行 ID，维表侧枚举同一组 salt。
- 把整张维表复制 32 份 → 180 GB 维表会带来巨量网络与内存成本 → 只复制已确认的热点 key 行，冷 key 保持 salt 0。
- 强制广播 180 GB 维表 → 每个 executor 都要承担广播数据，可能直接 OOM → 先裁剪和统计，只有在内存与并发预算证明安全后才广播。
- 过滤
UNKNOWN以让作业变快 → 输出语义和下游指标发生变化 → 先确认未知事件的业务契约，分流时也保持左连接结果。 - 只比较总作业耗时 → 性能提升可能来自少算、重复或丢数据 → 先做行数和业务守恒，再比较 task 分布、成本与 SLA。
追问及应对
AQE 已开启，为什么倾斜 join 仍没有被拆分？
先看最终 adaptive plan 和有效配置，确认 spark.sql.adaptive.enabled、skew join 开关、倍数阈值与 绝对字节阈值。两个阈值需要同时命中。再确认物理 join 是否属于 AQE 支持的路径、运行时统计是否 可用，以及计划是否被 hint 或平台配置限制。可以在代表性数据上调整阈值或评估强制 skew 优化， 同时量化新增 shuffle；若计划结构无法受益，再使用定向加盐。只看到配置文件中的 true 不能证明 运行计划已经拆分。
如果裁剪后的维表只有 6 GiB，可以直接广播吗？
6 GiB 仍需结合 executor heap、并发 task、序列化大小、广播超时和集群稳定性判断。广播可以消除 大侧按 join key 的 shuffle，因此可能自然避开该热点；它也会把维表复制到 executor。先用统计证明 实际广播字节，在小规模与生产形态压力测试中观察 peak memory 和 GC，再决定是否使用 hint。不能只 根据“比事实表小很多”得出可广播结论。
如果热点每天变化，手工维护 HOT_KEYS 怎么办？
优先让 AQE根据运行时统计处理。若仍需显式加盐，可以在主作业前从受控统计中产生达到记录数、字节 或成本阈值的热点 key 小表，并广播该小表决定是否加盐；热点列表要有数据日期、阈值、数量上限和 回退策略。动态方案增加一个规划阶段和运维状态，所以要用稳定的 SLA 收益证明它值得存在。
若慢操作是 groupBy，还需要复制维表吗？
不需要。对可结合聚合，先给热点记录加 salt，按 (key, salt) 做第一阶段部分聚合，再按 key 合并。 这把一个热点 key 的输入分给多个 task，第二阶段只处理少量部分结果。回答时应说明聚合函数是否满足 结合要求；依赖全局顺序或无法合并的状态不能直接采用该方法。
如何选择 32 个 salt bucket？
先用热点分区字节除以目标 task 输入量得到下界，再受可用核心数、小侧复制成本、调度开销和输出文件 约束。假设 720 GiB 希望降到约 32 GiB，理论下界约为 23，可以从 32 测试；这只是本题推导示例。 最终应比较多个候选的最大 task 输入、Stage 时间和 executor-hours，并给增长中的热点留出有限余量。
开启 speculative execution 能解决这个长尾吗？
确定性倾斜 task 的副本仍读取同一个 720 GiB 分区，通常只是让两台 executor 重复昂贵工作。推测执行 更适合偶发慢节点或不可重复的抖动。先看同一 task 重试到其他 executor 后是否仍慢；若输入量始终 异常，就应拆分数据工作，而非复制该工作。