【Apache Spark 】第 5 章Spark SQL 和 DataFrames：与外部数据源交互

用户定义的函数

虽然 Apache Spark 有大量的内置函数，但 Spark 的灵活性允许数据工程师和数据科学家也可以定义自己的函数。这些被称为用户定义函数(UDF)。

// In Scala
// Create cubed function
val cubed = (s: Long) => {
  s * s * s
}
 
// Register UDF
spark.udf.register("cubed", cubed)
 
// Create temporary view
spark.range(1, 9).createOrReplaceTempView("udf_test")

# In Python
from pyspark.sql.types import LongType
 
# Create cubed function
def cubed(s):
  return s * s * s
 
# Register UDF
spark.udf.register("cubed", cubed, LongType())
 
# Generate temporary view
spark.range(1, 9).createOrReplaceTempView("udf_test")

// In Scala/Python
// Query the cubed UDF
spark.sql("SELECT id, cubed(id) AS id_cubed FROM udf_test").show()
 
+---+--------+
| id|id_cubed|
+---+--------+
|  1|       1|
|  2|       8|
|  3|      27|
|  4|      64|
|  5|     125|
|  6|     216|
|  7|     343|
|  8|     512|
+---+--------+

spark.sql("SELECT s FROM test1 WHERE s IS NOT NULL AND strlen(s) > 1")

# In Python
# Import pandas
import pandas as pd
 
# Import various pyspark SQL functions including pandas_udf
from pyspark.sql.functions import col, pandas_udf
from pyspark.sql.types import LongType
 
# Declare the cubed function 
def cubed(a: pd.Series) -> pd.Series:
    return a * a * a
 
# Create the pandas UDF for the cubed function 
cubed_udf = pandas_udf(cubed, returnType=LongType())

# Create a Pandas Series
x = pd.Series([1, 2, 3])
 
# The function for a pandas_udf executed with local Pandas data
print(cubed(x))

0     1
1     8
2    27
dtype: int64

# Create a Spark DataFrame, 'spark' is an existing SparkSession
df = spark.range(1, 4)
 
# Execute function as a Spark vectorized UDF
df.select("id", cubed_udf(col("id"))).show()

+---+---------+
| id|cubed(id)|
+---+---------+
|  1|        1|
|  2|        8|
|  3|       27|
+---+---------+

使用 Spark SQL Shell

执行 Spark SQL 查询的便捷工具是spark-sqlCLI。虽然此实用程序在本地模式下与 Hive 元存储服务通信，但它不与Thrift JDBC/ODBC 服务器（又名Spark Thrift 服务器或STS）通信。STS 允许 JDBC/ODBC 客户端通过 Apache Spark 上的 JDBC 和 ODBC 协议执行 SQL 查询。

要启动 Spark SQL CLI，请在文件夹中执行以下命令$SPARK_HOME：

./bin/spark-sql

启动 shell 后，您可以使用它以交互方式执行 Spark SQL 查询。让我们看几个例子。

spark-sql> CREATE TABLE people (name STRING,age int);

20/01/11 22:42:16 WARN HiveMetaStore: Location: file:/user/hive/warehouse/people
specified for non-external table:people
Time taken: 0.63 seconds

INSERT INTO people SELECT name, age FROM ...

spark-sql> INSERT INTO people VALUES ("Michael", NULL);
Time taken: 1.696 seconds
spark-sql> INSERT INTO people VALUES ("Andy", 30);
Time taken: 0.744 seconds
spark-sql> INSERT INTO people VALUES ("Samantha", 19);
Time taken: 0.637 seconds
spark-sql>

spark-sql> SHOW TABLES;
default   people     false
Time taken: 0.016 seconds, Fetched 1 row(s)

spark-sql> SELECT * FROM people WHERE age < 20;
Samantha  19
Time taken: 0.593 seconds, Fetched 1 row(s)

spark-sql> SELECT name FROM people WHERE age IS NULL;
Michael
Time taken: 0.272 seconds, Fetched 1 row(s)

如果您使用过 Apache Hive，您可能熟悉命令行工具Beeline，这是一种针对 HiveServer2 运行 HiveQL 查询的常用实用程序。Beeline 是一个基于SQLLine CLI的 JDBC 客户端。您可以使用相同的实用程序对 Spark Thrift 服务器执行 Spark SQL 查询。请注意，当前实现的 Thrift JDBC/ODBC 服务器对应于 Hive 1.2.1 中的 HiveServer2。您可以使用 Spark 或 Hive 1.2.1 附带的以下 Beeline 脚本来测试 JDBC 服务器。

./sbin/start-thriftserver.sh

./bin/beeline

