Spark权威指南(中文版)----第8章 表连接joins

第7章讨论了聚合单个数据集，这很有帮助，但通常情况下，您的Spark应用程序将汇集大量不同的数据集。因此，连接几乎是所有Spark工作负载的重要组成部分。Spark能够与不同的数据进行对话，这意味着您能够访问公司内的各种数据源。本章不仅介绍了Spark中存在哪些连接以及如何使用它们，还介绍了一些基本的内部机制，以便您可以考虑Spark实际上是如何在集群上执行连接的。这些基本知识可以帮助你避免内存不足，并解决你以前无法解决的问题。

8.1 Join表达式

join汇集了两组数据,左边和右边,通过比较左右两组数据的一个或多个key(键),计算join表达式的结果来决定Spark是否应该汇集左边数据集和右边的数据集。最常见的连接表达式是等连接，它比较左右数据集中指定的键是否相等。如果它们相等，Spark将组合左和右数据集。对于不匹配的键则相反;Spark丢弃没有匹配键的行。除了等连接之外，Spark还允许更复杂的连接策略。我们甚至可以使用复杂类型并执行一些操作，比如在执行连接时检查数组中是否存在键。

8.2.   Join类型

连接表达式决定是否应该连接两行，而连接类型决定结果集中应该包含什么。Spark中有多种不同的连接类型可供您使用:

• Inner joins (keep rows with keys that exist in the left and rightdatasets)

• Outer joins (keep rows with keys in either the left or rightdatasets)

• Left outer joins (keep rows with keys in the left dataset)

• Right outer joins (keep rows with keys in the right dataset)

• Left semi joins (keep the rows in the left, and only the left,dataset where the key appears in the right dataset)

• Left anti joins (keep the rows in the left, and only the left,dataset where they do not appear in the right dataset)

• Natural joins (perform a join by implicitly matching the columnsbetween the two datasets with the same names)

• Cross (or Cartesian) joins (match every row in the left dataset withevery row in the right dataset)

如果您曾经与关系数据库系统，甚至Excel电子表格进行过交互，那么将不同数据集连接在一起的概念不应该太抽象。让我们继续展示每种连接类型的示例。这将使您更容易理解如何将这些应用到您自己的问题中。要做到这一点，让我们创建一些简单的数据集，可以用在我们的例子中:

// in Scalaval person = Seq(    (0, "Bill Chambers", 0, Seq(100)),    (1, "Matei Zaharia", 1, Seq(500, 250, 100)),    (2, "Michael Armbrust", 1, Seq(250, 100)))  .toDF("id", "name", "graduate_program", "spark_status")val graduateProgram = Seq(    (0, "Masters", "School of Information", "UC Berkeley"),    (2, "Masters", "EECS", "UC Berkeley"),    (1, "Ph.D.", "EECS", "UC Berkeley"))  .toDF("id", "degree", "department", "school")val sparkStatus = Seq(    (500, "Vice President"),    (250, "PMC Member"),    (100, "Contributor"))  .toDF("id", "status")# in Pythonperson = spark.createDataFrame([    (0, "Bill Chambers", 0, [100]),    (1, "Matei Zaharia", 1, [500, 250, 100]),    (2, "Michael Armbrust", 1, [250, 100])])  .toDF("id", "name", "graduate_program", "spark_status")graduateProgram = spark.createDataFrame([    (0, "Masters", "School of Information", "UC Berkeley"),    (2, "Masters", "EECS", "UC Berkeley"),    (1, "Ph.D.", "EECS", "UC Berkeley")])  .toDF("id", "degree", "department", "school")sparkStatus = spark.createDataFrame([    (500, "Vice President"),    (250, "PMC Member"),    (100, "Contributor")])  .toDF("id", "status")

接下来，让我们将这些表注册为表，以便在整个章节中使用它们:

person.createOrReplaceTempView("person") graduateProgram.createOrReplaceTempView("graduateProgram") sparkStatus.createOrReplaceTempView("sparkStatus")

8.3.   Innerjoin

内部连接计算两个DataFrame或表中的key，并且只包含(并连接在一起)计算为true的行。在下面的例子中，我们将graduateProgram DataFrame与person DataFrame连接起来，创建一个新的DataFrame:

// in Scalaval joinExpression = person.col("graduate_program") === graduateProgram.col("id")# in PythonjoinExpression = person["graduate_program"] == graduateProgram['id']

