Apache Spark 3.0 SQL DataFrame和DataSet指南

「爱情、让人受尽委屈。」 2022-12-04 06:25 125阅读 0赞

**目录**

简介

SQL

数据集和数据框

入门

起点：SparkSession

Scala语言

Java语言

Python语言

R语言

创建DataFrame

Scala语言

Java语言

Python语言

R语言

未类型化的数据集操作（也称为DataFrame操作）

Scala语言

Java语言

Python语言

R语言

以编程方式运行SQL查询

Scala语言

Java语言

Python语言

R语言

全局临时视图

Scala语言

Java语言

Python语言

SQL

创建DataSet

Scala语言

Java语言

与RDD互操作

使用反射推断架构

以编程方式指定架构

标量函数

汇总功能

数据源

性能调优

在内存中缓存数据

其他配置选项

加入针对SQL查询的策略提示

Scala语言

Java语言

Python语言

R语言

SQL

SQL查询的合并提示

自适应查询执行

合并后的Shuffle分区

将排序合并联接转换为广播联接

优化偏斜连接

分布式SQL引擎

运行Thrift JDBC / ODBC服务器

运行Spark SQL CLI

PySpark使用Apache Arrow的熊猫使用指南

PySpark中的Apache Arrow

确保已安装PyArrow

启用与熊猫之间的转换

熊猫UDF（又名矢量化UDF）

系列到系列

系列迭代器到系列迭代器

多个系列的迭代器到系列的迭代器

系列到标量

熊猫函数API

分组Map

Map操作

共同Map

使用说明

支持的SQL类型

设置箭头批处理大小

带时区语义的时间戳

推荐的Pandas和PyArrow版本

PyArrow> = 0.15.0和Spark 2.3.x，2.4.x的兼容性设置

迁移指南

SQL参考

--------------------

#  简介  #

Spark SQL是用于结构化数据处理的Spark模块。与基本的Spark RDD API不同，Spark SQL提供的接口为Spark提供了有关数据结构和正在执行的计算的更多信息。在内部，Spark SQL使用这些额外的信息来执行额外的优化。与Spark SQL交互的方法有几种，包括SQL和Dataset API。计算结果时，将使用相同的执行引擎，而与要用来表达计算的API /语言无关。这种统一意味着开发人员可以轻松地在不同的API之间来回切换，从而提供最自然的方式来表达给定的转换。

此页面上的所有示例均使用Spark发行版中包含的示例数据，并且可以在`spark-shell`，`pyspark`shell或`sparkR`shell中运行。

## **SQL** ##

Spark SQL的一种用途是执行SQL查询。Spark SQL还可以用于从现有的Hive安装中读取数据。有关如何配置此功能的更多信息，请参考[Hive Tables][]部分。从另一种编程语言运行SQL时，结果将作为[Dataset / DataFrame][Dataset _ DataFrame]返回。您还可以使用[命令行][Link 1] 或通过[JDBC / ODBC][JDBC _ ODBC]与SQL接口进行交互。

## **数据集和数据框** ##

数据集是数据的分布式集合。数据集是Spark 1.6中添加的新接口，它具有RDD的优点（强类型输入，使用强大的Lambda函数的能力）以及Spark SQL的优化执行引擎的优点。数据集可以被[构造][Link 2]从JVM对象，然后使用功能性的转换（操作`map`，`flatMap`，`filter`等等）。Dataset API在[Scala][]和 [Java中][Java]可用。Python不支持Dataset API。但是由于Python的动态特性，Dataset API的许多优点已经可用（即，您可以自然地通过名称访问行的字段 `row.columnName`）。R的情况类似。

DataFrame是组织为命名列的*数据集*。从概念上讲，它等效于关系数据库中的表或R / Python中的数据框，但是在后台进行了更丰富的优化。可以从多种[资源][Link 3]构造DataFrame，例如：结构化数据文件，Hive中的表，外部数据库或现有的RDD。DataFrame API在Scala，Java，[Python][]和[R中][R]可用。在Scala和Java中，DataFrame由的数据集表示`Row`。在[Scala API中][Scala]，`DataFrame`只是类型别名`Dataset[Row]`。而在[Java API中][Java]，用户需要使用`Dataset<Row>`来代表`DataFrame`。

在整个文档中，我们通常将的Scala / Java数据集`Row`称为DataFrames。

# 入门 #

## **起点：SparkSession** ##

### **Scala语言** ###

[`SparkSession`][SparkSession]类是Spark中所有功能的入口点。要创建一个基本的`SparkSession`，只需使用`SparkSession.builder()`：

import org.apache.spark.sql.SparkSession
    
    val spark = SparkSession
      .builder()
      .appName("Spark SQL basic example")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()
    
    // For implicit conversions like converting RDDs to DataFrames
    import spark.implicits._

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

`SparkSession`Spark 2.0中的内置支持Hive功能，包括使用HiveQL编写查询，访问Hive UDF以及从Hive表读取数据的功能。要使用这些功能，您不需要现有的Hive设置。

### **Java语言** ###

[`SparkSession`][SparkSession 1]类是Spark中所有功能的入口点。要创建一个基本的`SparkSession`，只需使用`SparkSession.builder()`：

import org.apache.spark.sql.SparkSession;
    
    SparkSession spark = SparkSession
      .builder()
      .appName("Java Spark SQL basic example")
      .config("spark.some.config.option", "some-value")
      .getOrCreate();

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

`SparkSession`Spark 2.0中的内置支持Hive功能，包括使用HiveQL编写查询，访问Hive UDF以及从Hive表读取数据的功能。要使用这些功能，您不需要现有的Hive设置。

### **Python语言** ###

[`SparkSession`][SparkSession 2]类是Spark中所有功能的入口点。要创建一个基本的`SparkSession`，只需使用`SparkSession.builder`：

from pyspark.sql import SparkSession
    
    spark = SparkSession \
        .builder \
        .appName("Python Spark SQL basic example") \
        .config("spark.some.config.option", "some-value") \
        .getOrCreate()

在Spark存储库中的“ examples / src / main / python / sql / basic.py”中找到完整的示例代码。

`SparkSession`Spark 2.0中的内置支持Hive功能，包括使用HiveQL编写查询，访问Hive UDF以及从Hive表读取数据的功能。要使用这些功能，您不需要现有的Hive设置。

### **R语言** ###

[`SparkSession`][SparkSession 3]类是Spark中所有功能的入口点。要初始化一个基本的`SparkSession`，只需调用`sparkR.session()`：

sparkR.session(appName = "R Spark SQL basic example", sparkConfig = list(spark.some.config.option = "some-value"))

在Spark存储库中的“ examples / src / main / r / RSparkSQLExample.R”中找到完整的示例代码。

请注意，首次调用时，`sparkR.session()`将初始化全局`SparkSession`单例实例，并始终为连续调用返回对此实例的引用。这样，用户只需初始化`SparkSession`一次，然后SparkR之类的函数`read.df`就可以隐式访问此全局实例，并且用户无需传递该`SparkSession`实例。

`SparkSession`Spark 2.0中的内置支持Hive功能，包括使用HiveQL编写查询，访问Hive UDF以及从Hive表读取数据的功能。要使用这些功能，您不需要现有的Hive设置。

## **创建DataFrame** ##

### **Scala语言** ###

使用`SparkSession`，应用程序可以从[现有的`RDD`][RDD]，Hive表的或[Spark数据源][Link 3]创建DataFrame 。

例如，以下内容基于JSON文件的内容创建一个DataFrame：

val df = spark.read.json("examples/src/main/resources/people.json")
    
    // Displays the content of the DataFrame to stdout
    df.show()
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

### **Java语言** ###

使用`SparkSession`，应用程序可以从[现有的`RDD`][RDD]，Hive表的或[Spark数据源][Link 3]创建DataFrame 。

例如，以下内容基于JSON文件的内容创建一个DataFrame：

import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    Dataset<Row> df = spark.read().json("examples/src/main/resources/people.json");
    
    // Displays the content of the DataFrame to stdout
    df.show();
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

### **Python语言** ###

使用`SparkSession`，应用程序可以从[现有的`RDD`][RDD]，Hive表的或[Spark数据源][Link 3]创建DataFrame 。

例如，以下内容基于JSON文件的内容创建一个DataFrame：

# spark is an existing SparkSession
    df = spark.read.json("examples/src/main/resources/people.json")
    # Displays the content of the DataFrame to stdout
    df.show()
    # +----+-------+
    # | age|   name|
    # +----+-------+
    # |null|Michael|
    # |  30|   Andy|
    # |  19| Justin|
    # +----+-------+

在Spark存储库中的“ examples / src / main / python / sql / basic.py”中找到完整的示例代码。

### **R语言** ###

使用`SparkSession`，应用程序可以从本地R data.frame，Hive表或[Spark数据源][Link 3]创建DataFrame 。

例如，以下内容基于JSON文件的内容创建一个DataFrame：

df <- read.json("examples/src/main/resources/people.json")
    
    # Displays the content of the DataFrame
    head(df)
    ##   age    name
    ## 1  NA Michael
    ## 2  30    Andy
    ## 3  19  Justin
    
    # Another method to print the first few rows and optionally truncate the printing of long values
    showDF(df)
    ## +----+-------+
    ## | age|   name|
    ## +----+-------+
    ## |null|Michael|
    ## |  30|   Andy|
    ## |  19| Justin|
    ## +----+-------+

在Spark存储库中的“ examples / src / main / r / RSparkSQLExample.R”中找到完整的示例代码。

## **未类型化的数据集操作（也称为DataFrame操作）** ##

DataFrames为[Scala][]，[Java][]，[Python][]和[R中的][R 1]结构化数据操作提供了一种特定于域的语言。

如上所述，在Spark 2.0中，DataFrames只是`Row`Scala和Java API中的的数据集。与强类型的Scala / Java数据集附带的“类型转换”相反，这些操作也称为“非类型转换”。

