pyspark - 这两种全外连接有什么区别？

Question

完整示例可以在这里找到。

我在使用pyspark对两个数据框进行全外连接时，发现了两种不同的输出结果：

users1_df. \
    join(users2_df, users1_df.email == users2_df.email, 'full_outer'). \
    show()

这个代码的结果是：

+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+
|               email|first_name|last_name|gender|   ip_address|               email|first_name| last_name|gender|     ip_address|
+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+
| alovett0@nsw.gov.au|   Aundrea|   Lovett|  Male|  62.72.1.143|                null|      null|      null|  null|           null|
|bjowling1@spiegel.de|   Bettine|  Jowling|Female|26.250.197.47|bjowling1@spiegel.de|    Putnam|Alfonsetti|Female|  167.97.48.246|
|                null|      null|     null|  null|         null|  lbutland7@time.com|     Lilas|   Butland|Female|109.110.131.151|
+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+

注意到email这一列出现了重复，并且对于在两个数据框中都不存在的email，显示为null。

接下来是下面这段代码：

users1_df. \
    join(users2_df, 'email', 'full_outer'). \
    show()

我得到的结果是：

+--------------------+----------+---------+------+-------------+----------+----------+------+---------------+
|               email|first_name|last_name|gender|   ip_address|first_name| last_name|gender|     ip_address|
+--------------------+----------+---------+------+-------------+----------+----------+------+---------------+
| alovett0@nsw.gov.au|   Aundrea|   Lovett|  Male|  62.72.1.143|      null|      null|  null|           null|
|bjowling1@spiegel.de|   Bettine|  Jowling|Female|26.250.197.47|    Putnam|Alfonsetti|Female|  167.97.48.246|
|  lbutland7@time.com|      null|     null|  null|         null|     Lilas|   Butland|Female|109.110.131.151|
+--------------------+----------+---------+------+-------------+----------+----------+------+---------------+

注意到email这一列没有重复，并且也没有null值。

我是不是漏掉了什么？这个行为在pyspark.join的文档中哪里有提到呢？

Answer 1

当你只用一列作为条件时，Spark会把这列当作等值连接来处理，也就是说，它会把两个数据表中的这一列合并在一起。你可以把这个过程想象成使用coalesce(df1.email, df2.email)，正如@Lingesh所提到的，coalesce会返回提供的列中第一个不为空的值。

但是，如果你明确指定了条件，Spark会把所有的列都拿出来，你就需要自己去处理这些列了。

Answer 2

我花了一些时间去理解官方文档和代码之间的差异。以下是我的发现：

你提到了join方法的两种不同用法：

1. 明确的连接条件

users1_df. \
    join(users2_df, users1_df.email == users2_df.email, 'full_outer'). \
    show()

在后端代码中，有以下信息作为一个函数的文档：

当连接条件明确写出时：df.name == df2.name，这会产生所有名字匹配的记录，以及那些不匹配的记录（因为这是外连接）。如果在df2中有一些名字在df中不存在，它们会在df的name列中显示为NULL，反之亦然。

这意味着当你执行show操作时，我们会看到所有的列，包括重复的连接列email，最终输出如下：

+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+
|email               |first_name|last_name|gender|ip_address   |email               |first_name|last_name |gender|ip_address     |
+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+
|alovett0@nsw.gov.au |Aundrea   |Lovett   |Male  |62.72.1.143  |null                |null      |null      |null  |null           |
|bjowling1@spiegel.de|Bettine   |Jowling  |Female|26.250.197.47|bjowling1@spiegel.de|Putnam    |Alfonsetti|Female|167.97.48.246  |
|null                |null      |null     |null  |null         |lbutland7@time.com  |Lilas     |Butland   |Female|109.110.131.151|
+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+

2. 隐式连接条件

另一种用法是使用join方法的隐式语法：

users1_df. \
    join(users2_df, 'email', 'full_outer'). \
    show()

同样，GitHub的函数文档指出：

当你直接提供列名作为连接条件时，Spark会将两个名字列视为一个，不会为df.name和df2.name生成单独的列。这避免了输出中出现重复的列。

这意味着明确的连接条件的结果等同于下面的隐式调用：

(users1_df.alias("u1")
 .join(users2_df.alias("u2"), 'email', 'full_outer')
 .selectExpr("u1.email", "u1.first_name", "u1.last_name", "u1.gender", "u1.ip_address",
             "u2.email", "u2.first_name", "u2.last_name", "u2.gender", "u2.ip_address")
 .show(1000, False))

在这里，我调用了隐式语法，但对两个数据框users1_df和users2_df使用了别名（u1和u2），然后最终明确列出了所有的列，输出与明确连接条件产生的结果类似：

+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+
|email               |first_name|last_name|gender|ip_address   |email               |first_name|last_name |gender|ip_address     |
+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+
|alovett0@nsw.gov.au |Aundrea   |Lovett   |Male  |62.72.1.143  |null                |null      |null      |null  |null           |
|bjowling1@spiegel.de|Bettine   |Jowling  |Female|26.250.197.47|bjowling1@spiegel.de|Putnam    |Alfonsetti|Female|167.97.48.246  |
|null                |null      |null     |null  |null         |lbutland7@time.com  |Lilas     |Butland   |Female|109.110.131.151|
+--------------------+----------+---------+------+-------------+--------------------+----------+----------+------+---------------+

有趣的是，如果我们在连接列email上应用coalesce方法，我们会得到只有一个email列的输出：

(users1_df.alias("u1")
 .join(users2_df.alias("u2"), 'email', 'full_outer')
 .selectExpr("coalesce(u1.email, u2.email) as email", "u1.first_name", "u1.last_name", "u1.gender",
             "u1.ip_address", "u2.first_name", "u2.last_name", "u2.gender", "u2.ip_address")
 .show(1000, False))

最终输出如下：

+--------------------+----------+---------+------+-------------+----------+----------+------+---------------+
|email               |first_name|last_name|gender|ip_address   |first_name|last_name |gender|ip_address     |
+--------------------+----------+---------+------+-------------+----------+----------+------+---------------+
|alovett0@nsw.gov.au |Aundrea   |Lovett   |Male  |62.72.1.143  |null      |null      |null  |null           |
|bjowling1@spiegel.de|Bettine   |Jowling  |Female|26.250.197.47|Putnam    |Alfonsetti|Female|167.97.48.246  |
|lbutland7@time.com  |null      |null     |null  |null         |Lilas     |Butland   |Female|109.110.131.151|
+--------------------+----------+---------+------+-------------+----------+----------+------+---------------+

这个coalesce方法生成了唯一的email列，正如GitHub文档中所述，它只是避免了输出中出现重复的列。这是我进行的一个实验，旨在解释pyspark在隐式连接时如何决定生成输出。如果我找到实现这个功能的方法，我会相应地更新这个回答。