两个DataFrame中不存在的键将不会显示在结果DataFrame中。例如，下面的表达式将导致生成的DataFrame中的值为零:

// in Scalaval wrongJoinExpression = person.col("name") === graduateProgram.col("school")# in PythonwrongJoinExpression = person["name"] == graduateProgram["school"]

内部连接是默认连接，所以我们只需要在连接表达式中指定左边的DataFrame和右边的连接:

person.join(graduateProgram, joinExpression).show()-- in SQLSELECT * FROM person JOIN graduateProgram  ON person.graduate_program = graduateProgram.id

我们还可以通过传递第三个参数joinType显式地指定它:

// in Scalavar joinType = "inner"# in PythonjoinType = "inner"person.join(graduateProgram, joinExpression, joinType).show()-- in SQLSELECT * FROM person INNER JOIN graduateProgram  ON person.graduate_program = graduateProgram.id

8.4.   OuterJoins

外部连接对两个DataFrame或表中的键求值，并包含(并连接在一起)求值为true或false的行。如果在左或右DataFrame中没有对应的行，Spark将插入null:

joinType = "outer"person.join(graduateProgram, joinExpression, joinType).show()-- in SQL SELECT * FROM person FULL OUTER JOIN graduateProgram ON graduate_program = graduateProgram.id

8.5.   LeftOuter Joins

左外连接计算两个DataFrame或表中的键值，并包含来自左DataFrame的所有行以及右DataFrame中与左DataFrame匹配的任何行。如果在右边的DataFrame中没有对应的行，Spark将插入null:

joinType = "left_outer"graduateProgram.join(person, joinExpression, joinType).show()-- in SQLSELECT * FROM graduateProgram LEFT OUTER JOIN person  ON person.graduate_program = graduateProgram.id

8.6.   RightOuter Joins

右外连接计算两个DataFrame或表中的键值，包括来自右DataFrame的所有行，以及来自左DataFrame中与右DataFrame匹配的任何行。如果在左边的DataFrame中没有对应的行，Spark将插入null:

joinType = "right_outer"person.join(graduateProgram, joinExpression, joinType).show()-- in SQLSELECT * FROM person RIGHT OUTER JOIN graduateProgram  ON person.graduate_program = graduateProgram.id

8.7.   LeftSemi Joins

半连接与其他连接有点不同。它们实际上不包含来自右 DataFrame的任何值。它们只比较值，以查看值是否存在于第二个DataFrame中。如果值确实存在，这些行将保存在结果中，即使在左边的DataFrame中有重复的键。将左半连接看作DataFrame上的过滤器，而不是传统连接的功能:

joinType = "left_semi"graduateProgram.join(person, joinExpression, joinType).show()

// in Scalaval gradProgram2 = graduateProgram.union(Seq(    (0, "Masters", "Duplicated Row", "Duplicated School")).toDF())
gradProgram2.createOrReplaceTempView("gradProgram2")# in PythongradProgram2 = graduateProgram.union(spark.createDataFrame([    (0, "Masters", "Duplicated Row", "Duplicated School")]))
gradProgram2.createOrReplaceTempView("gradProgram2")gradProgram2.join(person, joinExpression, joinType).show()-- in SQLSELECT * FROM gradProgram2 LEFT SEMI JOIN person  ON gradProgram2.id = person.graduate_program

8.8.   LeftAnti Joins

左反连接是左半连接的反义词。与左半连接一样，它们实际上不包含来自右DataFrame的任何值。它们只比较值，以查看值是否存在于第二个DataFrame中。但是，它们只保留第二个DataFrame中没有对应键的值，而不保留第二个DataFrame中存在的值。把反连接看作一个not in sql风格的过滤器:

joinType = "left_anti"graduateProgram.join(person, joinExpression, joinType).show()-- in SQLSELECT * FROM graduateProgram LEFT ANTI JOIN person  ON graduateProgram.id = person.graduate_program

8.9.   NaturalJoins

自然连接对要连接的列进行隐式猜测。它找到匹配的列并返回结果。支持左、右和外部自然连接。

警告

Implicit总是危险的!下面的查询将给出不正确的结果，因为这两个DataFrames/tables共享一个列名(id)，但它在数据集中表示不同的内容。您应该始终谨慎使用此连接。

-- in SQLSELECT * FROMgraduateProgram NATURAL JOIN person

8.10.   Cross(Cartesian) Joins