这里我们包括一些使用数据集进行结构化数据处理的基本示例：

### **Scala语言** ###

// This import is needed to use the $-notation
    import spark.implicits._
    // Print the schema in a tree format
    df.printSchema()
    // root
    // |-- age: long (nullable = true)
    // |-- name: string (nullable = true)
    
    // Select only the "name" column
    df.select("name").show()
    // +-------+
    // |   name|
    // +-------+
    // |Michael|
    // |   Andy|
    // | Justin|
    // +-------+
    
    // Select everybody, but increment the age by 1
    df.select($"name", $"age" + 1).show()
    // +-------+---------+
    // |   name|(age + 1)|
    // +-------+---------+
    // |Michael|     null|
    // |   Andy|       31|
    // | Justin|       20|
    // +-------+---------+
    
    // Select people older than 21
    df.filter($"age" > 21).show()
    // +---+----+
    // |age|name|
    // +---+----+
    // | 30|Andy|
    // +---+----+
    
    // Count people by age
    df.groupBy("age").count().show()
    // +----+-----+
    // | age|count|
    // +----+-----+
    // |  19|    1|
    // |null|    1|
    // |  30|    1|
    // +----+-----+

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

有关可对数据集执行的操作类型的完整列表，请参阅[API文档][Scala]。

除了简单的列引用和表达式外，数据集还具有丰富的函数库，包括字符串处理，日期算术，通用数学运算等。完整列表可在[DataFrame Function Reference中获得][DataFrame Function Reference]。

### **Java语言** ###

// col("...") is preferable to df.col("...")
    import static org.apache.spark.sql.functions.col;
    
    // Print the schema in a tree format
    df.printSchema();
    // root
    // |-- age: long (nullable = true)
    // |-- name: string (nullable = true)
    
    // Select only the "name" column
    df.select("name").show();
    // +-------+
    // |   name|
    // +-------+
    // |Michael|
    // |   Andy|
    // | Justin|
    // +-------+
    
    // Select everybody, but increment the age by 1
    df.select(col("name"), col("age").plus(1)).show();
    // +-------+---------+
    // |   name|(age + 1)|
    // +-------+---------+
    // |Michael|     null|
    // |   Andy|       31|
    // | Justin|       20|
    // +-------+---------+
    
    // Select people older than 21
    df.filter(col("age").gt(21)).show();
    // +---+----+
    // |age|name|
    // +---+----+
    // | 30|Andy|
    // +---+----+
    
    // Count people by age
    df.groupBy("age").count().show();
    // +----+-----+
    // | age|count|
    // +----+-----+
    // |  19|    1|
    // |null|    1|
    // |  30|    1|
    // +----+-----+

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

有关可对数据集执行的操作类型的完整列表，请参阅[API文档][API]。

除了简单的列引用和表达式外，数据集还具有丰富的函数库，包括字符串处理，日期算术，通用数学运算等。完整列表可在[DataFrame Function Reference中获得][DataFrame Function Reference 1]。

### **Python语言** ###

在Python中，可以通过属性（`df.age`）或通过索引（`df['age']`）访问DataFrame的列。尽管前者便于交互式数据探索，但强烈建议用户使用后者形式，这是将来的证明，并且不会与列名保持一致，列名也是DataFrame类的属性。

# spark, df are from the previous example
    # Print the schema in a tree format
    df.printSchema()
    # root
    # |-- age: long (nullable = true)
    # |-- name: string (nullable = true)
    
    # Select only the "name" column
    df.select("name").show()
    # +-------+
    # |   name|
    # +-------+
    # |Michael|
    # |   Andy|
    # | Justin|
    # +-------+
    
    # Select everybody, but increment the age by 1
    df.select(df['name'], df['age'] + 1).show()
    # +-------+---------+
    # |   name|(age + 1)|
    # +-------+---------+
    # |Michael|     null|
    # |   Andy|       31|
    # | Justin|       20|
    # +-------+---------+
    
    # Select people older than 21
    df.filter(df['age'] > 21).show()
    # +---+----+
    # |age|name|
    # +---+----+
    # | 30|Andy|
    # +---+----+
    
    # Count people by age
    df.groupBy("age").count().show()
    # +----+-----+
    # | age|count|
    # +----+-----+
    # |  19|    1|
    # |null|    1|
    # |  30|    1|
    # +----+-----+

在Spark存储库中的“ examples / src / main / python / sql / basic.py”中找到完整的示例代码。

有关可在DataFrame上执行的操作类型的完整列表，请参阅[API文档][Python]。

除了简单的列引用和表达式外，DataFrames还具有丰富的函数库，包括字符串处理，日期算术，通用数学运算等。完整列表可在[DataFrame Function Reference中获得][DataFrame Function Reference 2]。

### **R语言** ###

# Create the DataFrame
    df <- read.json("examples/src/main/resources/people.json")
    
    # Show the content of the DataFrame
    head(df)
    ##   age    name
    ## 1  NA Michael
    ## 2  30    Andy
    ## 3  19  Justin
    
    
    # Print the schema in a tree format
    printSchema(df)
    ## root
    ## |-- age: long (nullable = true)
    ## |-- name: string (nullable = true)
    
    # Select only the "name" column
    head(select(df, "name"))
    ##      name
    ## 1 Michael
    ## 2    Andy
    ## 3  Justin
    
    # Select everybody, but increment the age by 1
    head(select(df, df$name, df$age + 1))
    ##      name (age + 1.0)
    ## 1 Michael          NA
    ## 2    Andy          31
    ## 3  Justin          20
    
    # Select people older than 21
    head(where(df, df$age > 21))
    ##   age name
    ## 1  30 Andy
    
    # Count people by age
    head(count(groupBy(df, "age")))
    ##   age count
    ## 1  19     1
    ## 2  NA     1
    ## 3  30     1

在Spark存储库中的“ examples / src / main / r / RSparkSQLExample.R”中找到完整的示例代码。

有关可在DataFrame上执行的操作类型的完整列表，请参阅[API文档][R]。

除了简单的列引用和表达式外，DataFrames还具有丰富的函数库，包括字符串处理，日期算术，通用数学运算等。完整列表可在[DataFrame Function Reference中获得][R 1]。

## **以编程方式运行SQL查询** ##

### **Scala语言** ###

上的`sql`函数`SparkSession`使应用程序能够以编程方式运行SQL查询，并以形式返回结果`DataFrame`。

// Register the DataFrame as a SQL temporary view
    df.createOrReplaceTempView("people")
    
    val sqlDF = spark.sql("SELECT * FROM people")
    sqlDF.show()
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

### **Java语言** ###

上的`sql`函数`SparkSession`使应用程序能够以编程方式运行SQL查询，并以形式返回结果`Dataset<Row>`。

import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    // Register the DataFrame as a SQL temporary view
    df.createOrReplaceTempView("people");
    
    Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
    sqlDF.show();
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

### **Python语言** ###

上的`sql`函数`SparkSession`使应用程序能够以编程方式运行SQL查询，并以形式返回结果`DataFrame`。

# Register the DataFrame as a SQL temporary view
    df.createOrReplaceTempView("people")
    
    sqlDF = spark.sql("SELECT * FROM people")
    sqlDF.show()
    # +----+-------+
    # | age|   name|
    # +----+-------+
    # |null|Michael|
    # |  30|   Andy|
    # |  19| Justin|
    # +----+-------+

在Spark存储库中的“ examples / src / main / python / sql / basic.py”中找到完整的示例代码。

### **R语言** ###

该`sql`函数使应用程序能够以编程方式运行SQL查询，并以形式返回结果`SparkDataFrame`。

df <- sql("SELECT * FROM table")

在Spark存储库中的“ examples / src / main / r / RSparkSQLExample.R”中找到完整的示例代码。

## **全局临时视图** ##

Spark SQL中的临时视图是会话作用域的，如果创建它的会话终止，它将消失。如果要在所有会话之间共享一个临时视图并保持活动状态，直到Spark应用程序终止，则可以创建全局临时视图。全局临时视图与系统保留的数据库相关联`global_temp`，我们必须使用限定名称来引用它，例如`SELECT * FROM global_temp.view1`。

### **Scala语言** ###

// Register the DataFrame as a global temporary view
    df.createGlobalTempView("people")
    
    // Global temporary view is tied to a system preserved database `global_temp`
    spark.sql("SELECT * FROM global_temp.people").show()
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+
    
    // Global temporary view is cross-session
    spark.newSession().sql("SELECT * FROM global_temp.people").show()
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

### **Java语言** ###

// Register the DataFrame as a global temporary view
    df.createGlobalTempView("people");
    
    // Global temporary view is tied to a system preserved database `global_temp`
    spark.sql("SELECT * FROM global_temp.people").show();
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+
    
    // Global temporary view is cross-session
    spark.newSession().sql("SELECT * FROM global_temp.people").show();
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

### **Python语言** ###

# Register the DataFrame as a global temporary view
    df.createGlobalTempView("people")
    
    # Global temporary view is tied to a system preserved database `global_temp`
    spark.sql("SELECT * FROM global_temp.people").show()
    # +----+-------+
    # | age|   name|
    # +----+-------+
    # |null|Michael|
    # |  30|   Andy|
    # |  19| Justin|
    # +----+-------+
    
    # Global temporary view is cross-session
    spark.newSession().sql("SELECT * FROM global_temp.people").show()
    # +----+-------+
    # | age|   name|
    # +----+-------+
    # |null|Michael|
    # |  30|   Andy|
    # |  19| Justin|
    # +----+-------+

在Spark存储库中的“ examples / src / main / python / sql / basic.py”中找到完整的示例代码。

### **SQL** ###

CREATE GLOBAL TEMPORARY VIEW temp_view AS SELECT a + 1, b * 2 FROM tbl
    
    SELECT * FROM global_temp.temp_view

## **创建DataSet** ##

