如何在PySpark RDD中从单元素集合逐步生成多元素分组？

嗨，我来帮你搞定这个Spark RDD组合的问题～

首先得说，你之前用rdd.reduce(lambda x,y: (x+ " " + y))的思路其实走偏了——reduce是用来把整个RDD聚合成一个单一值的，所以你得到的是把所有元素拼起来的字符串，这可不是生成组合的正确姿势哦。而且foreach函数是用来给每个元素执行副作用操作（比如打印、写文件）的，不是用来生成新RDD的，生成组合得用转换操作来搞。

下面分两种场景给你解决方案：

场景1：数据量不大（可以安全拉到本地处理） #

这种情况用Python的itertools.combinations最省心，它天生就是用来生成无序、不重复元素的组合的，完美符合你的需求——自动排除同元素组合（比如'1 1'）和顺序不同的重复组合（比如'1 5'和'5 1'）。

代码示例：

from itertools import combinations
from pyspark import SparkContext

# 初始化SparkContext
sc = SparkContext("local", "CombinationDemo")
# 你的原始RDD
original_rdd = sc.parallelize(['1','5','7','8'])

# 生成两两组合的RDD
local_list = original_rdd.collect()
pair_rdd = sc.parallelize([' '.join(pair) for pair in combinations(local_list, 2)])
print(pair_rdd.collect())
# 输出：['1 5', '1 7', '1 8', '5 7', '5 8', '7 8']

# 生成三三组合的RDD
triple_rdd = sc.parallelize([' '.join(triple) for triple in combinations(local_list, 3)])
print(triple_rdd.collect())
# 输出：['1 5 7', '1 5 8', '1 7 8', '5 7 8']

场景2：数据量很大（不能拉到本地，要分布式处理） #

这种情况就用Spark的cartesian笛卡尔积操作，再通过过滤条件来筛选符合要求的组合：

from pyspark import SparkContext

sc = SparkContext("local", "DistributedCombinationDemo")
original_rdd = sc.parallelize(['1','5','7','8'])

# 分布式生成两两组合
pair_rdd_distributed = original_rdd.cartesian(original_rdd) \
    # 过滤掉同元素组合，以及顺序相反的重复组合（这里假设元素是可比较的字符串，要是是数字可以转成int再比较）
    .filter(lambda x: x[0] < x[1]) \
    .map(lambda x: ' '.join(x))

print(pair_rdd_distributed.collect())
# 输出和之前一样：['1 5', '1 7', '1 8', '5 7', '5 8', '7 8']

# 分布式生成三三组合的话，思路类似但会复杂点——可以先做两次cartesian，再过滤重复元素和顺序问题，不过更推荐用上面的combinations方法如果数据量允许；如果必须分布式，也可以用flatMap结合窗口或者其他方式，但一般小数据量用combinations足够了

补充说明 #

• combinations(iterable, k)：生成的是长度为k的无序组合，元素顺序和原迭代器一致，不会生成重复的顺序组合，也不会包含相同元素的组合，完全匹配你的需求。
• 为什么不用foreach？foreach是行动操作，用来执行副作用，比如打印每个元素、写入数据库，但它不会返回新的RDD，所以不适合用来生成组合这种需要转换数据的场景。

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