最后一个连接是交叉连接或笛卡尔积。交叉连接在最简单的术语中是不指定谓词的内部连接。交叉连接将把左边DataFrame中的每一行连接到右边DataFrame中的任何一行。这将导致结果DataFrame中包含的行数绝对激增。如果每个DataFrame中有1,000行，那么这些数据的交叉连接将产生1,000,000 (1,000 x 1,000)行。因此，必须使用cross join关键字非常明确地声明需要交叉连接:

joinType = "cross"graduateProgram.join(person, joinExpression, joinType).show()-- in SQLSELECT * FROM graduateProgram CROSS JOIN person  ON graduateProgram.id = person.graduate_program

如果你真的想有一个交叉连接，你可以明确地写：

8.11.   使用连接时的挑战

在执行连接时，会出现一些特定的挑战和一些常见的问题。本章的其余部分将提供这些常见问题的答案，然后解释Spark在高层次上如何执行连接。这将提示我们将在本书后面的部分中讨论的一些优化。

8.11.1.复杂类型上的连接

尽管这看起来像是一个挑战，但实际上并不是。任何表达式都是有效的连接表达式，假设它返回一个布尔值:

import org.apache.spark.sql.functions.expr
person.withColumnRenamed("id", "personId")  .join(sparkStatus, expr("array_contains(spark_status, id)")).show()# in Pythonfrom pyspark.sql.functions import expr
person.withColumnRenamed("id", "personId")  .join(sparkStatus, expr("array_contains(spark_status, id)")).show()-- in SQLSELECT * FROM  (select id as personId, name, graduate_program, spark_status FROM person)  INNER JOIN sparkStatus ON array_contains(spark_status, id)

8.11.2.处理重复的列名

join中出现的一个棘手问题是处理结果DataFrame中的重复列名。在DataFrame中，每个列在Spark的SQL引擎Catalyst中都有一个惟一的ID。这个惟一ID纯粹是内部的，不能直接引用。当您有一个具有重复列名的DataFrame时，这使得引用特定列非常困难。让我们创建一个问题数据集，我们可以用它来说明这些问题:

val gradProgramDupe = graduateProgram.withColumnRenamed("id", "graduate_program")val joinExpr = gradProgramDupe.col("graduate_program") === person.col("graduate_program")

注意，现在有两个graduate_program列，即使我们在这个键上连接:

person.join(gradProgramDupe, joinExpr).show()

当我们参考其中一栏时，就会遇到挑战:

person.join(gradProgramDupe, joinExpr).select("graduate_program").show()

给定前面的代码片段，我们将收到一个错误。在这个特殊的例子中，Spark生成以下消息:

org.apache.spark.sql.AnalysisException: Reference 'graduate_program' is ambiguous, could be: graduate_program#40, graduate_program#1079.;

方法1:不同的连接表达式

当您有两个名称相同的键时，最简单的修复方法可能是将join表达式从Boolean表达式更改为字符串或序列。这将在连接期间自动删除其中一列：

person.join(gradProgramDupe,"graduate_program").select("graduate_program").show()

方法2:在连接之后删除列

另一种方法是在连接之后删除有问题的列。这样做时，我们需要通过原始源DataFrame引用列。如果连接使用相同的键名，或者源数据名的列具有相同的名称，我们可以这样做:

person.join(gradProgramDupe, joinExpr).drop(person.col("graduate_program"))  .select("graduate_program").show()val joinExpr = person.col("graduate_program") === graduateProgram.col("id")person.join(graduateProgram, joinExpr).drop(graduateProgram.col("id")).show()

这是Spark的SQL分析过程的一个模块，其中显式引用的列将通过分析，因为Spark不需要解析列。注意该列如何使用.col方法而不是列函数。这允许我们通过列的特定ID隐式地指定该列。

方法3:在连接之前重命名列

如果我们在连接之前重命名其中一列，就可以完全避免这个问题:

val gradProgram3 = graduateProgram.withColumnRenamed("id", "grad_id") val joinExpr = person.col("graduate_program") === gradProgram3.col("grad_id") person.join(gradProgram3, joinExpr).show()

8.12.   Spark如何执行连接

要理解Spark如何执行连接，您需要理解两个核心资源:节点到节点的通信策略和每个节点的计算策略。这些内部机制可能与您的业务问题无关。然而，理解Spark如何执行连接可能意味着快速完成的任务和根本不完成的任务之间的区别。

8.12.1.连接策略

