在 PySpark 配对的 RDD 中搜索值，以获取来自另一个 RDD 的键答案

【问题标题】：search values within a PySpark paired RDD , for keys coming from another RDD在 PySpark 配对的 RDD 中搜索值，以获取来自另一个 RDD 的键
【发布时间】：2019-05-17 14:56:59
【问题描述】：

我是 PySpark 的新手，我想做以下事情，

考虑以下代码，

import numpy as np
b =np.array([[1,2,100],[3,4,200],[5,6, 300],[7,8, 400]])
a = np.array([[1,2],[3,4],[11,6],[7,8], [1, 2], [7,8]])
RDDa = sc.parallelize(a)
RDDb = sc.parallelize(b)
dsmRDD = RDDb.map(lambda x: (list(x[:2]), x[2]))

我想获取与每个 RDDa 值关联的值作为 dsmRDD 的键，即

result = [100, 200, 0, 400, 100, 400]

非常感谢您。

【问题讨论】：

您的代码有语法错误。

标签： pyspark key-value

【解决方案1】：

如果您的数据不是太大，您可以像这样使用数据框：

dfa = spark.createDataFrame([[1,2],[3,4],[11,6],[7,8], [1, 2], [7,8]], ["c1", "c2"])
dfb = spark.createDataFrame([[1,2,100],[3,4,200],[5,6, 300],[7,8, 400]], ["c1", "c2", "value"])
>>> dfa.join(dfb, on=["c1","c2"], how="left").na.fill(0).show()
+---+---+-----+
| c1| c2|value|
+---+---+-----+
|  7|  8|  400|
|  7|  8|  400|
| 11|  6|    0|
|  3|  4|  200|
|  1|  2|  100|
|  1|  2|  100|
+---+---+-----+

temp = dfa.join(dfb, on=["c1","c2"], how="left").na.fill(0)

>>> [i.value for i in temp.select("value").collect()]
[400, 400, 0, 200, 100, 100]

【讨论】：

非常感谢您的回复！我现在就试试。更正一下，我的数据真的很大，只是一个小例子来说明问题！我正在使用 PySpark！

【解决方案2】：

正如另一个答案所暗示的，您可以转换为数据框和join。如果您只愿意继续使用rdd，您可以这样做，

import numpy as np

a = np.array([[1,2],[3,4],[11,6],[7,8], [1, 2], [7,8]])
b = np.array([[1,2,100],[3,4,200],[5,6, 300],[7,8, 400]])

RDDa = sc.parallelize(a)
RDDb = sc.parallelize(b)

dsmRDD = RDDa.zipWithIndex()\
         .map(lambda x: (tuple(x[0].tolist()),(0,x[1])))\
         .leftOuterJoin(RDDb.map(lambda x: (tuple(x[:2].tolist()), x[2])))\
         .map(lambda x: (x[1][0][1], x[1][1]) if x[1][1] is not None else (x[1][0][1],x[1][0][0]))

output = map(lambda x:x[1], sorted(dsmRDD.collect()))
print output

给你输出，

[100, 200, 0, 400, 100, 400]

【讨论】：

非常感谢！但是，它没有按照顺序！顺序对我的工作很重要。正确的顺序应该是 100, 200, 0, 400, 100, 400]。
Spark 数据结构不保留行的顺序。我们可以做的是添加一个额外的索引列，然后对行进行排序。检查编辑。
如果您觉得答案有帮助，请点赞并接受，以便其他用户也能轻松找到它！