企业官网建设流程全解析

集合操作用得溜不溜，直接决定了你写的 Spark 作业是跑十分钟还是跑一个小时。很多人入门 Spark 的时候，map、filter这类基础算子用得飞起，一旦碰到union、join、cartesian这些涉及多个数据集的操作，就开始凭感觉写，结果要么结果不对，要么集群直接 OOM。这篇就围绕交并集、笛卡尔积、Join 这三个最常用的集合操作，把事情讲透。

先说清楚这篇文章适合谁。如果你已经会用 RDD 或 DataFrame 做基本的 map/filter，但对集合类操作的底层行为、性能特征、坑点边界还不清楚，那这篇文章就是给你写的。我尽量少贴大段源码，多讲为什么这么设计、什么时候用哪个、踩过哪些坑，所有例子都用 Scala 写，但因为 Spark 的 API 设计高度统一，换成 Python 也只是语法差异，思想完全通用。

1. 集合操作的本质与两套 API 的设计差异

1.1 集合操作在数据加工里的定位

做数据处理，单个数据集内部的变换只是基础，真正的业务逻辑往往发生在数据集和数据集之间：两个用户表要合并、两份订单日志要找共同用户、用户表和订单表要按用户 ID 关联、商品和商品之间要比对相似度。这些场景对应到集合论里就是交集、并集、差集、笛卡尔积、连接。

这些操作看起来都是数学课上讲过的东西，但放到分布式系统里，性质完全变了。单机内存里的集合操作，直接把数据拉一起算就行；分布式环境下，数据分散在几十上百个 Executor 上，两个数据集之间的操作意味着数据要跨节点传输，这就是 shuffle。shuffle 是整个 Spark 作业里最贵的操作，没有之一。所以同样的一个intersection，在本地集合上做是纳秒级，在 Spark 上做可能是分钟级，理解这一点是学好集合操作的第一步。

1.2 RDD 与 DataFrame 两套 API 的关系

Spark 早期只有 RDD 一套 API，后来推出 DataFrame/DataSet。很多初学者会问，到底学哪个？我的建议是：两套都要懂，但侧重点不同。

RDD 的集合操作（union、intersection、subtract、cartesian、join、cogroup）是一等公民，API 直接、语义清晰、使用零门槛。DataFrame 的集合操作（union、intersect、except、join）在语义上对齐 SQL，底层经过 Catalyst 优化器，很多情况下会比 RDD 直接跑更聪明——比如自动选择 Broadcast Join、自动做谓词下推、自动裁剪列。

两套 API 的集合操作在行为上有微妙的差异，比如 RDD 的union不去重，DataFrame 的union也不去重，但 DataFrame 的intersect默认去重而 RDD 的intersection也去重，这里面的语义差别后面会细说。一句话总结：RDD 的集合操作更偏底层、行为可预测；DataFrame 的集合操作更偏高层、有优化空间。生产环境里我绝大多数时候用 DataFrame，但调试问题、做快速原型时，RDD 反而更能看清执行细节。

2. 交并集与差集：语义、去重与性能

2.1 union 家族：union 与 unionByName

union是横向拼接，把两个数据集的行首尾相连，不涉及 key 匹配，所以它是最简单的集合操作。但简单不代表没坑。

先看 RDD 版本：

val rdd1 = sc.parallelize(Seq(1, 2, 3, 3)) val rdd2 = sc.parallelize(Seq(3, 4, 5)) rdd1.union(rdd2).collect() // 结果: Array(1, 2, 3, 3, 3, 4, 5)

注意两个细节：第一，union不去重，重复元素会原样保留；第二，union不需要 shuffle，它是纯 partition 级别的拼接——把两个 RDD 的 partition 简单合到一起，所以性能极好，几乎是集合操作里唯一没有 shuffle 的。

DataFrame 版本就不一样了。DataFrame 的union要求两个表的 schema 完全一致，包括列名、列顺序、列类型。这是个非常容易踩的坑：

