代码之家 › 专栏 › 技术社区 › alborzdev

如何在PySpark中加入两个RDD?

rdd pyspark apache-spark python

alborzdev · 技术社区 · 3 年前

我很难找到正确的方式加入PySpark中的RDD,以达到预期的效果。

Here is the first RDD:
+------+---+
|    _1| _2|
+------+---+
|Python| 36|
|     C|  6|
|    C#|  8|
+------+---+

Here is the second RDD:
+------+---+
|    _1| _2|
+------+---+
|Python| 10|
|     C|  1|
|    C#|  1|
+------+---+

Here is the result I want:
+------+---+---+
|    _1| _2| _3|
+------+---+---+
|Python| 36| 10|
|     C|  6|  1|
|    C#|  8|  1|
+------+---+---+

我试过各种各样的方法 .join() 和 .union() 两个RDD之间存在差异,但无法正确使用,如有任何帮助,将不胜感激!!

1 回复 | 直到 3 年前

pltc 3 年前

用RDD

rdd1 = sc.parallelize([('python', 36), ('c', 6), ('c#', 8)])
rdd2 = sc.parallelize([('python', 10), ('c', 1), ('c#', 1)])
rdd1.join(rdd2).map(lambda x: (x[0], *x[1])).toDF().show()
+------+---+---+
|    _1| _2| _3|
+------+---+---+
|python| 36| 10|
|     c|  6|  1|
|    c#|  8|  1|
+------+---+---+

使用DF

df1 = rdd1.toDF(['c1', 'c2'])
df2 = rdd2.toDF(['c1', 'c3'])
rdd3 = df1.join(df2, on=['c1'], how='inner').rdd

推荐文章

user3579222 · 阅读以前的Spark API

4 月前

JFlo · 在PySpark笔记本中读取多个Parquet文件

5 月前

Matthew Thomas · partition覆盖动态和“逻辑”分区

9 月前

Jamal Khan · 如何在Apache Spark中读取500 GB的大文件CSV文件并对其执行聚合?

10 月前

Nakeuh · 从数组列中新建数据帧列

10 月前

maximodesousadias · 如何根据条件删除日期后的记录

10 月前

Ajay S Pal · 当调用函数时传递参数时,PySpark没有在函数内部创建Dataframe

1 年前

SUBHOJEET · 如何使用pyspark读取rds文件?

1 年前

Shankar Panda · 如何从org.apache.spark.sql获取密钥。在scala中键入列并将其放入列表变量中?

1 年前

Aaron Brazier · 连接2个pyspark数据帧并继续运行窗口sum和max

1 年前