Spark在连接期间以两种不同的方式处理集群通信。它要么引发shuffle连接，从而导致all-to-all通信或广播连接。请记住，有很多细节比我们在这一点上透露的更多，这是有意的。随着基于成本的优化器的新改进和通信策略的改进，其中一些内部优化可能会随着时间的推移而改变。出于这个原因,我们将专注于高级的例子来帮助您理解到底发生了什么在的一些比较常见的场景中,并让你利用一些唾手可得的,您可以使用马上来加快你的一些工作负载。我们简化连接视图的核心基础是，在Spark中，您将拥有一个大表或一个小表。虽然这显然是一个频谱(如果您有一个“中型表”，事情的发生也会有所不同)，但是为了解释这个原因，对这种区别进行这样处理可能会有所帮助。

大表对大表

当您将一个大表连接到另一个大表时，您将得到一个shuffle连接，如图8-1所示。

在shuffle join中，每个节点与每个其他节点通信，它们根据哪个节点具有某个键或一组键(您正在连接的键)共享数据。这些连接非常昂贵，因为网络可能会因为流量而变得拥挤，尤其是在数据分区不好的情况下。这个连接描述将一个大数据表连接到另一个大数据表。例如，一家公司每天从物联网接收数十亿条信息，需要识别每天发生的变化。方法是在一列中加入deviceId、messageType和date，在另一列中加入date - 1天。在图8-1中，DataFrame1和DataFrame 2都是大型数据aframe。这意味着所有工作节点(可能还有每个分区)都需要在整个连接过程中彼此通信(没有对数据进行智能分区)。大表对小表如果还有一些多余的资源空间时，我们可以优化连接。虽然我们可以使用大表到大表的通信策略，但是使用广播连接通常会更有效。这意味着我们将把我们的小DataFrame复制到集群中的每个工作节点上(无论它位于一台机器上还是多台机器上)。这听起来很耗资源。然而，这将阻止我们在整个连接过程中执行all-to-all通信。相反，我们在开始时只执行一次，然后让每个单独的工作节点执行工作，而不必等待或与任何其他工作节点通信，如图8-2所示。

在此连接的开始，将进行大型通信，就像前面的连接类型一样。然而，紧接在第一个节点之后，节点之间将不再通信。这意味着连接将分别在每个节点上执行，这使得CPU成为最大的瓶颈。对于我们当前的数据集，我们可以看到Spark通过查看explain计划自动将其设置为广播连接:

使用DataFrame API，我们还可以显式地向优化器提示，我们希望使用广播连接，方法是围绕所讨论的小DataFrame使用正确的函数。在本例中，这些结果与我们刚才看到的计划相同;然而，情况并非总是如此:

import org.apache.spark.sql.functions.broadcastval joinExpr = person.col("graduate_program") === graduateProgram.col("id")person.join(broadcast(graduateProgram), joinExpr).explain()

SQL接口还包括提供执行连接的提示的功能。但是，这些没有强制执行，因此优化器可能选择忽略它们。您可以使用特殊的注释语法设置其中一个提示。MAPJOIN、BROADCAST和BROADCASTJOIN都做同样的事情，并且都得到了支持:

-- in SQLSELECT /*+ MAPJOIN(graduateProgram) */ * FROM person JOIN graduateProgram  ON person.graduate_program = graduateProgram.id

这也不是免费的:如果您试图广播太大的东西，您可能会崩溃驱动程序节点(因为收集是昂贵的)。这可能是未来优化的一个领域。小表对小表在对小表执行连接时，通常最好让Spark决定如何连接它们。如果注意到奇怪的行为，可以强制广播连接。

8.13.   结束语

在本章中，我们讨论了join，这可能是最常见的用例之一。我们没有提及但很重要考虑的是如果你分区数据正确连接之前,你可以得到更有效的执行,因为即使改组计划,如果数据从两个不同的DataFrames已经坐落在同一台机器上,Spark可以避免shuffle。对一些数据进行试验，并尝试预先分区，看看在执行这些连接时是否可以注意到速度的提高。在第9章中，我们将讨论Spark的数据源api。当您决定连接应该以什么顺序进行时，还有其他的含义。因为一些连接充当过滤器，所以这可以降低工作负载的挂起可能性，因为可以保证减少通过网络交换的数据。下一章将不再讨论用户操作，正如我们在前几章中看到的，而是讨论使用结构化api读写数据。