PySpark中如何基于公共列值关联RDD行并统计配对次数

我有如下文本文件数据集：

Benz,25
BMW,27
BMW,25
Land Rover,22
Audi,25
Benz,25

期望得到的结果是：

[((Benz,BMW),2),((Benz,Audi),1),((BMW,Audi),1)]

该结果会将具有公共值的汽车配对，并统计它们共同出现的次数。我目前的代码如下：

cars= sc.textFile('cars.txt')
carpair= cars.flatMap(lambda x: float(x.split(',')))
carpair.map(lambda x: (x[0], x[1])).groupByKey().collect()

作为初学者，我连值的映射都无法完成，统计次数是次要需求。

• flatMap(lambda x: float(x.split(',')))：这行逻辑完全错误，x.split(',')会把每行拆成['品牌', '数值']的字符串列表，直接转float会因字符串无法转换报错；而且flatMap会把列表的每个元素单独输出，导致后续无法通过x[0]、x[1]取元组元素——此时每个元素是单个字符串或数值，不是元组。
• 后续的map(lambda x: (x[0], x[1]))自然也会报错，因为输入不是可索引的元组结构。

1. 正确读取并转换数据 #

先把每行文本分割成（品牌，数值）的元组，将数值转为整数：

cars = sc.textFile('cars.txt')
# 分割每行，生成（品牌，数值）元组
car_pairs = cars.map(lambda line: (line.split(',')[0], int(line.split(',')[1])))

执行后，car_pairs的元素是类似('Benz', 25)、('BMW', 27)的结构。

2. 按数值分组，统计每个数值下的品牌出现次数 #

先统计每个数值下各品牌的出现次数，再按数值分组：

# 先转成((数值, 品牌), 1)的结构，再按键累加次数
brand_count_per_value = car_pairs.map(lambda x: ((x[1], x[0]), 1)).reduceByKey(lambda a, b: a + b)
# 再转换为（数值，(品牌, 次数)）的结构，按数值分组
grouped_brand_count = brand_count_per_value.map(lambda x: (x[0][0], (x[0][1], x[1]))).groupByKey()

执行后，数值25对应的分组结果是[('Benz', 2), ('BMW', 1), ('Audi', 1)]，其他数值因只有单个品牌，不会产生配对。

3. 生成品牌配对并计算总次数 #

利用itertools.combinations生成每个数值下的品牌两两组合，计算组合的总次数（两个品牌次数的乘积）：

from itertools import combinations

def generate_pairs(brand_counts):
    brand_list = list(brand_counts)
    # 生成所有不重复的两两品牌组合（自动按顺序排列，避免(BMW,Benz)这类重复配对）
    pairs = combinations(brand_list, 2)
    # 返回（(品牌1, 品牌2), 总次数）的结构
    return [((b1[0], b2[0]), b1[1] * b2[1]) for b1, b2 in pairs]

# 应用函数并展开所有配对，最终收集结果
result = grouped_brand_count.flatMap(lambda x: generate_pairs(x[1])).collect()

执行后，result就是你期望的[((Benz,BMW),2),((Benz,Audi),1),((BMW,Audi),1)]。

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