!connect jdbc:hive2://localhost:10000

0: jdbc:hive2://localhost:10000> SHOW tables;
 
+-----------+------------+--------------+
| database  | tableName  | isTemporary  |
+-----------+------------+--------------+
| default   | people     | false        |
+-----------+------------+--------------+
1 row selected (0.417 seconds)
 
0: jdbc:hive2://localhost:10000> SELECT * FROM people;
 
+-----------+-------+
|   name    |  age  |
+-----------+-------+
| Samantha  | 19    |
| Andy      | 30    |
| Michael   | NULL  |
+-----------+-------+
3 rows selected (1.512 seconds)
 
0: jdbc:hive2://localhost:10000>

./sbin/stop-thriftserver.sh

使用 Tableau

与通过 Beeline 或 Spark SQL CLI 运行查询类似，您可以通过 Thrift JDBC/ODBC 服务器将您最喜欢的 BI 工具连接到 Spark SQL。在本节中，我们将向您展示如何将 Tableau Desktop（版本 2019.2）连接到您的本地 Apache Spark 实例。

./sbin/start-thriftserver.sh

启动 Tableau

如果您是第一次启动 Tableau，您会看到一个“连接”对话框，允许您连接到大量数据源。默认情况下，左侧的“To a Server”菜单中不会包含 Spark SQL 选项（见图 5-2）。

图 5-2。Tableau 连接对话框

要访问 Spark SQL 选项，请单击该列表底部的更多...，然后从主面板中显示的列表中选择 Spark SQL，如图 5-3所示。

图 5-3。选择更多... > Spark SQL 以连接到 Spark SQL

这将弹出 Spark SQL 对话框（图 5-4）。当您连接到本地 Apache Spark 实例时，您可以使用具有以下参数的非安全用户名身份验证模式：

• 服务器：本地主机

• 端口：10000（默认）

• 类型：SparkThriftServer（默认）

• 身份验证：用户名

• 用户名：您的登录名，例如 user@learningspark.org

• 需要 SSL：未选中

图 5-4。Spark SQL 对话框

成功连接到 Spark SQL 数据源后，您将看到类似于图 5-5的 Data Source Connections 视图。

图 5-5。Tableau 数据源连接视图，连接到本地 Spark 实例

从左侧的 Select Schema 下拉菜单中，选择“default”。然后输入要查询的表名（见图5-6）。请注意，您可以单击放大镜图标来获取可用表格的完整列表。

图 5-6。选择要查询的架构和表

笔记

有关使用 Tableau 连接到 Spark SQL 数据库的详细信息，请参阅 Tableau 的Spark SQL 文档和 Databricks Tableau 文档。

输入people表格名称，然后将表格从左侧拖放到主对话框中（在标有“将表格拖到此处”的空间中）。您应该看到类似于图 5-7的内容。

图 5-7。连接到本地 Spark 实例中的人员表

单击立即更新，Tableau 将在后台查询您的 Spark SQL 数据源（图 5-8）。

您现在可以对 Spark 数据源、联接表等执行查询，就像对任何其他 Tableau 数据源一样。

图 5-8。查询本地 Spark 数据源的 Tableau 工作表表视图

./sbin/stop-thriftserver.sh

JDBC 和 SQL 数据库

Spark SQL 包含一个数据源 API，可以使用JDBC从其他数据库读取数据。它简化了对这些数据源的查询，因为它将结果作为 DataFrame 返回，从而提供了 Spark SQL 的所有优点（包括性能和与其他数据源连接的能力）。

首先，您需要为您的 JDBC 数据源指定 JDBC 驱动程序，并且它需要位于 Spark 类路径中。从$SPARK_HOME文件夹中，您将发出如下命令：

./bin/spark-shell --driver-class-path $database.jar --jars $database.jar

使用数据源 API，可以将远程数据库中的表加载为 DataFrame 或 Spark SQL 临时视图。用户可以在数据源选项中指定 JDBC 连接属性。表 5-1包含 Spark 支持的一些更常见的连接属性（不区分大小写）。

有关连接属性的完整列表，请参阅Spark SQL 文档。

PostgreSQL

要连接到 PostgreSQL 数据库，请从Maven构建或下载 JDBC jar并将其添加到您的类路径中。然后启动 Spark shell（spark-shell或pyspark），指定该 jar：

bin/spark-shell --jars postgresql-42.2.6.jar

以下示例展示了如何在 Scala 中使用 Spark SQL 数据源 API 和 JDBC 从 PostgreSQL 数据库中加载和保存：

