如何使用Python在Spark中执行两个RDD表的基本联接？

Question

您将如何使用python在Spark中执行基本联接？在R中，您可以使用merg（）进行此操作。在spark上使用python的语法是什么：

1. 内部联接
2. 左外连接
3. 交叉连接

具有两个表（RDD），每个表中的单个列具有一个公共键。

RDD(1):(key,U)
RDD(2):(key,V)

我认为内部联接是这样的：

rdd1.join(rdd2).map(case (key, u, v) => (key, ls ++ rs));

是对的吗？我已经在互联网上搜索了，找不到很好的加入示例。提前致谢。

Answer 1

可以使用PairRDDFunctions或Spark数据帧来完成。由于数据帧操作受益于Catalyst
Optimizer，
因此第二个选项值得考虑。

假设您的数据如下所示：

rdd1 =  sc.parallelize([("foo", 1), ("bar", 2), ("baz", 3)])
rdd2 =  sc.parallelize([("foo", 4), ("bar", 5), ("bar", 6)])

使用PairRDD：

内部联接：

rdd1.join(rdd2)

左外连接：

rdd1.leftOuterJoin(rdd2)

笛卡尔积（不需要RDD[(T, U)]）：

rdd1.cartesian(rdd2)

广播加入（不需要RDD[(T, U)]）：

• 请参见Spark：将2元组键RDD与单键RDD结合在一起的最佳策略是什么？

最后，cogroup它没有直接的SQL等效项，但在某些情况下可能有用：

cogrouped = rdd1.cogroup(rdd2)

cogrouped.mapValues(lambda x: (list(x[0]), list(x[1]))).collect()
## [('foo', ([1], [4])), ('bar', ([2], [5, 6])), ('baz', ([3], []))]

使用Spark数据框

您可以使用SQL DSL或使用来执行原始SQL sqlContext.sql。

df1 = spark.createDataFrame(rdd1, ('k', 'v1'))
df2 = spark.createDataFrame(rdd2, ('k', 'v2'))

# Register temporary tables to be able to use `sparkSession.sql`
df1.createOrReplaceTempView('df1')
df2.createOrReplaceTempView('df2')

内部联接：

# inner is a default value so it could be omitted
df1.join(df2, df1.k == df2.k, how='inner') 
spark.sql('SELECT * FROM df1 JOIN df2 ON df1.k = df2.k')

左外连接：

df1.join(df2, df1.k == df2.k, how='left_outer')
spark.sql('SELECT * FROM df1 LEFT OUTER JOIN df2 ON df1.k = df2.k')

交叉连接（在Spark 2.0中需要显式交叉连接或配置更改-Spark
2.x的spark.sql.crossJoin.enabled
）：

df1.crossJoin(df2)
spark.sql('SELECT * FROM df1 CROSS JOIN df2')

~~~~

df1.join(df2)
sqlContext.sql('SELECT * FROM df JOIN df2')

从1.6（Scala中为1.5）开始，这些broadcast功能都可以与功能结合使用：

from pyspark.sql.functions import broadcast

df1.join(broadcast(df2), df1.k == df2.k)

执行广播加入。另请参阅为什么我的BroadcastHashJoin比Spark中的ShuffledHashJoin慢