数据集与RDD相似，但是它们不是使用Java序列化或Kryo，而是使用专用的[Encoder][]对对象进行序列化以进行网络处理或传输。虽然编码器和标准序列化都负责将对象转换为字节，但是编码器是动态生成的代码，并使用一种格式，该格式允许Spark执行许多操作，如过滤，排序和哈希处理，而无需将字节反序列化为对象。

### **Scala语言** ###

case class Person(name: String, age: Long)
    
    // Encoders are created for case classes
    val caseClassDS = Seq(Person("Andy", 32)).toDS()
    caseClassDS.show()
    // +----+---+
    // |name|age|
    // +----+---+
    // |Andy| 32|
    // +----+---+
    
    // Encoders for most common types are automatically provided by importing spark.implicits._
    val primitiveDS = Seq(1, 2, 3).toDS()
    primitiveDS.map(_ + 1).collect() // Returns: Array(2, 3, 4)
    
    // DataFrames can be converted to a Dataset by providing a class. Mapping will be done by name
    val path = "examples/src/main/resources/people.json"
    val peopleDS = spark.read.json(path).as[Person]
    peopleDS.show()
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

### **Java语言** ###

import java.util.Arrays;
    import java.util.Collections;
    import java.io.Serializable;
    
    import org.apache.spark.api.java.function.MapFunction;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.Encoder;
    import org.apache.spark.sql.Encoders;
    
    public static class Person implements Serializable {
      private String name;
      private int age;
    
      public String getName() {
        return name;
      }
    
      public void setName(String name) {
        this.name = name;
      }
    
      public int getAge() {
        return age;
      }
    
      public void setAge(int age) {
        this.age = age;
      }
    }
    
    // Create an instance of a Bean class
    Person person = new Person();
    person.setName("Andy");
    person.setAge(32);
    
    // Encoders are created for Java beans
    Encoder<Person> personEncoder = Encoders.bean(Person.class);
    Dataset<Person> javaBeanDS = spark.createDataset(
      Collections.singletonList(person),
      personEncoder
    );
    javaBeanDS.show();
    // +---+----+
    // |age|name|
    // +---+----+
    // | 32|Andy|
    // +---+----+
    
    // Encoders for most common types are provided in class Encoders
    Encoder<Integer> integerEncoder = Encoders.INT();
    Dataset<Integer> primitiveDS = spark.createDataset(Arrays.asList(1, 2, 3), integerEncoder);
    Dataset<Integer> transformedDS = primitiveDS.map(
        (MapFunction<Integer, Integer>) value -> value + 1,
        integerEncoder);
    transformedDS.collect(); // Returns [2, 3, 4]
    
    // DataFrames can be converted to a Dataset by providing a class. Mapping based on name
    String path = "examples/src/main/resources/people.json";
    Dataset<Person> peopleDS = spark.read().json(path).as(personEncoder);
    peopleDS.show();
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

## **与RDD互操作** ##

Spark SQL支持两种将现有RDD转换为数据集的方法。第一种方法使用反射来推断包含特定对象类型的RDD的架构。这种基于反射的方法可以使代码更简洁，当您在编写Spark应用程序时已经了解架构时，可以很好地工作。

创建数据集的第二种方法是通过编程界面，该界面允许您构造模式，然后将其应用于现有的RDD。尽管此方法较为冗长，但可以在运行时才知道列及其类型的情况下构造数据集。

### **使用反射推断架构** ###

**Scala语言**

Spark SQL的Scala接口支持自动将包含案例类的RDD转换为DataFrame。案例类定义表的架构。案例类的参数名称使用反射读取，并成为列的名称。Case类也可以嵌套或包含`Seq`s或`Array`s之类的复杂类型。可以将该RDD隐式转换为DataFrame，然后将其注册为表。表可以在后续的SQL语句中使用。

// For implicit conversions from RDDs to DataFrames
    import spark.implicits._
    
    // Create an RDD of Person objects from a text file, convert it to a Dataframe
    val peopleDF = spark.sparkContext
      .textFile("examples/src/main/resources/people.txt")
      .map(_.split(","))
      .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
      .toDF()
    // Register the DataFrame as a temporary view
    peopleDF.createOrReplaceTempView("people")
    
    // SQL statements can be run by using the sql methods provided by Spark
    val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")
    
    // The columns of a row in the result can be accessed by field index
    teenagersDF.map(teenager => "Name: " + teenager(0)).show()
    // +------------+
    // |       value|
    // +------------+
    // |Name: Justin|
    // +------------+
    
    // or by field name
    teenagersDF.map(teenager => "Name: " + teenager.getAs[String]("name")).show()
    // +------------+
    // |       value|
    // +------------+
    // |Name: Justin|
    // +------------+
    
    // No pre-defined encoders for Dataset[Map[K,V]], define explicitly
    implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]]
    // Primitive types and case classes can be also defined as
    // implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()
    
    // row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
    teenagersDF.map(teenager => teenager.getValuesMap[Any](List("name", "age"))).collect()
    // Array(Map("name" -> "Justin", "age" -> 19))

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

**Java语言**

Spark SQL支持将[JavaBean][]的RDD自动转换 为DataFrame。的`BeanInfo`，使用反射得到，定义了表的模式。当前，Spark SQL不支持包含`Map`字段的JavaBean 。 不过，支持嵌套JavaBean和`List`或`Array`字段。您可以通过创建一个实现Serializable且其所有字段具有getter和setter的类来创建JavaBean。

import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.MapFunction;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.Encoder;
    import org.apache.spark.sql.Encoders;
    
    // Create an RDD of Person objects from a text file
    JavaRDD<Person> peopleRDD = spark.read()
      .textFile("examples/src/main/resources/people.txt")
      .javaRDD()
      .map(line -> {
        String[] parts = line.split(",");
        Person person = new Person();
        person.setName(parts[0]);
        person.setAge(Integer.parseInt(parts[1].trim()));
        return person;
      });
    
    // Apply a schema to an RDD of JavaBeans to get a DataFrame
    Dataset<Row> peopleDF = spark.createDataFrame(peopleRDD, Person.class);
    // Register the DataFrame as a temporary view
    peopleDF.createOrReplaceTempView("people");
    
    // SQL statements can be run by using the sql methods provided by spark
    Dataset<Row> teenagersDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19");
    
    // The columns of a row in the result can be accessed by field index
    Encoder<String> stringEncoder = Encoders.STRING();
    Dataset<String> teenagerNamesByIndexDF = teenagersDF.map(
        (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
        stringEncoder);
    teenagerNamesByIndexDF.show();
    // +------------+
    // |       value|
    // +------------+
    // |Name: Justin|
    // +------------+
    
    // or by field name
    Dataset<String> teenagerNamesByFieldDF = teenagersDF.map(
        (MapFunction<Row, String>) row -> "Name: " + row.<String>getAs("name"),
        stringEncoder);
    teenagerNamesByFieldDF.show();
    // +------------+
    // |       value|
    // +------------+
    // |Name: Justin|
    // +------------+

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

**Python语言**

Spark SQL可以将Row对象的RDD转换为DataFrame，从而推断数据类型。通过将键/值对列表作为kwargs传递给Row类来构造行。该列表的键定义表的列名，并且通过对整个数据集进行采样来推断类型，类似于对JSON文件执行的推断。

from pyspark.sql import Row
    
    sc = spark.sparkContext
    
    # Load a text file and convert each line to a Row.
    lines = sc.textFile("examples/src/main/resources/people.txt")
    parts = lines.map(lambda l: l.split(","))
    people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))
    
    # Infer the schema, and register the DataFrame as a table.
    schemaPeople = spark.createDataFrame(people)
    schemaPeople.createOrReplaceTempView("people")
    
    # SQL can be run over DataFrames that have been registered as a table.
    teenagers = spark.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
    
    # The results of SQL queries are Dataframe objects.
    # rdd returns the content as an :class:`pyspark.RDD` of :class:`Row`.
    teenNames = teenagers.rdd.map(lambda p: "Name: " + p.name).collect()
    for name in teenNames:
        print(name)
    # Name: Justin

在Spark存储库中的“ examples / src / main / python / sql / basic.py”中找到完整的示例代码。

### **以编程方式指定架构** ###

**Scala语言**

如果无法提前定义案例类（例如，记录的结构编码为字符串，或者将解析文本数据集，并且针对不同的用户对字段进行不同的投影），`DataFrame`则可以通过三个步骤以编程方式创建a 。

1.  `Row`从原始RDD 创建一个的RDD；
2.  在步骤1中创建的RDD中，创建`StructType`与`Row`s 的结构匹配 的模式。
3.  `Row`通过`createDataFrame`提供的方法将架构应用于的RDD `SparkSession`。

例如：

import org.apache.spark.sql.Row
    
    import org.apache.spark.sql.types._
    
    // Create an RDD
    val peopleRDD = spark.sparkContext.textFile("examples/src/main/resources/people.txt")
    
    // The schema is encoded in a string
    val schemaString = "name age"
    
    // Generate the schema based on the string of schema
    val fields = schemaString.split(" ")
      .map(fieldName => StructField(fieldName, StringType, nullable = true))
    val schema = StructType(fields)
    
    // Convert records of the RDD (people) to Rows
    val rowRDD = peopleRDD
      .map(_.split(","))
      .map(attributes => Row(attributes(0), attributes(1).trim))
    
    // Apply the schema to the RDD
    val peopleDF = spark.createDataFrame(rowRDD, schema)
    
    // Creates a temporary view using the DataFrame
    peopleDF.createOrReplaceTempView("people")
    
    // SQL can be run over a temporary view created using DataFrames
    val results = spark.sql("SELECT name FROM people")
    
    // The results of SQL queries are DataFrames and support all the normal RDD operations
    // The columns of a row in the result can be accessed by field index or by field name
    results.map(attributes => "Name: " + attributes(0)).show()
    // +-------------+
    // |        value|
    // +-------------+
    // |Name: Michael|
    // |   Name: Andy|
    // | Name: Justin|
    // +-------------+

