如何高效对170万行的Spark RDD执行条件自连接操作？

RDD的惰性求值特性仅代表所有转换操作不会在定义时立即执行，并不会改变算子本身的计算逻辑。笛卡尔积的底层计算逻辑就是生成两份RDD所有元素的两两组合，所有组合生成后才会进入后续的filter算子执行过滤，你看到的反复读取同一块RDD分区的日志，正是笛卡尔积的计算特性导致的：源RDD的每一个分区都需要和另一份源RDD的所有分区做配对计算，所以单个分区会被反复读取多次。
170万条整数的全量笛卡尔积会生成接近3万亿条中间数据，哪怕你缓存了源RDD，如此大的计算量也不可能在短时间内完成，后续的过滤逻辑根本没有机会执行。

可以根据你的过滤条件选择不同的优化方案，都能完全避开全量笛卡尔积的O(n²)开销：

• 如果过滤条件为等值匹配规则，直接使用Spark原生join算子：先将RDD转换为键值对格式，键为你用来做匹配判断的字段，再执行自连接，Spark会通过shuffle将相同键的元素分到同一个分区，仅对相同键的元素做配对，中间数据量直接降到和你预期的2400万条输出接近。示例代码逻辑：

// 示例规则：两个数的模10结果相等才保留
val keyedRdd = setA.map(x => (x % 10, x))
val result = keyedRdd.join(keyedRdd).map(_._2)

• 如果过滤条件为范围匹配规则（比如两个数的差值不超过100），使用分桶裁剪优化：先给所有整数按范围分桶，仅对相同桶或相邻桶的元素做配对，再应用你的自定义过滤规则，可直接砍掉90%以上的无效配对计算。
• 如果你的源数据量较小（170万条整数仅占几十MB内存），可直接使用广播变量优化：将整个源RDD数据广播到所有Executor节点，在每个Executor本地做遍历配对过滤，完全避免shuffle开销，性能会有数量级的提升。

内容的提问来源于stack exchange，提问作者weak_at_math