我有一个针对一个键的多个值(列表)的 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()
但是我得到了一些错误
请您参考如下方法:
主要思想是将 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'])]
