pyspark 加入多个条件答案

【问题标题】：pyspark join multiple conditionspyspark 加入多个条件
【发布时间】：2016-03-06 15:12:09
【问题描述】：

如何指定很多条件 pyspark 当我使用.join()

示例：带蜂巢：

query= "select a.NUMCNT,b.NUMCNT as RNUMCNT ,a.POLE,b.POLE as RPOLE,a.ACTIVITE,b.ACTIVITE as RACTIVITE FROM rapexp201412 b \
    join rapexp201412 a where (a.NUMCNT=b.NUMCNT and a.ACTIVITE = b.ACTIVITE and a.POLE =b.POLE  )\

但在 PySpark 中我不知道如何制作它，因为以下原因：

df_rapexp201412.join(df_aeveh,df_rapexp2014.ACTIVITE==df_rapexp2014.ACTIVITE and df_rapexp2014.POLE==df_aeveh.POLE,'inner')

没用！！

【问题讨论】：

能否请您粘贴 DataFrame.join 的错误消息？或者尝试在RDD中使用keyBy/join，它非常支持equi-join条件。
在两者之间使用 & 并确保将每个条件放在 () 中
How to join on multiple columns in Pyspark?的可能重复

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

【解决方案1】：

引自 spark 文档：

(https://spark.apache.org/docs/1.5.2/api/python/pyspark.sql.html?highlight=dataframe%20join#pyspark.sql.DataFrame.join)

join(other, on=None, how=None) 与另一个 DataFrame 连接，使用给定连接表达式。

下面执行 df1 和 df2 之间的完全外连接。

参数：other – 右边的join on – 一个用于join的字符串列名、列名列表、连接表达式 (Column) 或列列表。如果 on 是一个字符串或一个字符串列表，表示连接列的名称，列必须存在于两边，这执行内部等值连接。如何 - str，默认为“内部”。一内、外、左外、右外、半连接。

>>> df.join(df2, df.name == df2.name, 'outer').select(df.name, df2.height).collect()
 [Row(name=None, height=80), Row(name=u'Alice', height=None), Row(name=u'Bob', height=85)]


>>> cond = [df.name == df3.name, df.age == df3.age]
>>> df.join(df3, cond, 'outer').select(df.name, df3.age).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]

所以你需要像上一个例子一样使用“条件作为列表”选项。

【讨论】：

this cond = [df.name == df3.name, df.age == df3.age] 表示“和”还是“或”？如果我想要 df.name == df1.name 或 df.age == df1.age 怎么办？这将如何运作？
改用cond = [(df.name == df3.name) | (df.age == df3.age)]，其中|代表“或”条件。
也可以使用使用字符串而不是硬编码列名的连接表达式列表，例如stackoverflow.com/a/45813075/38368

【解决方案2】：

>>> cond = [df.name == df3.name, df.age == df3.age]
>>> df.join(df3, cond, 'outer').select(df.name, df3.age).collect()
[Row(name=u'Bob', age=5), Row(name=u'Alice', age=2)]

这不适用于 pyspark 1.3.1。我收到“AssertionError：joinExprs 应该是 Column”

相反，我使用原始 sql 加入数据框，如下所示

df.registerTempTable("df")
df3.registerTempTable("df3")

sqlContext.sql("Select df.name,df3.age from df outer join df3 on df.name = df3.name and df.age =df3.age").collect()

【讨论】：

【解决方案3】：

报告@S V Praveen 回复，因为我在连接条件下无法表达OR：

cond = [df.surname == df3.surname, 
        (df.name == df3.name) | (df.age == df3.age),
        df.orders >= df3.orders ]

df.join(df3, cond, 'outer')

其中| 表示“或”条件

【讨论】：

【解决方案4】：

您正在寻找的是以下内容

cond = [((df1.col1 == df2.col2) &\
         (df1.col3 == df2.col4))]

joined_df = df1.join(df2, on = cond, how = "inner")

【讨论】：