val df1 = Seq(("a", 1)).toDF("name", "age") val df2 = Seq(("b", 2)).toDF("age", "name") // df1.union(df2) → 报错：字段名不匹配 // df1.unionByName(df2) → 正常，按列名匹配

unionByName是后来加的救火队员，它按列名而不是列位置来匹配，还支持allowMissingColumns参数，允许两个表列不一样时自动补 null。我见过太多线上作业因为上游表加了列、调整了顺序导致union炸掉，后来我定了一条规矩：只要涉及多表 union，一律用unionByName，并且显式指定列顺序。

还有一个性能细节：union在 DataFrame 里同样不需要 shuffle，但它有个隐藏行为——结果的行数等于两个表行数之和。如果你要的是集合意义上的"并集"（去重后的并集），得自己加distinct()。很多业务方说的"并集"其实是要去重的，这里一定要确认清楚。

2.2 intersection：求交集时的 shuffle 代价

intersection求两个数据集的共同元素，在集合论里是很自然的操作，但在分布式环境里它是个"重量级"选手。

val rdd1 = sc.parallelize(Seq(1, 2, 3, 3, 4)) val rdd2 = sc.parallelize(Seq(3, 3, 4, 5)) rdd1.intersection(rdd2).collect() // 结果: Array(3, 4) —— 注意会去重

为什么intersection要 shuffle？因为两个 RDD 中的元素分布在不同的 partition 上，要找到"共同元素"，本质上是一个全局的 group-by 操作：需要把相同元素拉到同一个节点上进行比较。Spark 的做法是先对两边的元素做哈希分区，再在每个分区内求交集，最后合并结果。这个过程天然包含 shuffle，而且两边的数据量越大，shuffle 的量就越大。

DataFrame 的intersect语义和 RDD 的intersection一致，都是"集合交集"——结果会去重。如果你想去重之前的版本，Spark 3.x 提供了intersectAll，它保留重复元素的交集次数。举个例子：

这个区别在业务上很关键。比如你算两个时间窗口内的重复下单用户，如果同一个用户在一个窗口内下了三次单，另一个窗口内下了两次单，你要的是"哪些用户在两段内都出现过"，用intersect；你要的是"两个窗口内共同产生的订单行为"，用intersectAll。语义差之毫厘，结果谬以千里。

性能上intersection没有捷径。如果一边的数据很小，可以考虑把小的那侧收集到 Driver 端，用广播变量结合filter实现，避免全量 shuffle。这在"小表过滤大表"的场景里非常实用。

2.3 subtract 和 except：差集的正确打开方式

差集在 RDD 里叫subtract，在 DataFrame 里叫except/exceptAll。语义都很好理解：A 减 B，返回在 A 中但不在 B 中的元素。

val rdd1 = sc.parallelize(Seq(1, 2, 3, 3, 4)) val rdd2 = sc.parallelize(Seq(3, 4, 5)) rdd1.subtract(rdd2).collect() // 结果: Array(1, 2) —— 注意去重后的差集 rdd1.map(x => (x, x)).subtractByKey(rdd2.map(x => (x, x))).collect() // 结果: Array((1,1), (2,2), (3,3)) —— subtractByKey 不去重，保留所有 A 中不匹配的记录

注意subtract默认去重，所以 [3,3] 里的两个 3 被当成一个处理了。要去重版差集用subtract即可；要保留重复的差集，在 RDD 里需要绕一下，或者用 DataFrame 的exceptAll。

这里有个非常容易误用的场景：做数据质量核对。比如你有一个基准表 A（预期数据）和一个实际产出表 B（实际数据），要找出 A 中有但 B 中没有的"缺失记录"，很多人直接写A.except(B)。但except是去重语义，如果 B 中有重复记录，结果可能会让你误判；反过来，exceptAll保留重复，又可能把重复导致的差异也报出来。实际做核对时，我最常用的做法是给数据加一列唯一 ID，用 anti join 来精确找缺失，这个后面再说。

