如何加入 Pyspark 中的多个列？答案

【问题标题】：How to join on multiple columns in Pyspark?如何加入 Pyspark 中的多个列？
【发布时间】：2016-02-18 04:13:44
【问题描述】：

我正在使用 Spark 1.3，并希望使用 python 接口 (SparkSQL) 加入多个列

以下作品：

我首先将它们注册为临时表。

numeric.registerTempTable("numeric")
Ref.registerTempTable("Ref")

test  = numeric.join(Ref, numeric.ID == Ref.ID, joinType='inner')

我现在想根据多个列加入它们。

我收到 SyntaxError: invalid syntax with this:

test  = numeric.join(Ref,
   numeric.ID == Ref.ID AND numeric.TYPE == Ref.TYPE AND
   numeric.STATUS == Ref.STATUS ,  joinType='inner')

【问题讨论】：

标签： python apache-spark join pyspark apache-spark-sql

【解决方案1】：

您应该使用& / | 运算符并注意operator precedence（== 的优先级低于按位AND 和OR）：

df1 = sqlContext.createDataFrame(
    [(1, "a", 2.0), (2, "b", 3.0), (3, "c", 3.0)],
    ("x1", "x2", "x3"))

df2 = sqlContext.createDataFrame(
    [(1, "f", -1.0), (2, "b", 0.0)], ("x1", "x2", "x3"))

df = df1.join(df2, (df1.x1 == df2.x1) & (df1.x2 == df2.x2))
df.show()

## +---+---+---+---+---+---+
## | x1| x2| x3| x1| x2| x3|
## +---+---+---+---+---+---+
## |  2|  b|3.0|  2|  b|0.0|
## +---+---+---+---+---+---+

【讨论】：

当你说“注意运算符优先级”时，你是什么意思？你的意思是我应该把括号放在正确的地方把正确的表格放在一起吗？
@Chogg，他的意思是，如果你不小心使用括号，那么短语 df1.x1 == df2.x1 & df1.x2 == df2.x2，（括号已删除）将被 Python 解释器评估为 df1.x1 == (df2.x1 & df1.x2) == df2.x2，这可能抛出一个令人困惑的非描述性错误。

【解决方案2】：

另一种方法是：

df1 = sqlContext.createDataFrame(
    [(1, "a", 2.0), (2, "b", 3.0), (3, "c", 3.0)],
    ("x1", "x2", "x3"))

df2 = sqlContext.createDataFrame(
    [(1, "f", -1.0), (2, "b", 0.0)], ("x1", "x2", "x4"))

df = df1.join(df2, ['x1','x2'])
df.show()

哪个输出：

+---+---+---+---+
| x1| x2| x3| x4|
+---+---+---+---+
|  2|  b|3.0|0.0|
+---+---+---+---+

主要优点是连接表的列不会在输出中重复，从而降低遇到org.apache.spark.sql.AnalysisException: Reference 'x1' is ambiguous, could be: x1#50L, x1#57L.等错误的风险

只要两个表中的列具有不同的名称，（假设在上面的示例中，df2 具有列 y1、y2 和 y4），您可以使用以下语法：

df = df1.join(df2.withColumnRenamed('y1','x1').withColumnRenamed('y2','x2'), ['x1','x2'])

【讨论】：

如果我进行外部连接并希望只保留一次出现的键怎么办
这可能是我最少最喜欢的 pyspark 错误：Reference 'x1' is ambiguous, could be: x1#50L, x1#57L. 我不明白为什么它会让你执行df = df1.join(df2, df1.x1 == df2.x1) 之类的操作，然后一尝试就会出错对生成的df 做几乎任何事情。这只是一个小小的咆哮，但是有什么理由让你想要得到的 df 有重复的名字？

【解决方案3】：

test = numeric.join(Ref, 
   on=[
     numeric.ID == Ref.ID, 
     numeric.TYPE == Ref.TYPE,
     numeric.STATUS == Ref.STATUS 
   ], how='inner')

【讨论】：

欢迎来到 StackOverflow。你能解释一下你的代码吗？为什么会有这样的结构？它是如何工作的？等
答案很棒。但对于最佳实践，请提供解释。您只发布代码会使 OP 和未来的 commers 复制并粘贴您的答案，而不了解答案背后的逻辑。请提供一些解释的答案。谢谢！