从 Pyspark 中的 RDD 中提取字典

Question

这是一道作业题：

我有一个RDD，它是一个 os 元组的集合。我也有从每个输入元组返回字典的函数。不知何故，与 reduce 函数相反。

使用 map，我可以轻松地从 RDD 的元组转到 RDD 的字典。但是，由于字典是 (key, value) 对的集合，我想将字典的 RDD 转换为每个字典内容的 (key, value) 元组的 RDD。

这样，如果我的RDD 包含 10 个元组，那么我会得到一个 RDD，其中包含 10 个具有 5 个元素的字典（例如），最后我会得到一个包含 50 个元组的 RDD。

我认为这必须是可能的，但是，如何？ （可能问题是我不知道这个操作英文怎么叫）

Answer 1

我的 2 美分：

有一个名为“collectAsMap”的 PairRDD 函数，它从 RDD 返回一个字典。

让我给你看一个例子：

sample = someRDD.sample(0, 0.0001, 0)
sample_dict = sample.collectAsMap()
print sample.collect()
print sample_dict

[('hi', 4123.0)]
{'hi': 4123.0}

文档here

希望对您有所帮助！ 问候！

Answer 2

我猜你想要的只是一个flatMap：

dicts = sc.parallelize([{"foo": 1, "bar": 2}, {"foo": 3, "baz": -1, "bar": 5}])
dicts.flatMap(lambda x: x.items())

flatMap 从 RDD 的元素中获取一个函数到可迭代的，然后连接结果。 Spark 上下文之外的同类型操作的另一个名称是mapcat：

>>> from toolz.curried import map, mapcat, concat, pipe
>>> from itertools import repeat
>>> pipe(range(4), mapcat(lambda i: repeat(i, i + 1)), list)
[0, 1, 1, 2, 2, 2, 3, 3, 3, 3]

或者一步一步来：

>>> pipe(range(4), map(lambda i: repeat(i, i + 1)), concat, list)
[0, 1, 1, 2, 2, 2, 3, 3, 3, 3]

同样的事情使用itertools.chain

>>> from itertools import chain
>>> pipe((repeat(i, i + 1) for i in  range(4)), chain.from_iterable, list)
>>> [0, 1, 1, 2, 2, 2, 3, 3, 3, 3]