我有一个针对一个键的多个值（列表）的 rdd，我想从键中的每个值中过滤掉垃圾。
rdd 有这个数据
((key1, [('',val1),('', val2),..]),(key2,[...)
我想把它映射成这样的东西
((key1,[val1, val2,...]), key2[...)
我知道这里需要一个 map 函数，但我没有针对一个键使用 map 来处理多个值。
这是我的努力。
def mapper(x):
values = []
for a in x[1]:
values.append(a[1])
return(x[0], ap)
listRdd.map( mapper).collect()
但我遇到了一些错误
【问题讨论】：
标签： python apache-spark pyspark
主要思想是将RDD 的每个条目视为一个集合，并将其视为一个进程。意思是，如果我们考虑以下条目
entry = ("key1", [('',"val1"),('',"val2")])
要将这个集合处理成预期的输出，我们需要了解集合的结构
entry[0]
# 'key1'
entry[1]
# [('', 'val1'), ('', 'val2')]
现在让我们开始第二部分：
map(lambda x : x[1],entry[1])
# ['val1', 'val2']
我们现在可以定义一个函数，它将一个条目作为输入，结果输出将是一个 (key,[values...]) 元组。我们称之为mapper。我们可以将映射器应用于 rdd 中的每个条目。
将代码放在一起：
def mapper(entry):
return (entry[0],map(lambda x : x[1],entry[1]))
data = [("key1", [('',"val1"),('',"val2")]),("key2",[('',"val3"),('',"val2"),('',"val4")])]
rdd = sc.parallelize(data)
rdd2 = rdd.map(lambda x : mapper(x))
rdd2.collect()
# [('key1', ['val1', 'val2']), ('key2', ['val3', 'val2', 'val4'])]
【讨论】：