再说说subtractByKey和subtract的差别。subtract作用于整个元素，subtractByKey只按 key 过滤，并且不去重（至少不主动去重）。比如你有一份订单表按用户 ID 成对，想找"有订单但没注册的用户"，用subtractByKey会比先转回普通元素再subtract更高效，因为它省掉了一次全元素级别的比较。

3. 笛卡尔积：最强大也最危险的操作

3.1 笛卡尔积的代价与适用场景

笛卡尔积的意思是把两个数据集的所有行两两配对。A 有 m 行，B 有 n 行，结果就是 m × n 行。看起来很普通的一个数学操作，在分布式里它是灾难级别的：结果是平方级膨胀，而且需要极大规模的 shuffle。

val rdd1 = sc.parallelize(Seq(1, 2, 3)) val rdd2 = sc.parallelize(Seq("a", "b")) rdd1.cartesian(rdd2).collect() // 结果: (1,a),(1,b),(2,a),(2,b),(3,a),(3,b)

如果你有一个 10 亿行的表和另一个 10 亿行的表做笛卡尔积，结果是 10 的 18 次方行，世界上任何一个集群都扛不住。所以你在生产环境里几乎不会看到cartesian的直接调用——它是 Spark 里最危险的操作没有之一。

但不代表它没用。笛卡尔积最经典的场景是不太大的数据集之间的全量配对，比如：

• 商品相似度计算：几千个商品互相算相似度，结果量级是几百万，可以接受。
• 地理位置距离矩阵：几千个点两两算距离。
• 个性化推荐里的 item-item 协同过滤：候选集有限时，先算出所有 item pair 再过滤。
• 参数网格搜索：每组超参数和数据的排列组合。

如果用 DataFrame，cross join也是同样的效果，而且 Spark 3.x 之后默认禁止cross join，必须显式加上spark.sql.crossJoin.enabled=true或者使用df1.crossJoin(df2)这种显式写法才允许。这个限制是 Spark 官方做的一个保护性设计，防止开发者在 SQL 里写个不带条件的 join，莫名其妙变成笛卡尔积。我建议你保留这个保护，不要为了图省事把它关掉，因为一旦关掉，任何忘记写 join 条件的查询都会变成笛卡尔积，线上事故就这么来的。

3.2 把笛卡尔积变成可控操作的三个技巧

笛卡尔积本身不可怕，可怕的是无脑用、用错规模。我在实际项目里总结了几种把笛卡尔积"降级"成可控操作的方式。

技巧一：先用条件缩小维度再 cross

不要对全量数据做笛卡尔积，先做过滤、聚合、采样，把两边的数据量压到万级以下，再做cartesian。比如商品相似度，先按品类分组，只在同一个品类内做两两组合，而不是全量商品做组合。这一步能把数据量缩小几个数量级，算子本身不变，但结果规模完全不同。

技巧二：用加盐 join 代替纯笛卡尔积

如果你需要的"两两配对"里其实有一个 key 可以关联（比如同一个用户 ID），那就应该用 join，而不是cartesian。很多人写cartesian是因为不知道或者没想清楚还有 join 这条路。判断标准很简单：最终结果是否需要"相同 key 才能配对"？如果是，就一定能用 join 表达。只有完全没有任何关联条件、纯粹的全排列，才需要笛卡尔积。

技巧三：计算上避免一次性物化全部结果

笛卡尔积的结果如果继续做聚合，尽量让 Spark 在 map 阶段做部分聚合，而不是把全部配对结果落盘再算。比如你只需要每个配对的相似度大于某个阈值才保留，可以在flatMap里先过滤再输出，减少后续 shuffle 的数据量。这一步能救回很多作业的命。

实践中最安全的使用姿势是：先小数据集 sample 验证逻辑，再逐步放大，任何笛卡尔积作业都要有结果行数上限的估算。m 和 n 的数量级心里没数，就别写这个算子。

4. Join：从原理到实践的最全拆解

4.1 六种 Join 类型的语义与选择

Join 是集合操作里最常用、也最复杂的。很多人只会用 inner join，但实际业务里 left、right、full、semi、anti 各有各的用途，全掌握才能写出正确的数据加工逻辑。