在Spark存储库中的“ examples / src / main / scala / org / apache / spark / examples / sql / SparkSQLExample.scala”中找到完整的示例代码。

**Java语言**

当无法提前定义JavaBean类时（例如，记录的结构编码为字符串，或者将解析文本数据集，并且为不同的用户设计不同的字段），`Dataset<Row>`可以通过三个步骤以编程方式创建a 。

例如：

import java.util.ArrayList;
    import java.util.List;
    
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.function.Function;
    
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    import org.apache.spark.sql.types.DataTypes;
    import org.apache.spark.sql.types.StructField;
    import org.apache.spark.sql.types.StructType;
    
    // Create an RDD
    JavaRDD<String> peopleRDD = spark.sparkContext()
      .textFile("examples/src/main/resources/people.txt", 1)
      .toJavaRDD();
    
    // The schema is encoded in a string
    String schemaString = "name age";
    
    // Generate the schema based on the string of schema
    List<StructField> fields = new ArrayList<>();
    for (String fieldName : schemaString.split(" ")) {
      StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);
      fields.add(field);
    }
    StructType schema = DataTypes.createStructType(fields);
    
    // Convert records of the RDD (people) to Rows
    JavaRDD<Row> rowRDD = peopleRDD.map((Function<String, Row>) record -> {
      String[] attributes = record.split(",");
      return RowFactory.create(attributes[0], attributes[1].trim());
    });
    
    // Apply the schema to the RDD
    Dataset<Row> peopleDataFrame = spark.createDataFrame(rowRDD, schema);
    
    // Creates a temporary view using the DataFrame
    peopleDataFrame.createOrReplaceTempView("people");
    
    // SQL can be run over a temporary view created using DataFrames
    Dataset<Row> results = spark.sql("SELECT name FROM people");
    
    // The results of SQL queries are DataFrames and support all the normal RDD operations
    // The columns of a row in the result can be accessed by field index or by field name
    Dataset<String> namesDS = results.map(
        (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
        Encoders.STRING());
    namesDS.show();
    // +-------------+
    // |        value|
    // +-------------+
    // |Name: Michael|
    // |   Name: Andy|
    // | Name: Justin|
    // +-------------+

在Spark存储库中的“ examples / src / main / java / org / apache / spark / examples / sql / JavaSparkSQLExample.java”中找到完整的示例代码。

**Python语言**

如果无法提前定义kwarg的字典（例如，记录的结构编码为字符串，或者将解析文本数据集，并且为不同的用户投影不同的字段），则`DataFrame`可以通过编程方式创建三个脚步。

1.  从原始RDD创建元组或列表的RDD；
2.  `StructType`在步骤1中创建的RDD中创建由元组或列表结构的匹配表示的模式。
3.  通过`createDataFrame`提供的方法将架构应用于RDD `SparkSession`。

例如：

# Import data types
    from pyspark.sql.types import *
    
    sc = spark.sparkContext
    
    # Load a text file and convert each line to a Row.
    lines = sc.textFile("examples/src/main/resources/people.txt")
    parts = lines.map(lambda l: l.split(","))
    # Each line is converted to a tuple.
    people = parts.map(lambda p: (p[0], p[1].strip()))
    
    # The schema is encoded in a string.
    schemaString = "name age"
    
    fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
    schema = StructType(fields)
    
    # Apply the schema to the RDD.
    schemaPeople = spark.createDataFrame(people, schema)
    
    # Creates a temporary view using the DataFrame
    schemaPeople.createOrReplaceTempView("people")
    
    # SQL can be run over DataFrames that have been registered as a table.
    results = spark.sql("SELECT name FROM people")
    
    results.show()
    # +-------+
    # |   name|
    # +-------+
    # |Michael|
    # |   Andy|
    # | Justin|
    # +-------+

在Spark存储库中的“ examples / src / main / python / sql / basic.py”中找到完整的示例代码。

## **标量函数** ##

标量函数是每行返回一个值的函数，聚合函数则返回一组行的值，而标量函数则是每行返回一个值。Spark SQL支持多种[内置标量函数][Link 4]。它还支持[用户定义的标量函数][Link 5]。

## **汇总功能** ##

聚合函数是在一组行上返回单个值的函数。该[内置聚合函数][Link 6]提供通用聚合如`count()`，`countDistinct()`，`avg()`，`max()`，`min()`，等用户不限于预定义的聚集功能，可以创建自己的。有关用户定义的聚合函数的更多详细信息，请参阅[用户定义的聚合函数][Link 7]的文档 。

# **数据源** #

Spark SQL支持通过DataFrame接口对各种数据源进行操作。DataFrame可以使用关系转换进行操作，也可以用于创建临时视图。将DataFrame注册为临时视图使您可以对其数据运行SQL查询。本节介绍了使用Spark数据源加载和保存数据的一般方法，然后介绍了可用于内置数据源的特定选项。

*  [通用加载/保存功能][Link 8]
    
     *  [手动指定选项][Link 9]
     *  [直接在文件上运行SQL][SQL]
     *  [保存模式][Link 10]
     *  [保存到永久表][Link 11]
     *  [桶，分类和分区][Link 12]
 *  [通用文件源选项][Link 13]
    
     *  [忽略损坏的文件][Link 14]
     *  [忽略丢失的文件][Link 15]
     *  [路径全局过滤器][Link 16]
     *  [递归文件查找][Link 17]
 *  [实木复合地板文件][Link 18]
    
     *  [以编程方式加载数据][Link 19]
     *  [分区发现][Link 20]
     *  [模式合并][Link 21]
     *  [Hive Metastore Parquet表转换][Hive Metastore Parquet]
     *  [组态][Link 22]
 *  [ORC文件][ORC]
 *  [JSON文件][JSON]
 *  [蜂巢表][Hive Tables]
    
     *  [指定Hive表的存储格式][Hive]
     *  [与Hive Metastore的不同版本进行交互][Hive Metastore]
 *  [JDBC到其他数据库][JDBC]
 *  [Avro文件][Avro]
    
     *  [部署中][Link 23]
     *  [加载和保存功能][Link 24]
     *  [to\_avro（）和from\_avro（）][to_avro_from_avro]
     *  [数据源选项][Link 25]
     *  [组态][Link 26]
     *  [与Databricks spark-avro的兼容性][Databricks spark-avro]
     *  [Avro-> Spark SQL转换支持的类型][Avro-_ Spark SQL]
     *  [Spark SQL支持的类型-> Avro转换][Spark SQL_-_ Avro]
 *  [整个二进制文件][Link 27]
 *  [故障排除][Link 28]

# **性能调优** #

对于某些工作负载，可以通过在内存中缓存数据或打开某些实验选项来提高性能。

## **在内存中缓存数据** ##

Spark SQL可以通过调用`spark.catalog.cacheTable("tableName")`或使用内存列式格式缓存表`dataFrame.cache()`。然后，Spark SQL将仅扫描必需的列，并将自动调整压缩以最小化内存使用和GC压力。您可以调用`spark.catalog.uncacheTable("tableName")`从内存中删除表。

可以使用`setConf`on方法`SparkSession`或`SET key=value`使用SQL 运行 命令来完成内存中缓存的配置。

## **其他配置选项** ##

以下选项也可以用于调整查询执行的性能。随着自动执行更多优化，将来的发行版中可能会弃用这些选项。

<table> 
 <tbody> 
  <tr> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">物业名称</th> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">默认</th> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">含义</th> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">自版本</th> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.files.maxPartitionBytes</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">134217728（128 MB）</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">读取文件时打包到单个分区中的最大字节数。仅当使用基于文件的源（例如Parquet，JSON和ORC）时，此配置才有效。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">2.0.0</td> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.files.openCostInBytes</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">4194304（4 MB）</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">可以同时扫描以字节数衡量的打开文件的估计成本。将多个文件放入分区时使用。最好高估一下，然后具有较小文件的分区将比具有较大文件的分区（首先安排）更快。仅当使用基于文件的源（例如Parquet，JSON和ORC）时，此配置才有效。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">2.0.0</td> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.broadcastTimeout</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">300</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"> <p style="margin-left:0px;">广播加入中广播等待时间的秒数超时</p> </td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">1.3.0</td> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.autoBroadcastJoinThreshold</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">10485760（10 MB）</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">配置表的最大大小（以字节为单位），该表在执行联接时将广播到所有工作程序节点。通过将此值设置为-1，可以禁用广播。请注意，当前仅<code>ANALYZE TABLE &lt;tableName&gt; COMPUTE STATISTICS noscan</code>运行命令的Hive Metastore表支持统计信息&nbsp;。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">1.1.0</td> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.shuffle.partitions</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">200</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">配置在对联接或聚合进行数据混排时要使用的分区数。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">1.1.0</td> 
  </tr> 
 </tbody> 
</table>

## **加入针对SQL查询的策略提示** ##

连接策略提示，即`BROADCAST`，`MERGE`，`SHUFFLE_HASH`和`SHUFFLE_REPLICATE_NL`，指导星火与其他关系结合时，他们使用暗示策略上的每个特定关系。例如，当`BROADCAST`在表't1'上使用提示时，Spark将优先考虑以't1'作为生成端的广播连接（广播哈希连接或广播嵌套循环连接，取决于是否有任何等连接键）。即使统计信息建议的表't1'的大小在配置之上`spark.sql.autoBroadcastJoinThreshold`。

当在连接的两侧指定了不同的连接策略提示时，Spark会优先于`BROADCAST`提示而不是`MERGE`提示优先 `SHUFFLE_HASH`于`SHUFFLE_REPLICATE_NL` 提示。当使用`BROADCAST`提示或`SHUFFLE_HASH`提示指定双方时，Spark将根据联接类型和关系的大小选择构建方。

请注意，由于特定策略可能不支持所有联接类型，因此不能保证Spark将选择提示中指定的联接策略。

### **Scala语言** ###

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show()

有关更多详细信息，请参阅[Join Hints][]的文档。

### **Java语言** ###

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show();

有关更多详细信息，请参阅[Join Hints][]的文档。

### **Python语言** ###

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show()

有关更多详细信息，请参阅[Join Hints][]的文档。

### **R语言** ###

src <- sql("SELECT * FROM src")
    records <- sql("SELECT * FROM records")
    head(join(src, hint(records, "broadcast"), src$key == records$key))

有关更多详细信息，请参阅[Join Hints][]的文档。

### SQL ###

-- We accept BROADCAST, BROADCASTJOIN and MAPJOIN for broadcast hint
    SELECT /*+ BROADCAST(r) */ * FROM records r JOIN src s ON r.key = s.key

有关更多详细信息，请参阅[Join Hints][]的文档。

## **SQL查询的合并提示** ##

Coalesce提示使Spark SQL用户可以像一样控制输出文件的数量 `coalesce`，`repartition`并且`repartitionByRange`在Dataset API中，它们可以用于性能调整和减少输出文件的数量。“ COALESCE”提示仅具有分区号作为参数。“ REPARTITION”提示具有分区号和/或列作为参数。“ REPARTITION\_BY\_RANGE”提示必须具有列名，并且分区号是可选的。

<span style="color:#1d1f22"><span style="color:#333333"><code>SELECT /*+ COALESCE(3) */ * FROM t
    SELECT /*+ REPARTITION(3) */ * FROM t
    SELECT /*+ REPARTITION(c) */ * FROM t
    SELECT /*+ REPARTITION(3, c) */ * FROM t
    SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t
    SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t
    </code></span></span>

有关更多详细信息，请参阅“ [分区提示][Link 29] ”文档。

## **自适应查询执行** ##

自适应查询执行（AQE）是Spark SQL中的一种优化技术，它利用运行时统计信息来选择最有效的查询执行计划。默认情况下禁用AQE。Spark SQL可以使用的伞形配置`spark.sql.adaptive.enabled`来控制是否打开/关闭它。从Spark 3.0开始，AQE具有三个主要功能，包括合并后混洗分区，将排序合并联接转换为广播联接以及倾斜联接优化。

### **合并后的Shuffle分区** ###

当`spark.sql.adaptive.enabled`和`spark.sql.adaptive.coalescePartitions.enabled`配置均为true 时，此功能将根据地图输出统计信息合并后混洗分区。此功能简化了运行查询时对混洗分区号的调整。您无需设置适当的随机播放分区号即可适合您的数据集。一旦通过`spark.sql.adaptive.coalescePartitions.initialPartitionNum`配置设置了足够大的初始shuffle分区数量，Spark便可以在运行时选择正确的shuffle分区编号。

<table> 
 <tbody> 
  <tr> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">物业名称</th> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">默认</th> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">含义</th> 
   <th style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">自版本</th> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.adaptive.coalescePartitions.enabled</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">真正</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">如果为true和<code>spark.sql.adaptive.enabled</code>true，Spark将根据目标大小（由指定<code>spark.sql.adaptive.advisoryPartitionSizeInBytes</code>）合并连续的shuffle分区，以避免执行过多的小任务。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">3.0.0</td> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.adaptive.coalescePartitions.minPartitionNum</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">默认并行</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">合并后的最小混洗分区数。如果未设置，则默认值为Spark集群的默认并行度。仅当<code>spark.sql.adaptive.enabled</code>和<code>spark.sql.adaptive.coalescePartitions.enabled</code>都启用时，此配置才有效。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">3.0.0</td> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.adaptive.coalescePartitions.initialPartitionNum</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">200</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">合并之前的洗牌分区的初始数量。默认情况下，它等于<code>spark.sql.shuffle.partitions</code>。仅当<code>spark.sql.adaptive.enabled</code>和<code>spark.sql.adaptive.coalescePartitions.enabled</code>都启用时，此配置才有效。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">3.0.0</td> 
  </tr> 
  <tr> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;"><code>spark.sql.adaptive.advisoryPartitionSizeInBytes</code></td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">64兆字节</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">自适应优化过程中（随机数）随机播放分区的建议大小（以字节为单位<code>spark.sql.adaptive.enabled</code>）。当Spark合并较小的shuffle分区或拆分倾斜的shuffle分区时，此命令才会生效。</td> 
   <td style="border-color:#dddddd #cccccc #cccccc;text-align:left;vertical-align:top;">3.0.0</td> 
  </tr> 
 </tbody> 
</table>

### **将排序合并联接转换为广播联接** ###

当任何联接端的运行时统计信息小于广播哈希联接阈值时，AQE会将排序合并联接转换为广播哈希联接。这不像首先计划广播哈希连接那样有效，但是比继续进行排序合并连接要好，因为我们可以保存两个连接端的排序，并在本地读取随机文件以节省网络流量（如果`spark.sql.adaptive.localShuffleReader.enabled`是真的）

### **优化偏斜连接** ###

数据偏斜会严重降低联接查询的性能。此功能通过将倾斜的任务拆分（按需复制）为大小大致相等的任务来动态处理排序合并联接中的倾斜。同时启用`spark.sql.adaptive.enabled`和`spark.sql.adaptive.skewJoin.enabled`配置时，此选项才生效。

# **分布式SQL引擎** #

*  [运行Thrift JDBC / ODBC服务器][JDBC _ ODBC]
 *  [运行Spark SQL CLI][Link 1]

Spark SQL也可以使用其JDBC / ODBC或命令行界面充当分布式查询引擎。在这种模式下，最终用户或应用程序可以直接与Spark SQL交互以运行SQL查询，而无需编写任何代码。

## **运行Thrift JDBC / ODBC服务器** ##

此处实现的Thrift JDBC / ODBC服务器对应[`HiveServer2`][HiveServer2] 于内置的Hive。您可以使用Spark或兼容的Hive随附的beeline脚本测试JDBC服务器。

要启动JDBC / ODBC服务器，请在Spark目录中运行以下命令：

./sbin/start-thriftserver.sh

该脚本接受所有`bin/spark-submit`命令行选项，以及`--hiveconf`用于指定Hive属性的选项。您可以运行`./sbin/start-thriftserver.sh --help`以获取所有可用选项的完整列表。默认情况下，服务器在localhost：10000上侦听。您可以通过任一环境变量来覆盖此行为，即：

export HIVE_SERVER2_THRIFT_PORT=<listening-port>
    export HIVE_SERVER2_THRIFT_BIND_HOST=<listening-host>
    ./sbin/start-thriftserver.sh \
      --master <master-uri> \
      ...

或系统属性：

./sbin/start-thriftserver.sh \
      --hiveconf hive.server2.thrift.port=<listening-port> \
      --hiveconf hive.server2.thrift.bind.host=<listening-host> \
      --master <master-uri>
      ...

现在，您可以使用beeline测试Thrift JDBC / ODBC服务器：

./bin/beeline

通过以下方式直线连接到JDBC / ODBC服务器：

beeline> !connect jdbc:hive2://localhost:10000

Beeline会要求您提供用户名和密码。在非安全模式下，只需在计算机上输入用户名和空白密码即可。对于安全模式，请遵循[beeline文档中][beeline]给出的 [说明][beeline]。

蜂巢的结构是通过将您做`hive-site.xml`，`core-site.xml`和`hdfs-site.xml`文件`conf/`。

您还可以使用Hive随附的beeline脚本。

Thrift JDBC服务器还支持通过HTTP传输发送Thrift RPC消息。使用以下设置可以将HTTP模式作为系统属性或在`hive-site.xml`文件中启用`conf/`：

hive.server2.transport.mode - Set this to value: http
    hive.server2.thrift.http.port - HTTP port number to listen on; default is 10001
    hive.server2.http.endpoint - HTTP endpoint; default is cliservice

要进行测试，请使用beeline通过以下方式以http模式连接到JDBC / ODBC服务器：

beeline> !connect jdbc:hive2://<host>:<port>/<database>?hive.server2.transport.mode=http;hive.server2.thrift.http.path=<http_endpoint>

如果您关闭了会话并执行CTAS，则必须在中将其设置`fs.%s.impl.disable.cache`为true `hive-site.xml`。在[\[SPARK-21067\]中][SPARK-21067]查看更多详细信息。

## **运行Spark SQL CLI** ##

Spark SQL CLI是一种方便的工具，可以在本地模式下运行Hive Metastore服务并执行从命令行输入的查询。请注意，Spark SQL CLI无法与Thrift JDBC服务器对话。

要启动Spark SQL CLI，请在Spark目录中运行以下命令：

./bin/spark-sql

蜂巢的结构是通过将您做`hive-site.xml`，`core-site.xml`和`hdfs-site.xml`文件`conf/`。您可以运行`./bin/spark-sql --help`以获取所有可用选项的完整列表。

# **PySpark使用Apache Arrow的熊猫使用指南** #

## **PySpark中的Apache Arrow** ##

Apache Arrow是一种内存中的列式数据格式，在Spark中使用它来在JVM和Python进程之间有效地传输数据。目前，这对于使用Pandas / NumPy数据的Python用户最为有利。它的使用不是自动的，可能需要对配置或代码进行一些小的更改才能充分利用并确保兼容性。本指南将对如何在Spark中使用Arrow进行高层描述，并突出显示使用启用了Arrow的数据时的所有差异。

### **确保已安装PyArrow** ###

要在PySpark中使用Apache Arrow， 应安装[推荐版本的PyArrow][PyArrow]。如果使用pip安装PySpark，则可以使用命令引入PyArrow作为SQL模块的额外依赖项`pip install pyspark[sql]`。否则，您必须确保PyArrow已安装并在所有群集节点上可用。您可以从conda-forge频道使用pip或conda进行安装。有关详细信息，请参见PyArrow [安装][Link 30]。

## **启用与熊猫之间的转换** ##

使用调用将Spark DataFrame转换为Pandas DataFrame并使用`toPandas()`时从Pandas DataFrame创建Spark DataFrame时， 可以使用Arrow作为优化`createDataFrame(pandas_df)`。要在执行这些调用时使用Arrow，用户需要首先将Spark配置设置`spark.sql.execution.arrow.pyspark.enabled`为`true`。默认情况下禁用。

此外，`spark.sql.execution.arrow.pyspark.enabled`如果在Spark内的实际计算之前发生错误，启用的优化可能会自动回退到非箭头优化实现。可以通过控制`spark.sql.execution.arrow.pyspark.fallback.enabled`。

import numpy as np
    import pandas as pd
    
    # Enable Arrow-based columnar data transfers
    spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
    
    # Generate a Pandas DataFrame
    pdf = pd.DataFrame(np.random.rand(100, 3))
    
    # Create a Spark DataFrame from a Pandas DataFrame using Arrow
    df = spark.createDataFrame(pdf)
    
    # Convert the Spark DataFrame back to a Pandas DataFrame using Arrow
    result_pdf = df.select("*").toPandas()

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

将上述优化与Arrow一起使用将产生与未启用Arrow时相同的结果。请注意，即使使用Arrow，也会`toPandas()`导致将DataFrame中的所有记录收集到驱动程序中，并且应该对数据的一小部分进行处理。当前尚不支持所有的Spark数据类型，如果列的类型不受支持，则会引发错误，请参阅[支持的SQL类型][SQL 1]。如果在期间发生错误`createDataFrame()`，Spark将回退以创建不带箭头的DataFrame。

## **熊猫UDF（又名矢量化UDF）** ##

熊猫UDF是用户定义的函数，由Spark使用Arrow来传输数据，并通过Pandas与数据一起使用来进行矢量化操作。熊猫UDF是使用`pandas_udf`修饰符或包装功能定义的，不需要其他配置。熊猫UDF通常表现为常规的PySpark函数API。

在Spark 3.0之前，Pandas UDF以前是通过定义的`PandasUDFType`。在带有Python 3.6+的Spark 3.0中，您还可以使用[Python类型提示][Python 1]。首选使用Python类型提示`PandasUDFType`，在将来的版本中将不推荐使用。

请注意，`pandas.Series`在所有情况下都应使用`pandas.DataFrame`类型提示，但当输入或输出列为of时，应为其输入或输出类型提示使用一种变体`StructType`。以下示例显示了一个Pandas UDF，它使用长列，字符串列和结构列，并输出一个结构列。它需要的功能，以指定的类型提示`pandas.Series`和`pandas.DataFrame`如下：

import pandas as pd
    
    from pyspark.sql.functions import pandas_udf
    
    @pandas_udf("col1 string, col2 long")
    def func(s1: pd.Series, s2: pd.Series, s3: pd.DataFrame) -> pd.DataFrame:
        s3['col2'] = s1 + s2.str.len()
        return s3
    
    # Create a Spark DataFrame that has three columns including a sturct column.
    df = spark.createDataFrame(
        [[1, "a string", ("a nested string",)]],
        "long_col long, string_col string, struct_col struct<col1:string>")
    
    df.printSchema()
    # root
    # |-- long_column: long (nullable = true)
    # |-- string_column: string (nullable = true)
    # |-- struct_column: struct (nullable = true)
    # |    |-- col1: string (nullable = true)
    
    df.select(func("long_col", "string_col", "struct_col")).printSchema()
    # |-- func(long_col, string_col, struct_col): struct (nullable = true)
    # |    |-- col1: string (nullable = true)
    # |    |-- col2: long (nullable = true)

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

在以下各节中，它描述了受支持的类型提示的组合。为了简单起见， `pandas.DataFrame`省略了variant。

### **系列到系列** ###

类型提示可以表示为`pandas.Series`…-> `pandas.Series`。

通过`pandas_udf`与上面具有这种类型提示的函数一起使用，它将创建一个Pandas UDF，其中给定的函数将采用一个或多个`pandas.Series`并输出一个`pandas.Series`。函数的输出应始终与输入具有相同的长度。在内部，PySpark将列拆分成批处理并为每个批处理调用函数作为数据的子集，然后将结果串联在一起，从而执行Pandas UDF。

以下示例显示了如何创建此熊猫UDF来计算2列乘积。

import pandas as pd
    
    from pyspark.sql.functions import col, pandas_udf
    from pyspark.sql.types import LongType
    
    # Declare the function and create the UDF
    def multiply_func(a: pd.Series, b: pd.Series) -> pd.Series:
        return a * b
    
    multiply = pandas_udf(multiply_func, returnType=LongType())
    
    # The function for a pandas_udf should be able to execute with local Pandas data
    x = pd.Series([1, 2, 3])
    print(multiply_func(x, x))
    # 0    1
    # 1    4
    # 2    9
    # dtype: int64
    
    # Create a Spark DataFrame, 'spark' is an existing SparkSession
    df = spark.createDataFrame(pd.DataFrame(x, columns=["x"]))
    
    # Execute function as a Spark vectorized UDF
    df.select(multiply(col("x"), col("x"))).show()
    # +-------------------+
    # |multiply_func(x, x)|
    # +-------------------+
    # |                  1|
    # |                  4|
    # |                  9|
    # +-------------------+

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

有关详细用法，请参阅 [`pyspark.sql.functions.pandas_udf`][pyspark.sql.functions.pandas_udf]

### **系列迭代器到系列迭代器** ###

类型提示可以表示为`Iterator[pandas.Series]`\-> `Iterator[pandas.Series]`。

通过`pandas_udf`与上面具有此类类型提示的函数配合使用，它将创建Pandas UDF，其中给定函数采用的迭代器`pandas.Series`并输出的迭代器`pandas.Series`。该函数的整个输出的长度应与整个输入的长度相同。因此，只要长度相同，它就可以从输入迭代器中预取数据。在这种情况下，调用熊猫UDF时，创建的熊猫UDF需要一个输入列。要使用多个输入列，需要不同的类型提示。请参阅多序列的迭代器到序列的迭代器。

当UDF执行需要初始化一些状态时，它也很有用，尽管在内部它与“系列到系列”的情况相同。下面的伪代码说明了该示例。

@pandas_udf("long")
    def calculate(iterator: Iterator[pd.Series]) -> Iterator[pd.Series]:
        # Do some expensive initialization with a state     state = very_expensive_initialization()
        for x in iterator:
            # Use that state for whole iterator.         yield calculate_with_state(x, state)
    
    df.select(calculate("value")).show()

以下示例显示了如何创建此Pandas UDF：

from typing import Iterator
    
    import pandas as pd
    
    from pyspark.sql.functions import pandas_udf
    
    pdf = pd.DataFrame([1, 2, 3], columns=["x"])
    df = spark.createDataFrame(pdf)
    
    # Declare the function and create the UDF
    @pandas_udf("long")
    def plus_one(iterator: Iterator[pd.Series]) -> Iterator[pd.Series]:
        for x in iterator:
            yield x + 1
    
    df.select(plus_one("x")).show()
    # +-----------+
    # |plus_one(x)|
    # +-----------+
    # |          2|
    # |          3|
    # |          4|
    # +-----------+

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

有关详细用法，请参阅 [`pyspark.sql.functions.pandas_udf`][pyspark.sql.functions.pandas_udf]

### **多个系列的迭代器到系列的迭代器** ###

类型提示可以表示为`Iterator[Tuple[pandas.Series, ...]]`\-> `Iterator[pandas.Series]`。

通过`pandas_udf`与上面具有此类类型提示的函数配合使用，它将创建一个Pandas UDF，其中给定函数采用一个为multiple的元组`pandas.Series`的迭代器，并输出为的迭代器`pandas.Series`。在这种情况下，当调用Pandas UDF时，创建的Pandas UDF需要多个输入列，其数量与元组中的序列数相同。否则，它具有与系列迭代器到系列迭代器案例相同的特性和限制。

以下示例显示了如何创建此Pandas UDF：

from typing import Iterator, Tuple
    
    import pandas as pd
    
    from pyspark.sql.functions import pandas_udf
    
    pdf = pd.DataFrame([1, 2, 3], columns=["x"])
    df = spark.createDataFrame(pdf)
    
    # Declare the function and create the UDF
    @pandas_udf("long")
    def multiply_two_cols(
            iterator: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
        for a, b in iterator:
            yield a * b
    
    df.select(multiply_two_cols("x", "x")).show()
    # +-----------------------+
    # |multiply_two_cols(x, x)|
    # +-----------------------+
    # |                      1|
    # |                      4|
    # |                      9|
    # +-----------------------+

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

有关详细用法，请参阅 [`pyspark.sql.functions.pandas_udf`][pyspark.sql.functions.pandas_udf]

### **系列到标量** ###

类型提示可以表示为`pandas.Series`…-> `Any`。

通过`pandas_udf`与上面具有此类类型提示的函数一起使用，它会创建一个类似于PySpark的聚合函数的Pandas UDF。给定的函数采用`pandas.Series`并返回标量值。返回类型应该是原始数据类型，并且返回的标量可以是python原始类型（例如）`int`或`float`numpy数据类型（例如`numpy.int64`或）`numpy.float64`。 `Any`理想情况下应相应地为特定的标量类型。

该UDF也可以与`groupBy().agg()`和一起使用[`pyspark.sql.Window`][pyspark.sql.Window]。它定义了从一个或多个`pandas.Series`到标量值的聚合，其中每个都`pandas.Series` 代表组或窗口中的一列。

请注意，这种类型的UDF不支持部分聚合，并且组或窗口的所有数据都将被加载到内存中。另外，分组熊猫聚合UDF当前仅支持无边界窗口。下面的示例显示如何使用此类UDF通过group-by和window操作来计算均值

import pandas as pd
    
    from pyspark.sql.functions import pandas_udf
    from pyspark.sql import Window
    
    df = spark.createDataFrame(
        [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
        ("id", "v"))
    
    # Declare the function and create the UDF
    @pandas_udf("double")
    def mean_udf(v: pd.Series) -> float:
        return v.mean()
    
    df.select(mean_udf(df['v'])).show()
    # +-----------+
    # |mean_udf(v)|
    # +-----------+
    # |        4.2|
    # +-----------+
    
    df.groupby("id").agg(mean_udf(df['v'])).show()
    # +---+-----------+
    # | id|mean_udf(v)|
    # +---+-----------+
    # |  1|        1.5|
    # |  2|        6.0|
    # +---+-----------+
    
    w = Window \
        .partitionBy('id') \
        .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
    df.withColumn('mean_v', mean_udf(df['v']).over(w)).show()
    # +---+----+------+
    # | id|   v|mean_v|
    # +---+----+------+
    # |  1| 1.0|   1.5|
    # |  1| 2.0|   1.5|
    # |  2| 3.0|   6.0|
    # |  2| 5.0|   6.0|
    # |  2|10.0|   6.0|
    # +---+----+------+

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

有关详细用法，请参阅 [`pyspark.sql.functions.pandas_udf`][pyspark.sql.functions.pandas_udf]

## **熊猫函数API** ##

熊猫函数API可以`DataFrame`通过使用熊猫实例直接将Python本机函数应用于整个函数。在内部，它与Pandas UDF的工作方式类似，方法是使用Arrow传输数据，使用Pandas处理数据，从而实现矢量化操作。但是，Pandas Function API在PySpark下的行为类似于常规API，`DataFrame`而不是`Column`，并且Pandas Functions API中的Python类型提示是可选的，尽管将来可能会需要它们，但暂时不会影响它在内部的工作方式。

从Spark 3.0开始，分组地图熊猫UDF现在被归类为单独的熊猫功能API `DataFrame.groupby().applyInPandas()`。它仍然有可能与使用它`PandasUDFType` ，并`DataFrame.groupby().apply()`为它; 但是，最好`DataFrame.groupby().applyInPandas()`直接使用 。使用`PandasUDFType`将在未来被废弃。

### **分组Map** ###

支持具有Pandas实例的分组地图操作，`DataFrame.groupby().applyInPandas()` 该操作要求Python函数接受a `pandas.DataFrame`并返回另一个`pandas.DataFrame`。它将每个组映射到`pandas.DataFrame`Python函数中的每个组。

该API实现了“ split-apply-combine”模式，该模式包括三个步骤：

*  通过使用将数据分成几组`DataFrame.groupBy`。
 *  在每个组上应用功能。函数的输入和输出均为`pandas.DataFrame`。输入数据包含每个组的所有行和列。
 *  将结果合并到新的PySpark中`DataFrame`。

要使用`groupBy().applyInPandas()`，用户需要定义以下内容：

*  一个Python函数，用于定义每个组的计算。
 *  甲`StructType`对象或定义输出PySpark的模式的字符串`DataFrame`。

`pandas.DataFrame`如果指定为字符串，则返回的列标签必须与定义的输出模式中的字段名称匹配，或者如果不是字符串，则必须按位置匹配字段数据类型，例如整数索引。有关 在构造时如何标记列的信息，请参见[pandas.DataFrame][]`pandas.DataFrame`。

请注意，在应用该功能之前，组的所有数据将被加载到内存中。这可能会导致内存不足异常，尤其是在组大小偏斜的情况下。[maxRecordsPerBatch][]的配置 不适用于组，并且由用户决定是否将分组的数据放入可用内存中。

以下示例说明如何使用`groupby().applyInPandas()`从组中的每个值减去平均值

df = spark.createDataFrame(
        [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
        ("id", "v"))
    
    def subtract_mean(pdf):
        # pdf is a pandas.DataFrame
        v = pdf.v
        return pdf.assign(v=v - v.mean())
    
    df.groupby("id").applyInPandas(subtract_mean, schema="id long, v double").show()
    # +---+----+
    # | id|   v|
    # +---+----+
    # |  1|-0.5|
    # |  1| 0.5|
    # |  2|-3.0|
    # |  2|-1.0|
    # |  2| 4.0|
    # +---+----+

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

有关详细用法，请参阅[`pyspark.sql.GroupedData.applyInPandas`][pyspark.sql.GroupedData.applyInPandas]。

### **Map操作** ###

支持使用Pandas实例进行Map操作，`DataFrame.mapInPandas()`该操作将`pandas.DataFrame`s 的迭代器映射到s的另一个迭代器，`pandas.DataFrame`该迭代器表示当前PySpark `DataFrame`并将结果作为PySpark返回`DataFrame`。该函数接受并输出的迭代器`pandas.DataFrame`。与某些Pandas UDF相比，它可以返回任意长度的输出，尽管在内部它与Series to Series Pandas UDF类似。

以下示例显示如何使用`mapInPandas()`

df = spark.createDataFrame([(1, 21), (2, 30)], ("id", "age"))
    
    def filter_func(iterator):
        for pdf in iterator:
            yield pdf[pdf.id == 1]
    
    df.mapInPandas(filter_func, schema=df.schema).show()
    # +---+---+
    # | id|age|
    # +---+---+
    # |  1| 21|
    # +---+---+

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

有关详细用法，请参阅[`pyspark.sql.DataFrame.mapsInPandas`][pyspark.sql.DataFrame.mapsInPandas]。

### **共同Map** ###

支持与Pandas实例的联合分组Map操作，`DataFrame.groupby().cogroup().applyInPandas()`该操作允许`DataFrame`通过一个公用密钥将两个PySpark 联合分组，然后将Python函数应用于每个联合分组。它包括以下步骤：

*  对数据进行混洗，以使共享密钥的每个数据帧的组共同分组。
 *  将一个功能应用于每个共同组。该函数的输入为2 `pandas.DataFrame`（带有一个可选的表示键的元组）。该函数的输出为`pandas.DataFrame`。
 *  将`pandas.DataFrame`所有组中的合并到新的PySpark中`DataFrame`。

要使用`groupBy().cogroup().applyInPandas()`，用户需要定义以下内容：

*  一个Python函数，用于定义每个协同组的计算。
 *  甲`StructType`对象或定义输出PySpark的模式的字符串`DataFrame`。

请注意，在应用该功能之前，一个cogroup的所有数据将被加载到内存中。这可能会导致内存不足异常，尤其是在组大小偏斜的情况下。[maxRecordsPerBatch][]的配置 未应用，并且取决于用户，以确保共同分组的数据将适合可用内存。

以下示例显示了如何用于`groupby().cogroup().applyInPandas()`在两个数据集之间执行asof连接。

import pandas as pd
    
    df1 = spark.createDataFrame(
        [(20000101, 1, 1.0), (20000101, 2, 2.0), (20000102, 1, 3.0), (20000102, 2, 4.0)],
        ("time", "id", "v1"))
    
    df2 = spark.createDataFrame(
        [(20000101, 1, "x"), (20000101, 2, "y")],
        ("time", "id", "v2"))
    
    def asof_join(l, r):
        return pd.merge_asof(l, r, on="time", by="id")
    
    df1.groupby("id").cogroup(df2.groupby("id")).applyInPandas(
        asof_join, schema="time int, id int, v1 double, v2 string").show()
    # +--------+---+---+---+
    # |    time| id| v1| v2|
    # +--------+---+---+---+
    # |20000101|  1|1.0|  x|
    # |20000102|  1|3.0|  x|
    # |20000101|  2|2.0|  y|
    # |20000102|  2|4.0|  y|
    # +--------+---+---+---+

在Spark存储库中的“ examples / src / main / python / sql / arrow.py”中找到完整的示例代码。

有关详细用法，请参阅[`pyspark.sql.PandasCogroupedOps.applyInPandas()`][pyspark.sql.PandasCogroupedOps.applyInPandas]。

## **使用说明** ##

### **支持的SQL类型** ###

目前，所有Spark SQL数据类型是基于箭转换，除了支持`MapType`， `ArrayType`中`TimestampType`和嵌套`StructType`。

### **设置箭头批处理大小** ###

Spark中的数据分区将转换为Arrow记录批，这可能会暂时导致JVM中的内存使用率很高。为避免可能的内存不足异常，可以通过将conf“ spark.sql.execution.arrow.maxRecordsPerBatch”设置为整数来调整Arrow记录批的大小，该整数将确定每个批处理的最大行数。默认值为每批10,000条记录。如果列数很大，则应相应地调整该值。使用此限制，每个数据分区将被分为1个或多个记录批次以进行处理。

### **带时区语义的时间戳** ###

Spark在内部将时间戳存储为UTC值，并且在没有指定时区的情况下传入的时间戳数据将作为本地时间转换为具有微秒分辨率的UTC。当时间戳数据导出或在Spark中显示时，会话时区用于本地化时间戳值。会话时区使用配置“ spark.sql.session.timeZone”设置，如果未设置，则默认为JVM系统本地时区。Pandas使用`datetime64`具有纳秒分辨率的类型，`datetime64[ns]`每个列都具有可选的时区。

当时间戳数据从Spark传输到Pandas时，它将转换为纳秒，每列将转换为Spark会话时区，然后本地化到该时区，从而删除时区并将值显示为本地时间。这将在调用时`toPandas()`或`pandas_udf`使用时间戳列时发生。

当时间戳数据从Pandas传输到Spark时，它将转换为UTC微秒。在`createDataFrame`使用Pandas DataFrame进行调用或从返回时间戳时， 会发生这种情况`pandas_udf`。这些转换是自动完成的，以确保Spark拥有预期格式的数据，因此您无需自己进行任何转换。任何毫微秒的值都会被截断。

请注意，标准UDF（非Pandas）会将时间戳数据加载为Python日期时间对象，这与Pandas时间戳不同。建议在使用`pandas_udf`s中的时间戳时使用Pandas时间序列功能以获得最佳性能，有关详细信息，请参见 [此处][Link 31]。

### **推荐的Pandas和PyArrow版本** ###

对于与pyspark.sql一起使用，支持的Pandas版本是0.24.2，而PyArrow是0.15.1。可以使用更高版本，但是不能保证兼容性和数据正确性，并且应由用户验证。

### **PyArrow> = 0.15.0和Spark 2.3.x，2.4.x的兼容性设置** ###

从Arrow 0.15.0开始，二进制IPC格式的更改要求环境变量与Arrow <= 0.14.1的早期版本兼容。这仅对于版本2.3.x和2.4.x且已将PyArrow手动升级到0.15.0的PySpark用户需要执行。可以添加以下内容`conf/spark-env.sh`以使用旧版Arrow IPC格式：

ARROW_PRE_0_15_IPC_FORMAT=1

这将指示PyArrow> = 0.15.0将旧版IPC格式与Spark 2.3.x和2.4.x中的较旧Arrow Java一起使用。运行s或启用Arrow 时， 未设置此环境变量将导致类似[SPARK-29367中][SPARK-29367]所述的错误。有关Arrow IPC更改的更多信息，请参见Arrow 0.15.0版本[博客][Link 32]。`pandas_udftoPandas()`

# **迁移指南** #

现在，迁移指南已存档[在此页面上][Link 33]。

# **SQL参考** #

Spark SQL是Apache Spark的用于处理结构化数据的模块。本指南是结构化查询语言（SQL）的参考，包括语法，语义，关键字和常见SQL使用示例。它包含有关以下主题的信息：

*  [符合ANSI][ANSI]
 *  [资料类型][Link 34]
 *  [日期时间模式][Link 35]
 *  [功能][Link 36]
    
     *  [内建功能][Link 37]
     *  [标量用户定义函数（UDF）][Link 5]
     *  [用户定义的聚合函数（UDAF）][Link 7]
     *  [与Hive UDF / UDAF / UDTF集成][Hive UDF _ UDAF _ UDTF]
 *  [身份标识][Link 38]
 *  [文字][Link 39]
 *  [空语义][Link 40]
 *  [SQL语法][SQL 2]
    
     *  [DDL陈述式][DDL]
     *  [DML语句][DML]
     *  [数据检索语句][Link 41]
     *  [辅助声明][Link 42]

[Hive Tables]: https://spark.apache.org/docs/latest/sql-data-sources-hive-tables.html
[Dataset _ DataFrame]: https://spark.apache.org/docs/latest/sql-programming-guide.html#datasets-and-dataframes
[Link 1]: https://spark.apache.org/docs/latest/sql-distributed-sql-engine.html#running-the-spark-sql-cli
[JDBC _ ODBC]: https://spark.apache.org/docs/latest/sql-distributed-sql-engine.html#running-the-thrift-jdbcodbc-server
[Link 2]: https://spark.apache.org/docs/latest/sql-getting-started.html#creating-datasets
[Scala]: https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html
[Java]: https://spark.apache.org/docs/latest/api/java/index.html?org/apache/spark/sql/Dataset.html
[Link 3]: https://spark.apache.org/docs/latest/sql-data-sources.html
[Python]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame
[R]: https://spark.apache.org/docs/latest/api/R/index.html
[SparkSession]: https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/SparkSession.html
[SparkSession 1]: https://spark.apache.org/docs/latest/api/java/index.html#org.apache.spark.sql.SparkSession
[SparkSession 2]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.SparkSession
[SparkSession 3]: https://spark.apache.org/docs/latest/api/R/sparkR.session.html
[RDD]: https://spark.apache.org/docs/latest/sql-getting-started.html#interoperating-with-rdds
[R 1]: https://spark.apache.org/docs/latest/api/R/SparkDataFrame.html
[DataFrame Function Reference]: https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/functions$.html
[API]: https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Dataset.html
[DataFrame Function Reference 1]: https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/functions.html
[DataFrame Function Reference 2]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#module-pyspark.sql.functions
[Encoder]: https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Encoder.html
[JavaBean]: http://stackoverflow.com/questions/3295496/what-is-a-javabean-exactly
[Link 4]: https://spark.apache.org/docs/latest/sql-ref-functions.html#scalar-functions
[Link 5]: https://spark.apache.org/docs/latest/sql-ref-functions-udf-scalar.html
[Link 6]: https://spark.apache.org/docs/latest/sql-ref-functions-builtin.html#aggregate-functions
[Link 7]: https://spark.apache.org/docs/latest/sql-ref-functions-udf-aggregate.html
[Link 8]: https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html
[Link 9]: https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#manually-specifying-options
[SQL]: https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#run-sql-on-files-directly
[Link 10]: https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#save-modes
[Link 11]: https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#saving-to-persistent-tables
[Link 12]: https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#bucketing-sorting-and-partitioning
[Link 13]: https://spark.apache.org/docs/latest/sql-data-sources-generic-options.html
[Link 14]: https://spark.apache.org/docs/latest/sql-data-sources-generic-options.html#ignore-corrupt-iles
[Link 15]: https://spark.apache.org/docs/latest/sql-data-sources-generic-options.html#ignore-missing-iles
[Link 16]: https://spark.apache.org/docs/latest/sql-data-sources-generic-options.html#path-global-filter
[Link 17]: https://spark.apache.org/docs/latest/sql-data-sources-generic-options.html#recursive-file-lookup
[Link 18]: https://spark.apache.org/docs/latest/sql-data-sources-parquet.html
[Link 19]: https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#loading-data-programmatically
[Link 20]: https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#partition-discovery
[Link 21]: https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#schema-merging
[Hive Metastore Parquet]: https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#hive-metastore-parquet-table-conversion
[Link 22]: https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#configuration
[ORC]: https://spark.apache.org/docs/latest/sql-data-sources-orc.html
[JSON]: https://spark.apache.org/docs/latest/sql-data-sources-json.html
[Hive]: https://spark.apache.org/docs/latest/sql-data-sources-hive-tables.html#specifying-storage-format-for-hive-tables
[Hive Metastore]: https://spark.apache.org/docs/latest/sql-data-sources-hive-tables.html#interacting-with-different-versions-of-hive-metastore
[JDBC]: https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html
[Avro]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html
[Link 23]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#deploying
[Link 24]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#load-and-save-functions
[to_avro_from_avro]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#to_avro-and-from_avro
[Link 25]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#data-source-option
[Link 26]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#configuration
[Databricks spark-avro]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#compatibility-with-databricks-spark-avro
[Avro-_ Spark SQL]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#supported-types-for-avro---spark-sql-conversion
[Spark SQL_-_ Avro]: https://spark.apache.org/docs/latest/sql-data-sources-avro.html#supported-types-for-spark-sql---avro-conversion
[Link 27]: https://spark.apache.org/docs/latest/sql-data-sources-binaryFile.html
[Link 28]: https://spark.apache.org/docs/latest/sql-data-sources-troubleshooting.html
[Join Hints]: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-hints.html#join-hints
[Link 29]: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-hints.html#partitioning-hints
[HiveServer2]: https://cwiki.apache.org/confluence/display/Hive/Setting+Up+HiveServer2
[beeline]: https://cwiki.apache.org/confluence/display/Hive/HiveServer2+Clients
[SPARK-21067]: https://issues.apache.org/jira/browse/SPARK-21067
[PyArrow]: https://spark.apache.org/docs/latest/sql-pyspark-pandas-with-arrow.html#recommended-pandas-and-pyarrow-versions
[Link 30]: https://arrow.apache.org/docs/python/install.html
[SQL 1]: https://spark.apache.org/docs/latest/sql-pyspark-pandas-with-arrow.html#supported-sql-types
[Python 1]: https://www.python.org/dev/peps/pep-0484
[pyspark.sql.functions.pandas_udf]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.functions.pandas_udf
[pyspark.sql.Window]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.Window
[pandas.DataFrame]: https://pandas.pydata.org/pandas-docs/stable/generated/pandas.DataFrame.html#pandas.DataFrame
[maxRecordsPerBatch]: https://spark.apache.org/docs/latest/sql-pyspark-pandas-with-arrow.html#setting-arrow-batch-size
[pyspark.sql.GroupedData.applyInPandas]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.GroupedData.applyInPandas
[pyspark.sql.DataFrame.mapsInPandas]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame.mapInPandas
[pyspark.sql.PandasCogroupedOps.applyInPandas]: https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.PandasCogroupedOps.applyInPandas
[Link 31]: https://pandas.pydata.org/pandas-docs/stable/timeseries.html
[SPARK-29367]: https://issues.apache.org/jira/browse/SPARK-29367
[Link 32]: http://arrow.apache.org/blog/2019/10/06/0.15.0-release/#columnar-streaming-protocol-change-since-0140
[Link 33]: https://spark.apache.org/docs/latest/sql-migration-guide.html
[ANSI]: https://spark.apache.org/docs/latest/sql-ref-ansi-compliance.html
[Link 34]: https://spark.apache.org/docs/latest/sql-ref-datatypes.html
[Link 35]: https://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html
[Link 36]: https://spark.apache.org/docs/latest/sql-ref-functions.html
[Link 37]: https://spark.apache.org/docs/latest/sql-ref-functions-builtin.html
[Hive UDF _ UDAF _ UDTF]: https://spark.apache.org/docs/latest/sql-ref-functions-udf-hive.html
[Link 38]: https://spark.apache.org/docs/latest/sql-ref-identifier.html
[Link 39]: https://spark.apache.org/docs/latest/sql-ref-literals.html
[Link 40]: https://spark.apache.org/docs/latest/sql-ref-null-semantics.html
[SQL 2]: https://spark.apache.org/docs/latest/sql-ref-syntax.html
[DDL]: https://spark.apache.org/docs/latest/sql-ref-syntax-ddl.html
[DML]: https://spark.apache.org/docs/latest/sql-ref-syntax-dml.html
[Link 41]: https://spark.apache.org/docs/latest/sql-ref-syntax-qry.html
[Link 42]: https://spark.apache.org/docs/latest/sql-ref-syntax-aux.html