// In Scala
// Read Option 1: Loading data from a JDBC source using load method
val jdbcDF1 = spark
  .read
  .format("jdbc")
  .option("url", "jdbc:postgresql:[DBSERVER]")
  .option("dbtable", "[SCHEMA].[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .load()
 
// Read Option 2: Loading data from a JDBC source using jdbc method
// Create connection properties
import java.util.Properties
val cxnProp = new Properties()
cxnProp.put("user", "[USERNAME]") 
cxnProp.put("password", "[PASSWORD]")
 
// Load data using the connection properties
val jdbcDF2 = spark
  .read
  .jdbc("jdbc:postgresql:[DBSERVER]", "[SCHEMA].[TABLENAME]", cxnProp)
 
// Write Option 1: Saving data to a JDBC source using save method
jdbcDF1
  .write
  .format("jdbc")
  .option("url", "jdbc:postgresql:[DBSERVER]")
  .option("dbtable", "[SCHEMA].[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .save()
 
// Write Option 2: Saving data to a JDBC source using jdbc method
jdbcDF2.write
  .jdbc(s"jdbc:postgresql:[DBSERVER]", "[SCHEMA].[TABLENAME]", cxnProp)

以下是在 PySpark 中的操作方法：

# In Python
# Read Option 1: Loading data from a JDBC source using load method
jdbcDF1 = (spark
  .read
  .format("jdbc") 
  .option("url", "jdbc:postgresql://[DBSERVER]")
  .option("dbtable", "[SCHEMA].[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .load())
 
# Read Option 2: Loading data from a JDBC source using jdbc method
jdbcDF2 = (spark
  .read 
  .jdbc("jdbc:postgresql://[DBSERVER]", "[SCHEMA].[TABLENAME]",
          properties={"user": "[USERNAME]", "password": "[PASSWORD]"}))
 
# Write Option 1: Saving data to a JDBC source using save method
(jdbcDF1
  .write
  .format("jdbc")
  .option("url", "jdbc:postgresql://[DBSERVER]")
  .option("dbtable", "[SCHEMA].[TABLENAME]") 
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .save())
 
# Write Option 2: Saving data to a JDBC source using jdbc method
(jdbcDF2
  .write 
  .jdbc("jdbc:postgresql:[DBSERVER]", "[SCHEMA].[TABLENAME]",
          properties={"user": "[USERNAME]", "password": "[PASSWORD]"}))

MySQL

要连接到 MySQL 数据库，请从Maven或MySQL（后者更容易！）构建或下载 JDBC jar，并将其添加到您的类路径中。然后启动 Spark shell（spark-shell或pyspark），指定该 jar：

bin/spark-shell --jars mysql-connector-java_8.0.16-bin.jar

以下示例展示了如何使用 Spark SQL 数据源 API 和 Scala 中的 JDBC 从 MySQL 数据库中加载数据并将其保存到数据库中：

// In Scala
// Loading data from a JDBC source using load 
val jdbcDF = spark
  .read
  .format("jdbc")
  .option("url", "jdbc:mysql://[DBSERVER]:3306/[DATABASE]")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .load()
 
// Saving data to a JDBC source using save 
jdbcDF
  .write
  .format("jdbc")
  .option("url", "jdbc:mysql://[DBSERVER]:3306/[DATABASE]")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .save()

下面是如何在 Python 中做到这一点：

# In Python
# Loading data from a JDBC source using load 
jdbcDF = (spark
  .read
  .format("jdbc")
  .option("url", "jdbc:mysql://[DBSERVER]:3306/[DATABASE]")
  .option("driver", "com.mysql.jdbc.Driver") 
  .option("dbtable", "[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .load())
 
# Saving data to a JDBC source using save 
(jdbcDF
  .write 
  .format("jdbc") 
  .option("url", "jdbc:mysql://[DBSERVER]:3306/[DATABASE]")
  .option("driver", "com.mysql.jdbc.Driver") 
  .option("dbtable", "[TABLENAME]") 
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .save())

Azure Cosmos 数据库

若要连接到 Azure Cosmos DB 数据库，请从Maven或GitHub构建或下载 JDBC jar并将其添加到您的类路径中。然后启动一个 Scala 或 PySpark shell，指定这个 jar（注意这个例子使用的是 Spark 2.4）：

bin/spark-shell --jars azure-cosmosdb-spark_2.4.0_2.11-1.3.5-uber.jar

您还可以选择使用其 Maven 坐标从Spark 包--packages中拉出连接器：

export PKG="com.microsoft.azure:azure-cosmosdb-spark_2.4.0_2.11:1.3.5"
bin/spark-shell --packages $PKG

以下示例说明如何使用 Scala 和 PySpark 中的 Spark SQL 数据源 API 和 JDBC 从 Azure Cosmos DB 数据库加载数据并将其保存到。请注意，通常使用query_custom配置来利用 Cosmos DB 中的各种索引：

// In Scala
// Import necessary libraries
import com.microsoft.azure.cosmosdb.spark.schema._
import com.microsoft.azure.cosmosdb.spark._
import com.microsoft.azure.cosmosdb.spark.config.Config
 
// Loading data from Azure Cosmos DB
// Configure connection to your collection
val query = "SELECT c.colA, c.coln FROM c WHERE c.origin = 'SEA'"
val readConfig = Config(Map(
  "Endpoint" -> "https://[ACCOUNT].documents.azure.com:443/", 
  "Masterkey" -> "[MASTER KEY]",
  "Database" -> "[DATABASE]",
  "PreferredRegions" -> "Central US;East US2;",
  "Collection" -> "[COLLECTION]",
  "SamplingRatio" -> "1.0",
  "query_custom" -> query
))
 
// Connect via azure-cosmosdb-spark to create Spark DataFrame
val df = spark.read.cosmosDB(readConfig)
df.count
 
// Saving data to Azure Cosmos DB
// Configure connection to the sink collection
val writeConfig = Config(Map(
  "Endpoint" -> "https://[ACCOUNT].documents.azure.com:443/",
  "Masterkey" -> "[MASTER KEY]",
  "Database" -> "[DATABASE]",
  "PreferredRegions" -> "Central US;East US2;",
  "Collection" -> "[COLLECTION]",
  "WritingBatchSize" -> "100"
))
 
// Upsert the DataFrame to Azure Cosmos DB
import org.apache.spark.sql.SaveMode
df.write.mode(SaveMode.Overwrite).cosmosDB(writeConfig)

# In Python
# Loading data from Azure Cosmos DB
# Read configuration
query = "SELECT c.colA, c.coln FROM c WHERE c.origin = 'SEA'"
readConfig = {
  "Endpoint" : "https://[ACCOUNT].documents.azure.com:443/", 
  "Masterkey" : "[MASTER KEY]",
  "Database" : "[DATABASE]",
  "preferredRegions" : "Central US;East US2",
  "Collection" : "[COLLECTION]",
  "SamplingRatio" : "1.0",
  "schema_samplesize" : "1000",
  "query_pagesize" : "2147483647",
  "query_custom" : query
}
 
# Connect via azure-cosmosdb-spark to create Spark DataFrame
df = (spark
  .read
  .format("com.microsoft.azure.cosmosdb.spark")
  .options(**readConfig)
  .load())
 
# Count the number of flights
df.count()
 
# Saving data to Azure Cosmos DB
# Write configuration
writeConfig = {
 "Endpoint" : "https://[ACCOUNT].documents.azure.com:443/",
 "Masterkey" : "[MASTER KEY]",
 "Database" : "[DATABASE]",
 "Collection" : "[COLLECTION]",
 "Upsert" : "true"
}
 
# Upsert the DataFrame to Azure Cosmos DB
(df.write
  .format("com.microsoft.azure.cosmosdb.spark")
  .options(**writeConfig)
  .save())

有关详细信息，请参阅Azure Cosmos DB 文档。

微软 SQL 服务器

要连接到 MS SQL Server 数据库，请下载 JDBC jar并将其添加到您的类路径中。然后启动一个 Scala 或 PySpark shell，指定这个 jar：

bin/spark-shell --jars mssql-jdbc-7.2.2.jre8.jar

以下示例展示了如何使用 Scala 和 PySpark 中的 Spark SQL 数据源 API 和 JDBC 从 MS SQL Server 数据库加载数据并将其保存到：

// In Scala
// Loading data from a JDBC source
// Configure jdbcUrl
val jdbcUrl = "jdbc:sqlserver://[DBSERVER]:1433;database=[DATABASE]"
 
// Create a Properties() object to hold the parameters. 
// Note, you can create the JDBC URL without passing in the
// user/password parameters directly.
val cxnProp = new Properties()
cxnProp.put("user", "[USERNAME]") 
cxnProp.put("password", "[PASSWORD]") 
cxnProp.put("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
 
// Load data using the connection properties
val jdbcDF = spark.read.jdbc(jdbcUrl, "[TABLENAME]", cxnProp)
 
// Saving data to a JDBC source
jdbcDF.write.jdbc(jdbcUrl, "[TABLENAME]", cxnProp)

# In Python
# Configure jdbcUrl
jdbcUrl = "jdbc:sqlserver://[DBSERVER]:1433;database=[DATABASE]"
 
# Loading data from a JDBC source
jdbcDF = (spark
  .read
  .format("jdbc") 
  .option("url", jdbcUrl)
  .option("dbtable", "[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .load())
 
# Saving data to a JDBC source
(jdbcDF
  .write
  .format("jdbc") 
  .option("url", jdbcUrl)
  .option("dbtable", "[TABLENAME]")
  .option("user", "[USERNAME]")
  .option("password", "[PASSWORD]")
  .save())

其他外部来源

Apache Spark 可以连接的许多外部数据源只是其中的一部分；其他流行的数据源包括：

• 阿帕奇卡桑德拉

• 雪花

• MongoDB

选项 1：爆炸和收集

在这个嵌套的 SQL 语句中，我们首先为 中的每个元素 ( )explode(values)创建一个新行 (带有) ：idvaluevalues

-- In SQL
SELECT id, collect_list(value + 1) AS values
FROM  (SELECT id, EXPLODE(values) AS value
        FROM table) x
GROUP BY id

虽然collect_list()返回具有重复的对象列表，但该GROUP BY语句需要随机操作，这意味着重新收集的数组的顺序不一定与原始数组的顺序相同。由于values可以是任意数量的维度（一个非常宽和/或非常长的数组）并且我们正在做一个GROUP BY，这种方法可能非常昂贵。

选项 2：用户定义的函数

要执行相同的任务（添加1到 中的每个元素values），我们还可以创建一个map()用于遍历每个元素 ( value) 并执行添加操作的 UDF：

// In Scala
def addOne(values: Seq[Int]): Seq[Int] = {
    values.map(value => value + 1)
}
val plusOneInt = spark.udf.register("plusOneInt", addOne(_: Seq[Int]): Seq[Int])

然后我们可以在 Spark SQL 中使用这个 UDF，如下所示：

spark.sql("SELECT id, plusOneInt(values) A​​S values FROM table").show()

虽然这比使用更好，explode()而且collect_list()不会有任何排序问题，但序列化和反序列化过程本身可能很昂贵。然而，同样重要的是要注意，这collect_list()可能会导致执行程序遇到大型数据集的内存不足问题，而使用 UDF 可以缓解这些问题。

复杂数据类型的内置函数

除了使用这些可能很昂贵的技术外，您还可以使用一些内置函数来处理 Apache Spark 2.4 及更高版本中包含的复杂数据类型。一些比较常见的列在表 5-3（数组类型）和表 5-4（映射类型）中。

高阶函数

除了前面提到的内置函数之外，还有以匿名 lambda 函数作为参数的高阶函数。高阶函数的示例如下：

-- In SQL
transform(values, value -> lambda expression)

该transform()函数将一个数组 ( values) 和匿名函数 (lambda表达式) 作为输入。该函数通过对每个元素应用匿名函数透明地创建一个新数组，然后将结果分配给输出数组（类似于 UDF 方法，但更有效）。

让我们创建一个示例数据集，以便我们可以运行一些示例：

# In Python
from pyspark.sql.types import *
schema = StructType([StructField("celsius", ArrayType(IntegerType()))])
 
t_list = [[35, 36, 32, 30, 40, 42, 38]], [[31, 32, 34, 55, 56]]
t_c = spark.createDataFrame(t_list, schema)
t_c.createOrReplaceTempView("tC")
 
# Show the DataFrame
t_c.show()

// In Scala
// Create DataFrame with two rows of two arrays (tempc1, tempc2)
val t1 = Array(35, 36, 32, 30, 40, 42, 38)
val t2 = Array(31, 32, 34, 55, 56)
val tC = Seq(t1, t2).toDF("celsius")
tC.createOrReplaceTempView("tC")
 
// Show the DataFrame
tC.show()

这是输出：

+--------------------+
|             celsius|
+--------------------+
|[35, 36, 32, 30, ...|
|[31, 32, 34, 55, 56]|
+--------------------+

使用前面的 DataFrame，您可以运行以下高阶函数查询。

transform(array<T>, function<T, U>): array<U>

// In Scala/Python
// Calculate Fahrenheit from Celsius for an array of temperatures
spark.sql("""
SELECT celsius, 
 transform(celsius, t -> ((t * 9) div 5) + 32) as fahrenheit 
  FROM tC
""").show()
 
+--------------------+--------------------+
|             celsius|          fahrenheit|
+--------------------+--------------------+
|[35, 36, 32, 30, ...|[95, 96, 89, 86, ...|
|[31, 32, 34, 55, 56]|[87, 89, 93, 131,...|
+--------------------+--------------------+

filter(array<T>, function<T, Boolean>): array<T>

// In Scala/Python
// Filter temperatures > 38C for array of temperatures
spark.sql("""
SELECT celsius, 
 filter(celsius, t -> t > 38) as high 
  FROM tC
""").show()
 
+--------------------+--------+
|             celsius|    high|
+--------------------+--------+
|[35, 36, 32, 30, ...|[40, 42]|
|[31, 32, 34, 55, 56]|[55, 56]|
+--------------------+--------+

exists(array<T>, function<T, V, Boolean>): Boolean

// In Scala/Python
// Is there a temperature of 38C in the array of temperatures
spark.sql("""
SELECT celsius, 
       exists(celsius, t -> t = 38) as threshold
  FROM tC
""").show()
 
+--------------------+---------+
|             celsius|threshold|
+--------------------+---------+
|[35, 36, 32, 30, ...|     true|
|[31, 32, 34, 55, 56]|    false|
+--------------------+---------+

reduce(array<T>, B, function<B, T, B>, function<B, R>)

// In Scala/Python
// Calculate average temperature and convert to F
spark.sql("""
SELECT celsius, 
       reduce(
          celsius, 
          0, 
          (t, acc) -> t + acc, 
          acc -> (acc div size(celsius) * 9 div 5) + 32
        ) as avgFahrenheit 
  FROM tC
""").show()
 
+--------------------+-------------+
|             celsius|avgFahrenheit|
+--------------------+-------------+
|[35, 36, 32, 30, ...|           96|
|[31, 32, 34, 55, 56]|          105|
+--------------------+-------------+

Unions

Apache Spark 中的一个常见模式是将具有相同架构的两个不同 DataFrame 联合在一起。这可以使用以下union()方法来实现：

// Scala
// Union two tables
val bar = delays.union(foo)
bar.createOrReplaceTempView("bar")
bar.filter(expr("""origin == 'SEA' AND destination == 'SFO'
AND date LIKE '01010%' AND delay > 0""")).show()

# In Python
# Union two tables
bar = departureDelays.union(foo)
bar.createOrReplaceTempView("bar")
 
# Show the union (filtering for SEA and SFO in a specific time range)
bar.filter(expr("""origin == 'SEA' AND destination == 'SFO'
AND date LIKE '01010%' AND delay > 0""")).show()

barDataFrame 是foowith的并集delays。在 DataFrame 中使用相同的过滤标准bar，我们看到了重复的foo数据，正如预期的那样：

-- In SQL
spark.sql("""
SELECT * 
  FROM bar 
 WHERE origin = 'SEA' 
   AND destination = 'SFO' 
   AND date LIKE '01010%' 
   AND delay > 0
""").show()
 
+--------+-----+--------+------+-----------+
|    date|delay|distance|origin|destination|
+--------+-----+--------+------+-----------+
|01010710|   31|     590|   SEA|        SFO|
|01010955|  104|     590|   SEA|        SFO|
|01010730|    5|     590|   SEA|        SFO|
|01010710|   31|     590|   SEA|        SFO|
|01010955|  104|     590|   SEA|        SFO|
|01010730|    5|     590|   SEA|        SFO|
+--------+-----+--------+------+-----------+

Joins

一个常见的 DataFrame 操作是将两个 DataFrame（或表）连接在一起。默认情况下，Spark SQL 连接是inner join，选项为inner, cross, outer, full, full_outer, left, left_outer, right, right_outer, left_semi, 和left_anti. 文档中提供了更多信息（这适用于 Scala 和 Python）。

以下代码示例执行和DataFrameinner之间的默认连接：airportsfoo

// In Scala
foo.join(
  airports.as('air), 
  $"air.IATA" === $"origin"
).select("City", "State", "date", "delay", "distance", "destination").show()

# In Python
# Join departure delays data (foo) with airport info
foo.join(
  airports, 
  airports.IATA == foo.origin
).select("City", "State", "date", "delay", "distance", "destination").show()

-- In SQL
spark.sql("""
SELECT a.City, a.State, f.date, f.delay, f.distance, f.destination 
  FROM foo f
  JOIN airports a
    ON a.IATA = f.origin
""").show()

上述代码允许您从 DataFrame 中查看日期、延迟、距离和目的地信息，并从fooDataFrame 中加入城市和州信息airports：

+-------+-----+--------+-----+--------+-----------+
|   City|State|    date|delay|distance|destination|
+-------+-----+--------+-----+--------+-----------+
|Seattle|   WA|01010710|   31|     590|        SFO|
|Seattle|   WA|01010955|  104|     590|        SFO|
|Seattle|   WA|01010730|    5|     590|        SFO|
+-------+-----+--------+-----+--------+-----------+

Windowing

窗口函数使用窗口中的行（输入行的范围）中的值来返回一组值，通常以另一行的形式。使用窗口函数，可以对一组行进行操作，同时仍为每个输入行返回一个值。在本节中，我们将展示如何使用dense_rank()窗口函数；还有许多其他功能，如表 5-5 所示。

让我们先回顾一下从西雅图 (SEA)、旧金山 (SFO) 和纽约市 (JFK) 出发并前往一组特定目的地位置的航班所经历的TotalDelays（由 计算），如以下查询中所述sum(Delay)：

-- In SQL
DROP TABLE IF EXISTS departureDelaysWindow;
 
CREATE TABLE departureDelaysWindow AS
SELECT origin, destination, SUM(delay) AS TotalDelays 
  FROM departureDelays 
 WHERE origin IN ('SEA', 'SFO', 'JFK') 
   AND destination IN ('SEA', 'SFO', 'JFK', 'DEN', 'ORD', 'LAX', 'ATL') 
 GROUP BY origin, destination;
 
SELECT * FROM departureDelaysWindow
 
+------+-----------+-----------+
|origin|destination|TotalDelays|
+------+-----------+-----------+
|   JFK|        ORD|       5608|
|   SEA|        LAX|       9359|
|   JFK|        SFO|      35619|
|   SFO|        ORD|      27412|
|   JFK|        DEN|       4315|
|   SFO|        DEN|      18688|
|   SFO|        SEA|      17080|
|   SEA|        SFO|      22293|
|   JFK|        ATL|      12141|
|   SFO|        ATL|       5091|
|   SEA|        DEN|      13645|
|   SEA|        ATL|       4535|
|   SEA|        ORD|      10041|
|   JFK|        SEA|       7856|
|   JFK|        LAX|      35755|
|   SFO|        JFK|      24100|
|   SFO|        LAX|      40798|
|   SEA|        JFK|       4667|
+------+-----------+-----------+

如果您想为每个始发机场找出延误最多的三个目的地，该怎么办？您可以通过为每个来源运行三个不同的查询然后将结果合并在一起来实现这一点，如下所示：

-- In SQL
SELECT origin, destination, sum(TotalDelays) as sumTotalDelays
FROM departureDelaysWindow
WHERE origin = 'SEA'
GROUP BY origin, destination
ORDER BY sumTotalDelays DESC
LIMIT 3

其中[ORIGIN]是 、 和 的三个不同的JFK原点SEA值SFO。

但更好的方法是使用窗口函数dense_rank()来执行以下计算：

-- In SQL
spark.sql("""
SELECT origin, destination, TotalDelays, rank 
  FROM ( 
     SELECT origin, destination, TotalDelays, dense_rank() 
       OVER (PARTITION BY origin ORDER BY TotalDelays DESC) as rank 
       FROM departureDelaysWindow
  ) t 
 WHERE rank <= 3
""").show()
 
+------+-----------+-----------+----+
|origin|destination|TotalDelays|rank|
+------+-----------+-----------+----+
|   SEA|        SFO|      22293|   1|
|   SEA|        DEN|      13645|   2|
|   SEA|        ORD|      10041|   3|
|   SFO|        LAX|      40798|   1|
|   SFO|        ORD|      27412|   2|
|   SFO|        JFK|      24100|   3|
|   JFK|        LAX|      35755|   1|
|   JFK|        SFO|      35619|   2|
|   JFK|        ATL|      12141|   3|
+------+-----------+-----------+----+

通过使用dense_rank()窗口函数，我们可以快速确定三个始发城市延误最严重的目的地是：

• 西雅图 (SEA)：旧金山 (SFO)、丹佛 (DEN) 和芝加哥 (ORD)

• 旧金山 (SFO)：洛杉矶 (LAX)、芝加哥 (ORD) 和纽约 (JFK)

• 纽约 (JFK)：洛杉矶 (LAX)、旧金山 (SFO) 和亚特兰大 (ATL)

需要注意的是，每个窗口分组都需要适合单个执行程序，并且在执行期间将组合成单个分区。因此，您需要确保您的查询不是无界的（即限制您的窗口大小）.

Modifications

另一个常见的操作是对DataFrame进行修改。虽然 DataFrame 本身是不可变的，但您可以通过创建新的、不同的 DataFrame、具有不同列的操作来修改它们。（回想一下前面的章节，底层的 RDD 是不可变的——即它们不能被改变——以确保 Spark 操作存在数据沿袭。）让我们从之前的小型 DataFrame 示例开始：

// In Scala/Python
foo.show()
 
--------+-----+--------+------+-----------+
|    date|delay|distance|origin|destination|
+--------+-----+--------+------+-----------+
|01010710|   31|     590|   SEA|        SFO|
|01010955|  104|     590|   SEA|        SFO|
|01010730|    5|     590|   SEA|        SFO|
+--------+-----+--------+------+-----------+

// In Scala
import org.apache.spark.sql.functions.expr
val foo2 = foo.withColumn(
              "status", 
              expr("CASE WHEN delay <= 10 THEN 'On-time' ELSE 'Delayed' END")
           )

# In Python
from pyspark.sql.functions import expr
foo2 = (foo.withColumn(
          "status", 
          expr("CASE WHEN delay <= 10 THEN 'On-time' ELSE 'Delayed' END")
        ))

// In Scala/Python
foo2.show()
 
+--------+-----+--------+------+-----------+-------+
|    date|delay|distance|origin|destination| status|
+--------+-----+--------+------+-----------+-------+
|01010710|   31|     590|   SEA|        SFO|Delayed|
|01010955|  104|     590|   SEA|        SFO|Delayed|
|01010730|    5|     590|   SEA|        SFO|On-time|
+--------+-----+--------+------+-----------+-------+

// In Scala
val foo3 = foo2.drop("delay")
foo3.show()

# In Python
foo3 = foo2.drop("delay")
foo3.show()
 
+--------+--------+------+-----------+-------+
|    date|distance|origin|destination| status|
+--------+--------+------+-----------+-------+
|01010710|     590|   SEA|        SFO|Delayed|
|01010955|     590|   SEA|        SFO|Delayed|
|01010730|     590|   SEA|        SFO|On-time|
+--------+--------+------+-----------+-------+

// In Scala
val foo4 = foo3.withColumnRenamed("status", "flight_status")
foo4.show()

# In Python
foo4 = foo3.withColumnRenamed("status", "flight_status")
foo4.show()
 
+--------+--------+------+-----------+-------------+
|    date|distance|origin|destination|flight_status|
+--------+--------+------+-----------+-------------+
|01010710|     590|   SEA|        SFO|      Delayed|
|01010955|     590|   SEA|        SFO|      Delayed|
|01010730|     590|   SEA|        SFO|      On-time|
+--------+--------+------+-----------+-------------+

-- In SQL
SELECT destination, CAST(SUBSTRING(date, 0, 2) AS int) AS month, delay 
  FROM departureDelays 
 WHERE origin = 'SEA'
 
+-----------+-----+-----+
|destination|month|delay|
+-----------+-----+-----+
|        ORD|    1|   92|
|        JFK|    1|   -7|
|        DFW|    1|   -5|
|        MIA|    1|   -3|
|        DFW|    1|   -3|
|        DFW|    1|    1|
|        ORD|    1|  -10|
|        DFW|    1|   -6|
|        DFW|    1|   -2|
|        ORD|    1|   -3|
+-----------+-----+-----+
only showing top 10 rowsspan>

-- In SQL
SELECT * FROM (
SELECT destination, CAST(SUBSTRING(date, 0, 2) AS int) AS month, delay 
  FROM departureDelays WHERE origin = 'SEA' 
) 
PIVOT (
  CAST(AVG(delay) AS DECIMAL(4, 2)) AS AvgDelay, MAX(delay) AS MaxDelay
  FOR month IN (1 JAN, 2 FEB)
)
ORDER BY destination
 
+-----------+------------+------------+------------+------------+
|destination|JAN_AvgDelay|JAN_MaxDelay|FEB_AvgDelay|FEB_MaxDelay|
+-----------+------------+------------+------------+------------+
|        ABQ|       19.86|         316|       11.42|          69|
|        ANC|        4.44|         149|        7.90|         141|
|        ATL|       11.98|         397|        7.73|         145|
|        AUS|        3.48|          50|       -0.21|          18|
|        BOS|        7.84|         110|       14.58|         152|
|        BUR|       -2.03|          56|       -1.89|          78|
|        CLE|       16.00|          27|        null|        null|
|        CLT|        2.53|          41|       12.96|         228|
|        COS|        5.32|          82|       12.18|         203|
|        CVG|       -0.50|           4|        null|        null|
|        DCA|       -1.15|          50|        0.07|          34|
|        DEN|       13.13|         425|       12.95|         625|
|        DFW|        7.95|         247|       12.57|         356|
|        DTW|        9.18|         107|        3.47|          77|
|        EWR|        9.63|         236|        5.20|         212|
|        FAI|        1.84|         160|        4.21|          60|
|        FAT|        1.36|         119|        5.22|         232|
|        FLL|        2.94|          54|        3.50|          40|
|        GEG|        2.28|          63|        2.87|          60|
|        HDN|       -0.44|          27|       -6.50|           0|
+-----------+------------+------------+------------+------------+
only showing top 20 rows