val users = Seq( ("u1", "张三"), ("u2", "李四"), ("u4", "王五") ).toDF("uid", "name") val orders = Seq( ("u1", 100), ("u2", 200), ("u3", 300) ).toDF("uid", "amount")

这里要特别强调left_semi和left_anti的价值。它们做的是"过滤"而不是"拼接"，所以结果只保留左表的列，不会把右表的列带进来。很多人用left_outer加上where right.uid is null来模拟 anti join，这种方式也能跑，但会额外携带右表的列跟着 shuffle。数据量大时，left_anti比left_outer + is null更高效、也更语义清晰。我在做数据对账、找出"该有却没有"的记录时，几乎只用left_anti，配合唯一键，比except精确得多。

再补充一个join的条件写法。上面例子用的是字符串条件"uid"，Spark 会自动把同名列作为等值 join 条件。更常见的是写表达式：

users.join(orders, users("uid") === orders("uid"), "inner") users.join(orders, Seq("uid"), "left_outer") // 指定 key 列表，不会出现重复列

一个很坑的地方：如果 join 的两个 DataFrame 有重名列，结果 DataFrame 里列名会变得很难看。用字符串 key 方式 join 时，Spark 会只保留一个 key 列；用表达式方式 join 时，两个 uid 列都会出现，你得手动取列或重命名。我做项目时的经验是：能用字符串 key 列表就用字符串 key 列表，表达式方式只在需要非等值 join 时才用。

4.2 Broadcast Join 与 Shuffle Join：选型决定生死

同样是 join，底层实现方式决定了作业跑得动跑不动。Spark 有两类 join 策略：

Shuffle Join（标准 Join）：两边的数据都按 join key 做哈希分区，相同 key 的数据送到同一个 Executor 上再做匹配。这是通用方案，但代价高——两边全量数据都要 shuffle。如果一张表有 1 亿行，另一张有 5000 万行，shuffle 的数据量就是 1.5 亿行，集群的磁盘 IO、网络 IO、CPU 全部被拉满。

Broadcast Join（Map Join）：如果小表足够小（默认阈值 10MB，配置项是spark.sql.autoBroadcastJoinThreshold），Spark 会直接把小表完整复制到每个 Executor 的内存里，join 变成纯本地操作，大表只需要流式读取，完全不 shuffle。性能差距通常在一个数量级以上。

什么时候会触发 Broadcast Join？Catalyst 优化器会在物理计划阶段判断，如果小表尺寸低于阈值就自动选择。但自动判断有几个问题：

• 如果你用 RDD 手动 join，没有这个优化。
• 如果小表尺寸估算不准（比如实际数据膨胀），会选错。
• 如果你明确知道一张表就是很小，可以用注释强制走 broadcast：

users.join(broadcast(orders), "uid", "inner") // 或者 Hint 方式 users.join(orders.hint("broadcast"), "uid", "inner")

强制 broadcast 的风险在于：如果"小表"实际很大，每个 Executor 复制一份，内存会被撑爆。所以要先确认数据尺寸再决定是否强制。我见过线上作业本来 10 分钟跑完，因为加了 broadcast hint 而 OOM 挂掉，最后查出来小表实际有 200MB。自动阈值比较保守，手动 hint 一定要心里有数。

另一个和 join 相关的性能点：join 之前的预处理。两个表在 join 前如果能先过滤、先聚合，让参与 join 的数据量变小，shuffle 的代价也会大幅下降。这是整个 Spark 性能优化里性价比最高的一件事，比调 Executor 内存、调并行度都有效。

4.3 数据倾斜：join 跑不动的元凶

如果说有一个问题最常导致 Spark 作业挂掉，排在第一名的一定是数据倾斜。具体到 join 场景，就是某个 key 的数据量特别大，导致某个 Executor 上的数据量远超其他 Executor，整个 job 卡在那里。

症状很好识别：日志里看到某个 stage 大部分 task 几秒就结束，但一两个 task 跑了几十分钟甚至 OOM。这就是典型的倾斜。

