使用Pyspark进行多条件连接

Question

当我使用.join()时，如何在pyspark中指定多个条件？

例如：与hive一起使用：

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')

它不起作用！！

- malouke

---

请问您能否将DataFrame.join的错误信息粘贴出来？或者尝试在RDD中使用keyBy/join，它非常支持等值连接条件。 - Shawn Guo

使用 & 连接条件，并确保每个条件都在 () 中。 - Vivek

可能是如何在Pyspark中使用多个列进行连接？的重复问题。 - Florian

Answer 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 - 连接右侧的DataFrame，on - 用于连接的字符串或列名列表、连接表达式（Column）或列列表。如果on是指示连接列名称的字符串或字符串列表，则连接列必须存在于两个DataFrame中，并且这将执行内部等值连接。how - str，默认为'inner'。其中之一：inner、outer、left_outer、right_outer、semijoin。

>>> 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)]

所以你需要使用“条件列表”选项，就像最后一个例子中一样。

- user3689574

---

这个条件语句 [df.name == df3.name, df.age == df3.age] 表示的是“与（and）”还是“或（or）”？如果我想要的是 df.name == df1.name 或者 df.age == df1.age，应该怎么写呢？ - AJR

请使用 cond = [(df.name == df3.name) | (df.age == df3.age)] 替换原来的代码，并注意 | 表示“或”条件。 - S V Praveen

使用字符串而不是硬编码的列名来使用连接表达式列表也是可能的，例如 https://stackoverflow.com/a/45813075/38368。 - Danny Varod

如果“on”是列表，则通过布尔“And”进行减少，如果文档提到这一点将会很有帮助。 - Wassadamo

Answer 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 should be 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()

- Mohan

Answer 3

因为我无法表达连接条件中的OR，所以报告@S V Praveen的回复：

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

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

| 表示"OR"条件

- Vzzarr

Answer 4

你要找的是以下内容。

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

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

- Devam