如何在Spark RDD中对列表去重并生成带计数的键值对？

需求回顾 #

原始RDD结构：

[('key1', ['word1', 'word1', 'word2', ...]),
 ('key2', ['word1', 'word2', 'word2', ...]),
 ...
]

需要完成两个操作：

1. 对每个key对应的列表去重，得到：

[('key1', ['word1', 'word2', ...]),
 ('key2', ['word1', 'word2', ...]),
 ...
]

2. 将去重后的每个元素与对应key配对，生成带计数1的键值对，最终输出：

[('key1', ('word1', '1')),
 ('key1', ('word2', '1')),
 ...
 ('key2', ('word1', '1')),
 ('key2', ('word2', '1')),
 ...
]

错误代码分析 #

你之前的代码：

rdd.map(lambda x: (x[0], (x[1], 1))).collect()

问题在于：

• 没有对列表进行去重操作，直接保留了原始重复元素
• 使用map时直接将整个列表和1绑定，没有遍历列表中的单个元素，导致输出是整个列表和计数的配对，而非单个元素

正确实现方法 #

方法一：分步实现 #

1. 对每个key的列表去重
   如果不要求保留元素原始顺序，用集合去重最简便：

   # 第一步：去重每个key对应的列表
   dedup_rdd = rdd.map(lambda x: (x[0], list(set(x[1]))))
   

   如果需要保持元素在原列表中的出现顺序，可自定义去重函数：

   def dedup_list(lst):
       seen = set()
       result = []
       for item in lst:
           if item not in seen:
               seen.add(item)
               result.append(item)
       return result
   
   dedup_rdd = rdd.map(lambda x: (x[0], dedup_list(x[1])))
   

2. 展开列表生成目标键值对
   这里需要用flatMap而非map——flatMap会将每个输入元素转换为多个输出元素并展开，正好满足把列表中每个元素单独和key配对的需求：

   # 第二步：生成带计数1的键值对
   final_rdd = dedup_rdd.flatMap(lambda x: [(x[0], (word, '1')) for word in x[1]])
   

方法二：合并为一行代码 #

如果不需要保留中间的去重RDD，可以把两步合并：

# 不保留顺序的版本
final_rdd = rdd.flatMap(lambda x: [(x[0], (word, '1')) for word in list(set(x[1]))])

# 保留顺序的版本
final_rdd = rdd.flatMap(lambda x: [(x[0], (word, '1')) for word in dedup_list(x[1])])

验证输出 #

执行final_rdd.collect()即可得到你预期的结果。

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