倾斜的场景通常是这样：用户表里有一个"默认用户"或者"测试用户"，比如 ID 为 0 或者空字符串，外键关联时大量孤儿数据全部指向这一个 key。两个表做 join 时，这个 key 对应的数据全部集中到一个分区，一个 Executor 就要处理几百万条配对。

常用的解法有这么几种，从简单到复杂：

方案一：过滤掉倾斜 key（最粗暴但最有效）

如果你的业务逻辑允许，先把倾斜 key 的行过滤掉，单独跑一份，最后再 union 回来。比如空 ID 的订单本来就没有用户信息，直接filter(uid is not null and uid != "")之后再做 join，能去掉大半倾斜。

方案二：加盐打散（Salting）

给倾斜的 key 加一个随机前缀，把一个 key 拆成 N 个 key。对小的那侧，也做同样的操作，把数据也展开成 N 份。这样原来压在一个 task 上的负载会分散到 N 个 task 上。代价是数据量膨胀 N 倍，shuffle 量变大，但至少作业能跑完。

// 假设只有少量 key 倾斜 val saltedOrders = orders.withColumn("salt", when(col("uid").isin(skewKeys: _*), (rand() * 10).cast("int")) .otherwise(0) ).withColumn("joinKey", concat(col("uid"), lit("_"), col("salt")))

方案三：广播那个小表

如果倾斜不在小表身上，而小表本身又不大，直接 broadcast join，把两边数据都在本地处理，彻底避开 shuffle 倾斜问题。这是最简单的方案，但前提是表确实小。

实际排查倾斜时，我的做法是：先看task的输入数据量统计，找出 top 几个超大 task；再用 groupBy 统计 key 的分布；确定倾斜 key 后按上面的方案逐个尝试。倾斜永远是 join 场景里的头号问题，不清楚 key 分布就开始调参，等于盲人摸象。

5. 实操中常见的坑与排查实录

5.1 常见问题速查表

把我在项目里反复遇到的坑整理成一张表，遇到问题直接对照：

5.2 多个实操心得的复盘

第一，写集合操作前先确认去重语义。这是最容易出"结果看起来对"但实际错的地方。union不去重、intersect去重、except去重、exceptAll不去重，四兄弟语义各不相同。每次写完都要在心里过一遍：这个业务场景要的是集合还是多重集合？错了就是静默的数据错误，审计都很难发现。

第二，join 之前先看数据分布。任何 join 作业上线前，我都会对 join key 做一次 groupBy count，按数量倒序看一下 top 10。如果 top key 的占比超过百分之十，就要警惕倾斜。这一步十几秒就能跑完，但能避免作业跑到一半 OOM 的惨剧。

第三，能广播就广播，不能广播就预聚合。所有 join 场景里，我先问一句：小表能不能广播？如果能，就强制 broadcast。如果两边都大，就得看能不能在 join 前对某一边做聚合、压缩行数。实在不行再考虑换分布式方案。没有优化的 join 是最典型的"看起来写完就完了，一跑就挂"的写法。

第四，自己写的 join 条件，永远先跑一个小验证。用 limit 或者过滤一段小时间窗口，先看 join 结果字段对不对、行数符不符合直觉，再放全量跑。这不是小白的习惯，是生产环境老手的保命技。全量跑挂了重跑的成本，比先验证多花的十分钟贵得多。

结尾补一个建议

最后说个小技巧：排查集合操作问题时，我习惯把 DataFrame 的执行计划打出来看一眼——df.explain(true)，看它实际选择了 BroadcastJoin 还是 SortMergeJoin，shuffle 发生在哪一步。很多时候你以为自己在用 Broadcast join，实际因为某个配置没生效，默默走成了 SortMergeJoin，作业慢得莫名其妙。这个习惯帮我避过好几次坑，希望也能帮到你。这些集合操作的知识不是背出来的，是踩坑踩出来的。你能在一个具体的业务问题上动手试一遍，比读十篇博客都管用。