如何将列表中的值添加到 RDD 的每个项目中？

Question

假设我有常规 python 列表 [1,2] 我有一个包含 2 个项目的 rdd，比如 [('hi', 'bye'), ('hi', 'bye')] 我希望每个项目都成为

('hi', 'bye', 1)
('hi', 'bye', 2)

本质上是将列表中的每个项目附加到 rdd 中的每个项目。我觉得这应该很简单，但我想不出其中的逻辑：/

Answer 1

尝试匹配模式

a = [1,2]
b = [('hi', 'bye'), ('hi', 'bye')]

print([(x, y, z) for ((x, y), z) in zip(b, a)])

输出：

[('hi', 'bye', 1), ('hi', 'bye', 2)]

Answer 2

我想，这应该可以解决您的问题。 试试这个代码。

your_list = [1,2]
your_rdd = [('hi', 'bye'), ('hi', 'bye')]
print([(x, y, z) for ((x, y), z) in zip(your_rdd, your_list)])

说明：

为简单起见，您可以使用 str 个元素。

假设，你有一个 RDD，每一项都是

的形式

(key, [ele1, ele2, ele3, ..., elen])

每一项都是一个键值对，值是一个元素列表。

并且您想解压缩列表，以便我可以创建一个新的 RDD，其中每个项目都包含一个元素，如下所示：

(key, ele1)
(key, ele2)
(key, ele3)
.
.
.
(key, ele4)

比如这个，为了简单起见，可以使用str元素。

>>> rdd = sc.parallelize([('key', ['ele1', 'ele2'])])
>>> rdd.flatMap(lambda data: [(data[0], x) for x in data[1]]).collect()
[('key', 'ele1'), ('key', 'ele2')]

Answer 3

可以使用RDD的zip方法：

rdd1 = sc.parallelize([('hi', 'bye'), ('hi', 'bye')])
rdd2 = sc.parallelize([1, 2])

rdd3 = rdd1.zip(rdd2).map(lambda x: (x[0][0], x[0][1], x[1]))

rdd3.collect()
# [('hi', 'bye', 1), ('hi', 'bye', 2)]

如何将列表中的值添加到 RDD 的每个项目中？

How do I add values from a list into each item of an RDD?

python

apache-spark

rdd

apache-spark-sql

pyspark