Spark SQL中所有功能的入口点是 SparkSession 类. 要创建一个 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._

在一个 SparkSession中, 应用程序可以从一个 已经存在的 RDD, 从hive表, 或者从 Spark数据源中创建一个DataFrames.

举个例子, 下面就是基于一个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|

// +----+-------+

// 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|

// +----+-----+

能够在 DataFrame 上被执行的操作类型的完整列表请参考 API 文档.

除了简单的列引用和表达式之外, DataFrame 也有丰富的函数库, 包括 string 操作, date 算术, 常见的 math 操作以及更多.可用的完整列表请参考  DataFrame 函数指南.

SparkSession 的 sql 函数可以让应用程序以编程的方式运行 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 SQL 的 Scala 接口支持自动转换一个包含 case classes 的 RDD 为 DataFrame.Case class 定义了表的 Schema.Case class 的参数名使用反射读取并且成为了列名.Case class 也可以是嵌套的或者包含像 Seq 或者 Array 这样的复杂类型.这个 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))

当 case class 不能够在执行之前被定义（例如, records 记录的结构在一个 string 字符串中被编码了, 或者一个 text 文本 dataset 将被解析并且不同的用户投影的字段是不一样的）.一个 DataFrame 可以使用下面的三步以编程的方式来创建.

1. 从原始的 RDD 创建 RDD 的 Row（行）;
2. Step 1 被创建后, 创建 Schema 表示一个 StructType 匹配 RDD 中的 Row（行）的结构.
3. 通过 SparkSession 提供的 createDataFrame 方法应用 Schema 到 RDD 的 RowS（行）.

例如:

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|

// +-------------+

Users have to extend the UserDefinedAggregateFunction abstract class to implement a custom untyped aggregate function. For example, a user-defined average can look like:

import org.apache.spark.sql.expressions.MutableAggregationBuffer

import org.apache.spark.sql.expressions.UserDefinedAggregateFunction

import org.apache.spark.sql.types._

import org.apache.spark.sql.Row

import org.apache.spark.sql.SparkSession

object MyAverage extends UserDefinedAggregateFunction {

  // Data types of input arguments of this aggregate function

  def inputSchema: StructType = StructType(StructField("inputColumn", LongType) :: Nil)

  // Data types of values in the aggregation buffer

  def bufferSchema: StructType = {

    StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)

  }

  // The data type of the returned value

  def dataType: DataType = DoubleType

  // Whether this function always returns the same output on the identical input

  def deterministic: Boolean = true

  // Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to

  // standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides

  // the opportunity to update its values. Note that arrays and maps inside the buffer are still

  // immutable.

  def initialize(buffer: MutableAggregationBuffer): Unit = {

    buffer(0) = 0L

    buffer(1) = 0L

  }

  // Updates the given aggregation buffer `buffer` with new input data from `input`

  def update(buffer: MutableAggregationBuffer, input: Row): Unit = {

    if (!input.isNullAt(0)) {

      buffer(0) = buffer.getLong(0) + input.getLong(0)

      buffer(1) = buffer.getLong(1) + 1

    }

  }

  // Merges two aggregation buffers and stores the updated buffer values back to `buffer1`

  def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {

    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)

    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)

  }

  // Calculates the final result

  def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)

}

// Register the function to access it

spark.udf.register("myAverage", MyAverage)

val df = spark.read.json("examples/src/main/resources/employees.json")

df.createOrReplaceTempView("employees")

df.show()

// +-------+------+

// |   name|salary|

// +-------+------+

// |Michael|  3000|

// |   Andy|  4500|

// | Justin|  3500|

// |  Berta|  4000|

// +-------+------+

val result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees")

result.show()

// +--------------+

// |average_salary|

// +--------------+

// |        3750.0|

// +--------------+

// spark is an existing SparkSession

spark.catalog.refreshTable("my_table")

Spark SQL 可以 automatically infer （自动推断）JSON dataset 的 schema, 并将其作为 Dataset[Row] 加载. 这个 conversion （转换）可以在 Dataset[String] 上使用 SparkSession.read.json() 来完成, 或 JSON 文件.

请注意, 以 a json file 提供的文件不是典型的 JSON 文件. 每行必须包含一个 separate （单独的）, self-contained valid （独立的有效的）JSON 对象. 有关更多信息, 请参阅 JSON Lines text format, also called newline-delimited JSON .

对于 regular multi-line JSON file （常规的多行 JSON 文件）, 将 multiLine 选项设置为 true .

// Primitive types (Int, String, etc) and Product types (case classes) encoders are

// supported by importing this when creating a Dataset.

import spark.implicits._

// A JSON dataset is pointed to by path.

// The path can be either a single text file or a directory storing text files

val path = "examples/src/main/resources/people.json"

val peopleDF = spark.read.json(path)

// The inferred schema can be visualized using the printSchema() method

peopleDF.printSchema()

// root

//  |-- age: long (nullable = true)

//  |-- name: string (nullable = true)

// Creates a temporary view using the DataFrame

peopleDF.createOrReplaceTempView("people")

// SQL statements can be run by using the sql methods provided by spark

val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")

teenagerNamesDF.show()

// +------+

// |  name|

// +------+

// |Justin|

// +------+

// Alternatively, a DataFrame can be created for a JSON dataset represented by

// a Dataset[String] storing one JSON object per string

val otherPeopleDataset = spark.createDataset(

  """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)

val otherPeople = spark.read.json(otherPeopleDataset)

otherPeople.show()

// +---------------+----+

// |        address|name|

// +---------------+----+

// |[Columbus,Ohio]| Yin|

// +---------------+----+

// In 1.3.x, in order for the grouping column "department" to show up,

// it must be included explicitly as part of the agg function call.

df.groupBy("department").agg($"department", max("age"), sum("expense"))

// In 1.4+, grouping column "department" is included automatically.

df.groupBy("department").agg(max("age"), sum("expense"))

// Revert to 1.3 behavior (not retaining grouping column) by:

sqlContext.setConf("spark.sql.retainGroupColumns", "false")

sqlContext.udf.register("strLen", (s: String) => s.length())

Spark SQL 的所有数据类型都在包 org.apache.spark.sql.types 中. 你可以用下示例示例来访问它们.

import org.apache.spark.sql.types._

Spark SQL, DataFrames and Datasets Guide

• Overview
  • SQL
  • Datasets and DataFrames
• 开始入门
  • 起始点: SparkSession
  • 创建 DataFrames
  • 无类型的Dataset操作 (aka DataFrame 操作)
  • Running SQL Queries Programmatically
  • 全局临时视图
  • 创建Datasets
  • RDD的互操作性
    • 使用反射推断Schema
    • 以编程的方式指定Schema
  • Aggregations
    • Untyped User-Defined Aggregate Functions
    • Type-Safe User-Defined Aggregate Functions
• Data Sources （数据源）
  • Generic Load/Save Functions （通用 加载/保存 功能）
    • Manually Specifying Options （手动指定选项）
    • Run SQL on files directly （直接在文件上运行 SQL）
    • Save Modes （保存模式）
    • Saving to Persistent Tables （保存到持久表）
    • Bucketing, Sorting and Partitioning （分桶, 排序和分区）
  • Parquet Files
    • Loading Data Programmatically （以编程的方式加载数据）
    • Partition Discovery （分区发现）
    • Schema Merging （模式合并）
    • Hive metastore Parquet table conversion （Hive metastore Parquet table 转换）
      • Hive/Parquet Schema Reconciliation
      • Metadata Refreshing （元数据刷新）
    • Configuration （配置）
  • JSON Datasets （JSON 数据集）
  • Hive 表
    • 指定 Hive 表的存储格式
    • 与不同版本的 Hive Metastore 进行交互
  • JDBC 连接其它数据库
  • 故障排除
• 性能调优
  • 在内存中缓存数据
  • 其他配置选项
• 分布式 SQL 引擎
  • 运行 Thrift JDBC/ODBC 服务器
  • 运行 Spark SQL CLI
• 迁移指南
  • 从 Spark SQL 2.1 升级到 2.2
  • 从 Spark SQL 2.0 升级到 2.1
  • 从 Spark SQL 1.6 升级到 2.0
  • 从 Spark SQL 1.5 升级到 1.6
  • 从 Spark SQL 1.4 升级到 1.5
  • 从 Spark SQL 1.3 升级到 1.4
    • DataFrame data reader/writer interface
    • DataFrame.groupBy 保留 grouping columns（分组的列）
    • DataFrame.withColumn 上的行为更改
  • 从 Spark SQL 1.0-1.2 升级到 1.3
    • 重命名 DataFrame 的 SchemaRDD
    • Java 和 Scala APIs 的统一
    • 隔离隐式转换和删除 dsl 包（仅Scala）
    • 针对 DataType 删除在 org.apache.spark.sql 包中的一些类型别名（仅限于 Scala）
    • UDF 注册迁移到 sqlContext.udf 中 (Java & Scala)
    • Python DataTypes 不再是 Singletons（单例的）
  • 与 Apache Hive 的兼容
    • 在现有的 Hive Warehouses 中部署
    • 所支持的 Hive 特性
    • 未支持的 Hive 函数
• 参考
  • 数据类型
  • NaN Semantics

Overview

Spark SQL 是 Spark 处理结构化数据的一个模块.与基础的 Spark RDD API 不同, Spark SQL 提供了查询结构化数据及计算结果等信息的接口.在内部, Spark SQL 使用这个额外的信息去执行额外的优化.有几种方式可以跟 Spark SQL 进行交互, 包括 SQL 和 Dataset API.当使用相同执行引擎进行计算时, 无论使用哪种 API / 语言都可以快速的计算.这种统一意味着开发人员能够在基于提供最自然的方式来表达一个给定的 transformation API 之间实现轻松的来回切换不同的 .

该页面所有例子使用的示例数据都包含在 Spark 的发布中, 并且可以使用 spark-shell, pyspark shell, 或者 sparkR shell来运行.

SQL

Spark SQL 的功能之一是执行 SQL 查询.Spark SQL 也能够被用于从已存在的 Hive 环境中读取数据.更多关于如何配置这个特性的信息, 请参考 Hive 表 这部分. 当以另外的编程语言运行SQL 时, 查询结果将以 Dataset/DataFrame的形式返回.您也可以使用 命令行或者通过 JDBC/ODBC与 SQL 接口交互.

Datasets and DataFrames

一个 Dataset 是一个分布式的数据集合 Dataset 是在 Spark 1.6 中被添加的新接口, 它提供了 RDD 的优点（强类型化, 能够使用强大的 lambda 函数）与Spark SQL执行引擎的优点.一个 Dataset 可以从 JVM 对象来 构造 并且使用转换功能（map, flatMap, filter, 等等）. Dataset API 在Scala 和Java是可用的.Python 不支持 Dataset API.但是由于 Python 的动态特性, 许多 Dataset API 的优点已经可用了 (也就是说, 你可能通过 name 天生的row.columnName属性访问一行中的字段).这种情况和 R 相似.

一个 DataFrame 是一个 Dataset 组成的指定列.它的概念与一个在关系型数据库或者在 R/Python 中的表是相等的, 但是有很多优化. DataFrames 可以从大量的 sources 中构造出来, 比如: 结构化的文本文件, Hive中的表, 外部数据库, 或者已经存在的 RDDs. DataFrame API 可以在 Scala, Java, Python, 和 R中实现. 在 Scala 和 Java中, 一个 DataFrame 所代表的是一个多个 Row（行）的的 Dataset（数据集合）. 在 the Scala API中, DataFrame仅仅是一个 Dataset[Row]类型的别名. 然而, 在 Java API中, 用户需要去使用 Dataset<Row> 去代表一个 DataFrame.

在此文档中, 我们将常常会引用 Scala/Java Datasets 的 Rows 作为 DataFrames.

开始入门

起始点: SparkSession

• Scala
• Java
• Python
• R

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 2.0 中的SparkSession 为 Hive 特性提供了内嵌的支持, 包括使用 HiveQL 编写查询的能力, 访问 Hive UDF,以及从 Hive 表中读取数据的能力.为了使用这些特性, 你不需要去有一个已存在的 Hive 设置.

创建 DataFrames

• Scala
• Java
• Python
• R

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|

// +----+-------+

无类型的Dataset操作 (aka DataFrame 操作)

DataFrames 提供了一个特定的语法用在 Scala, Java, Python and R中机构化数据的操作.

正如上面提到的一样, Spark 2.0中, DataFrames在Scala 和 Java API中, 仅仅是多个 Rows的Dataset. 这些操作也参考了与强类型的Scala/Java Datasets中的”类型转换” 对应的”无类型转换” .

这里包括一些使用 Dataset 进行结构化数据处理的示例 :

• Scala
• Java
• Python
• R

// 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|

// +----+-----+

Running SQL Queries Programmatically

• Scala
• Java
• Python
• R

// 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 SQL中的临时视图是session级别的, 也就是会随着session的消失而消失. 如果你想让一个临时视图在所有session中相互传递并且可用, 直到Spark 应用退出, 你可以建立一个全局的临时视图.全局的临时视图存在于系统数据库 global_temp中, 我们必须加上库名去引用它, 比如. SELECT * FROM global_temp.view1.

• Scala
• Java
• Python
• Sql

// 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|

// +----+-------+

创建Datasets

Dataset 与 RDD 相似, 然而, 并不是使用 Java 序列化或者 Kryo 编码器 来序列化用于处理或者通过网络进行传输的对象. 虽然编码器和标准的序列化都负责将一个对象序列化成字节, 编码器是动态生成的代码, 并且使用了一种允许 Spark 去执行许多像 filtering, sorting 以及 hashing 这样的操作, 不需要将字节反序列化成对象的格式.

• Scala
• Java

// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,

// you can use custom classes that implement the Product interface

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|

// +----+-------+

RDD的互操作性

Spark SQL 支持两种不同的方法用于转换已存在的 RDD 成为 Dataset.第一种方法是使用反射去推断一个包含指定的对象类型的 RDD 的 Schema.在你的 Spark 应用程序中当你已知 Schema 时这个基于方法的反射可以让你的代码更简洁.

第二种用于创建 Dataset 的方法是通过一个允许你构造一个 Schema 然后把它应用到一个已存在的 RDD 的编程接口.然而这种方法更繁琐, 当列和它们的类型知道运行时都是未知时它允许你去构造 Dataset.

使用反射推断Schema

• Scala
• Java
• Python

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

以编程的方式指定Schema

• Scala
• Java
• Python

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|

// +-------------+

Aggregations

The built-in DataFrames functions provide common aggregations such as count(), countDistinct(), avg(), max(), min(), etc. While those functions are designed for DataFrames, Spark SQL also has type-safe versions for some of them in Scala and Java to work with strongly typed Datasets. Moreover, users are not limited to the predefined aggregate functions and can create their own.

Untyped User-Defined Aggregate Functions

• Scala
• Java

import org.apache.spark.sql.expressions.MutableAggregationBuffer

import org.apache.spark.sql.expressions.UserDefinedAggregateFunction

import org.apache.spark.sql.types._

import org.apache.spark.sql.Row

import org.apache.spark.sql.SparkSession

object MyAverage extends UserDefinedAggregateFunction {

  // Data types of input arguments of this aggregate function

  def inputSchema: StructType = StructType(StructField("inputColumn", LongType) :: Nil)

  // Data types of values in the aggregation buffer

  def bufferSchema: StructType = {

    StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)

  }

  // The data type of the returned value

  def dataType: DataType = DoubleType

  // Whether this function always returns the same output on the identical input

  def deterministic: Boolean = true

  // Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to

  // standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides

  // the opportunity to update its values. Note that arrays and maps inside the buffer are still

  // immutable.

  def initialize(buffer: MutableAggregationBuffer): Unit = {

    buffer(0) = 0L

    buffer(1) = 0L

  }

  // Updates the given aggregation buffer `buffer` with new input data from `input`

  def update(buffer: MutableAggregationBuffer, input: Row): Unit = {

    if (!input.isNullAt(0)) {

      buffer(0) = buffer.getLong(0) + input.getLong(0)

      buffer(1) = buffer.getLong(1) + 1

    }

  }

  // Merges two aggregation buffers and stores the updated buffer values back to `buffer1`

  def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {

    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)

    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)

  }

  // Calculates the final result

  def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)

}

// Register the function to access it

spark.udf.register("myAverage", MyAverage)

val df = spark.read.json("examples/src/main/resources/employees.json")

df.createOrReplaceTempView("employees")

df.show()

// +-------+------+

// |   name|salary|

// +-------+------+

// |Michael|  3000|

// |   Andy|  4500|

// | Justin|  3500|

// |  Berta|  4000|

// +-------+------+

val result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees")

result.show()

// +--------------+

// |average_salary|

// +--------------+

// |        3750.0|

// +--------------+

Type-Safe User-Defined Aggregate Functions

User-defined aggregations for strongly typed Datasets revolve around the Aggregator abstract class. For example, a type-safe user-defined average can look like:

• Scala
• Java

import org.apache.spark.sql.expressions.Aggregator

import org.apache.spark.sql.Encoder

import org.apache.spark.sql.Encoders

import org.apache.spark.sql.SparkSession

case class Employee(name: String, salary: Long)

case class Average(var sum: Long, var count: Long)

object MyAverage extends Aggregator[Employee, Average, Double] {

  // A zero value for this aggregation. Should satisfy the property that any b + zero = b

  def zero: Average = Average(0L, 0L)

  // Combine two values to produce a new value. For performance, the function may modify `buffer`

  // and return it instead of constructing a new object

  def reduce(buffer: Average, employee: Employee): Average = {

    buffer.sum += employee.salary

    buffer.count += 1

    buffer

  }

  // Merge two intermediate values

  def merge(b1: Average, b2: Average): Average = {

    b1.sum += b2.sum

    b1.count += b2.count

    b1

  }

  // Transform the output of the reduction

  def finish(reduction: Average): Double = reduction.sum.toDouble / reduction.count

  // Specifies the Encoder for the intermediate value type

  def bufferEncoder: Encoder[Average] = Encoders.product

  // Specifies the Encoder for the final output value type

  def outputEncoder: Encoder[Double] = Encoders.scalaDouble

}

val ds = spark.read.json("examples/src/main/resources/employees.json").as[Employee]

ds.show()

// +-------+------+

// |   name|salary|

// +-------+------+

// |Michael|  3000|

// |   Andy|  4500|

// | Justin|  3500|

// |  Berta|  4000|

// +-------+------+

// Convert the function to a `TypedColumn` and give it a name

val averageSalary = MyAverage.toColumn.name("average_salary")

val result = ds.select(averageSalary)

result.show()

// +--------------+

// |average_salary|

// +--------------+

// |        3750.0|

// +--------------+

Data Sources （数据源）

Spark SQL 支持通过 DataFrame 接口对各种 data sources （数据源）进行操作. DataFrame 可以使用 relational transformations （关系转换）操作, 也可用于创建 temporary view （临时视图）. 将 DataFrame 注册为 temporary view （临时视图）允许您对其数据运行 SQL 查询. 本节 描述了使用 Spark Data Sources 加载和保存数据的一般方法, 然后涉及可用于 built-in data sources （内置数据源）的 specific options （特定选项）.

Generic Load/Save Functions （通用 加载/保存 功能）

在最简单的形式中, 默认数据源（parquet, 除非另有配置 spark.sql.sources.default ）将用于所有操作.

• Scala
• Java
• Python
• R

val usersDF = spark.read.load("examples/src/main/resources/users.parquet")

usersDF.select("name", "favorite_color").write.save("namesAndFavColors.parquet")

Manually Specifying Options （手动指定选项）

您还可以 manually specify （手动指定）将与任何你想传递给 data source 的其他选项一起使用的 data source . Data sources 由其 fully qualified name （完全限定名称）（即 org.apache.spark.sql.parquet ）, 但是对于 built-in sources （内置的源）, 你也可以使用它们的 shortnames （短名称）（json, parquet, jdbc, orc, libsvm, csv, text）.从任何 data source type （数据源类型）加载 DataFrames 可以使用此 syntax （语法）转换为其他类型.

• Scala
• Java
• Python
• R

val peopleDF = spark.read.format("json").load("examples/src/main/resources/people.json")

peopleDF.select("name", "age").write.format("parquet").save("namesAndAges.parquet")

Run SQL on files directly （直接在文件上运行 SQL）

不使用读取 API 将文件加载到 DataFrame 并进行查询, 也可以直接用 SQL 查询该文件.

• Scala
• Java
• Python
• R

val sqlDF = spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`")

Save Modes （保存模式）

Save operations （保存操作）可以选择使用 SaveMode , 它指定如何处理现有数据如果存在的话. 重要的是要意识到, 这些 save modes （保存模式）不使用任何 locking （锁定）并且不是 atomic （原子）. 另外, 当执行 Overwrite 时, 数据将在新数据写出之前被删除.

Saving to Persistent Tables （保存到持久表）

DataFrames 也可以使用 saveAsTable 命令作为 persistent tables （持久表）保存到 Hive metastore 中. 请注意, existing Hive deployment （现有的 Hive 部署）不需要使用此功能. Spark 将为您创建默认的 local Hive metastore （本地 Hive metastore）（使用 Derby ）. 与 createOrReplaceTempView 命令不同, saveAsTable 将 materialize （实现） DataFrame 的内容, 并创建一个指向 Hive metastore 中数据的指针. 即使您的 Spark 程序重新启动, Persistent tables （持久性表）仍然存在, 因为您保持与同一个 metastore 的连接. 可以通过使用表的名称在 SparkSession上调用 table 方法来创建 persistent tabl （持久表）的 DataFrame .

对于 file-based （基于文件）的 data source （数据源）, 例如 text, parquet, json等, 您可以通过 path 选项指定 custom table path （自定义表路径）, 例如 df.write.option("path", "/some/path").saveAsTable("t") . 当表被 dropped （删除）时, custom table path （自定义表路径）将不会被删除, 并且表数据仍然存在. 如果未指定自定义表路径, Spark 将把数据写入 warehouse directory （仓库目录）下的默认表路径. 当表被删除时, 默认的表路径也将被删除.

从 Spark 2.1 开始, persistent datasource tables （持久性数据源表）将 per-partition metadata （每个分区元数据）存储在 Hive metastore 中. 这带来了几个好处:

• 由于 metastore 只能返回查询的必要 partitions （分区）, 因此不再需要将第一个查询上的所有 partitions discovering 到表中.
• Hive DDLs 如 ALTER TABLE PARTITION ... SET LOCATION 现在可用于使用 Datasource API 创建的表.

请注意, 创建 external datasource tables （外部数据源表）（带有 path 选项）的表时, 默认情况下不会收集 partition information （分区信息）. 要 sync （同步） metastore 中的分区信息, 可以调用 MSCK REPAIR TABLE .

Bucketing, Sorting and Partitioning （分桶, 排序和分区）

对于 file-based data source （基于文件的数据源）, 也可以对 output （输出）进行 bucket 和 sort 或者 partition . Bucketing 和 sorting 仅适用于 persistent tables :

• Scala
• Java
• Python
• Sql

peopleDF.write.bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed")

在使用 Dataset API 时, partitioning 可以同时与 save 和 saveAsTable 一起使用.

• Scala
• Java
• Python
• Sql

usersDF.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet")

可以为 single table （单个表）使用 partitioning 和 bucketing:

• Scala
• Java
• Python
• Sql

peopleDF

  .write

  .partitionBy("favorite_color")

  .bucketBy(42, "name")

  .saveAsTable("people_partitioned_bucketed")

partitionBy 创建一个 directory structure （目录结构）, 如 Partition Discovery 部分所述. 因此, 对 cardinality （基数）较高的 columns 的适用性有限. 相反, bucketBy 可以在固定数量的 buckets 中分配数据, 并且可以在 a number of unique values is unbounded （多个唯一值无界时）使用数据.

Parquet Files

Parquet 是许多其他数据处理系统支持的 columnar format （柱状格式）. Spark SQL 支持读写 Parquet 文件, 可自动保留 schema of the original data （原始数据的模式）. 当编写 Parquet 文件时, 出于兼容性原因, 所有 columns 都将自动转换为可空.

Loading Data Programmatically （以编程的方式加载数据）

使用上面例子中的数据:

• Scala
• Java
• Python
• R
• Sql

// Encoders for most common types are automatically provided by importing spark.implicits._

import spark.implicits._

val peopleDF = spark.read.json("examples/src/main/resources/people.json")

// DataFrames can be saved as Parquet files, maintaining the schema information

peopleDF.write.parquet("people.parquet")

// Read in the parquet file created above

// Parquet files are self-describing so the schema is preserved

// The result of loading a Parquet file is also a DataFrame

val parquetFileDF = spark.read.parquet("people.parquet")

// Parquet files can also be used to create a temporary view and then used in SQL statements

parquetFileDF.createOrReplaceTempView("parquetFile")

val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")

namesDF.map(attributes => "Name: " + attributes(0)).show()

// +------------+

// |       value|

// +------------+

// |Name: Justin|

// +------------+

Partition Discovery （分区发现）

Table partitioning （表分区）是在像 Hive 这样的系统中使用的常见的优化方法. 在 partitioned table （分区表）中, 数据通常存储在不同的目录中, partitioning column values encoded （分区列值编码）在每个 partition directory （分区目录）的路径中. Parquet data source （Parquet 数据源）现在可以自动 discover （发现）和 infer （推断）分区信息. 例如, 我们可以使用以下 directory structure （目录结构）将所有以前使用的 population data （人口数据）存储到 partitioned table （分区表）中, 其中有两个额外的列 gender 和 country 作为 partitioning columns （分区列）:

path

└── to

    └── table

        ├── gender=male

        │   ├── ...

        │   │

        │   ├── country=US

        │   │   └── data.parquet

        │   ├── country=CN

        │   │   └── data.parquet

        │   └── ...

        └── gender=female

            ├── ...

            │

            ├── country=US

            │   └── data.parquet

            ├── country=CN

            │   └── data.parquet

            └── ...

通过将 path/to/table 传递给 SparkSession.read.parquet 或 SparkSession.read.load , Spark SQL 将自动从路径中提取 partitioning information （分区信息）. 现在返回的 DataFrame 的 schema （模式）变成:

root

|-- name: string (nullable = true)

|-- age: long (nullable = true)

|-- gender: string (nullable = true)

|-- country: string (nullable = true)

请注意, 会自动 inferred （推断） partitioning columns （分区列）的 data types （数据类型）.目前, 支持 numeric data types （数字数据类型）和 string type （字符串类型）.有些用户可能不想自动推断 partitioning columns （分区列）的数据类型.对于这些用例, automatic type inference （自动类型推断）可以由 spark.sql.sources.partitionColumnTypeInference.enabled 配置, 默认为 true .当禁用 type inference （类型推断）时, string type （字符串类型）将用于 partitioning columns （分区列）.

从 Spark 1.6.0 开始, 默认情况下, partition discovery （分区发现）只能找到给定路径下的 partitions （分区）.对于上述示例, 如果用户将 path/to/table/gender=male 传递给 SparkSession.read.parquet 或 SparkSession.read.load , 则 gender 将不被视为 partitioning column （分区列）.如果用户需要指定 partition discovery （分区发现）应该开始的基本路径, 则可以在数据源选项中设置 basePath.例如, 当 path/to/table/gender=male 是数据的路径并且用户将 basePath 设置为 path/to/table/, gender 将是一个 partitioning column （分区列）.

Schema Merging （模式合并）

像 ProtocolBuffer , Avro 和 Thrift 一样, Parquet 也支持 schema evolution （模式演进）. 用户可以从一个 simple schema （简单的架构）开始, 并根据需要逐渐向 schema 添加更多的 columns （列）. 以这种方式, 用户可能会使用不同但相互兼容的 schemas 的 multiple Parquet files （多个 Parquet 文件）. Parquet data source （Parquet 数据源）现在能够自动检测这种情况并 merge （合并）所有这些文件的 schemas .

由于 schema merging （模式合并）是一个 expensive operation （相对昂贵的操作）, 并且在大多数情况下不是必需的, 所以默认情况下从 1.5.0 开始. 你可以按照如下的方式启用它:

1. 读取 Parquet 文件时, 将 data source option （数据源选项） mergeSchema 设置为 true （如下面的例子所示）, 或
2. 将 global SQL option （全局 SQL 选项） spark.sql.parquet.mergeSchema 设置为 true .

• Scala
• Java
• Python
• R

// This is used to implicitly convert an RDD to a DataFrame.

import spark.implicits._

// Create a simple DataFrame, store into a partition directory

val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")

squaresDF.write.parquet("data/test_table/key=1")

// Create another DataFrame in a new partition directory,

// adding a new column and dropping an existing column

val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")

cubesDF.write.parquet("data/test_table/key=2")

// Read the partitioned table

val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")

mergedDF.printSchema()

// The final schema consists of all 3 columns in the Parquet files together

// with the partitioning column appeared in the partition directory paths

// root

//  |-- value: int (nullable = true)

//  |-- square: int (nullable = true)

//  |-- cube: int (nullable = true)

//  |-- key: int (nullable = true)

Hive metastore Parquet table conversion （Hive metastore Parquet table 转换）

当读取和写入 Hive metastore Parquet 表时, Spark SQL 将尝试使用自己的 Parquet support （Parquet 支持）, 而不是 Hive SerDe 来获得更好的性能. 此 behavior （行为）由 spark.sql.hive.convertMetastoreParquet 配置控制, 默认情况下 turned on （打开）.

Hive/Parquet Schema Reconciliation

从 table schema processing （表格模式处理）的角度来说, Hive 和 Parquet 之间有两个关键的区别.

1. Hive 不区分大小写, 而 Parquet 不是
2. Hive 认为所有 columns （列）都可以为空, 而 Parquet 中的可空性是 significant （重要）的.

由于这个原因, 当将 Hive metastore Parquet 表转换为 Spark SQL Parquet 表时, 我们必须调整 metastore schema 与 Parquet schema. reconciliation 规则是:

1. 在两个 schema 中具有 same name （相同名称）的 Fields （字段）必须具有 same data type （相同的数据类型）, 而不管 nullability （可空性）. reconciled field 应具有 Parquet 的数据类型, 以便 nullability （可空性）得到尊重.

2. reconciled schema （调和模式）正好包含 Hive metastore schema 中定义的那些字段.

   • 只出现在 Parquet schema 中的任何字段将被 dropped （删除）在 reconciled schema 中.
   • 仅在 Hive metastore schema 中出现的任何字段在 reconciled schema 中作为 nullable field （可空字段）添加.

Metadata Refreshing （元数据刷新）

Spark SQL 缓存 Parquet metadata 以获得更好的性能. 当启用 Hive metastore Parquet table conversion （转换）时, 这些 converted tables （转换表）的 metadata （元数据）也被 cached （缓存）. 如果这些表由 Hive 或其他外部工具更新, 则需要手动刷新以确保 consistent metadata （一致的元数据）.

• Scala
• Java
• Python
• Sql

// spark is an existing SparkSession

spark.catalog.refreshTable("my_table")

Configuration （配置）

可以使用 SparkSession 上的 setConf 方法或使用 SQL 运行 SET key = value 命令来完成 Parquet 的配置.

JSON Datasets （JSON 数据集）

• Scala
• Java
• Python
• R
• Sql

// Primitive types (Int, String, etc) and Product types (case classes) encoders are

// supported by importing this when creating a Dataset.

import spark.implicits._

// A JSON dataset is pointed to by path.

// The path can be either a single text file or a directory storing text files

val path = "examples/src/main/resources/people.json"

val peopleDF = spark.read.json(path)

// The inferred schema can be visualized using the printSchema() method

peopleDF.printSchema()

// root

//  |-- age: long (nullable = true)

//  |-- name: string (nullable = true)

// Creates a temporary view using the DataFrame

peopleDF.createOrReplaceTempView("people")

// SQL statements can be run by using the sql methods provided by spark

val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")

teenagerNamesDF.show()

// +------+

// |  name|

// +------+

// |Justin|

// +------+

// Alternatively, a DataFrame can be created for a JSON dataset represented by

// a Dataset[String] storing one JSON object per string

val otherPeopleDataset = spark.createDataset(

  """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)

val otherPeople = spark.read.json(otherPeopleDataset)

otherPeople.show()

// +---------------+----+

// |        address|name|

// +---------------+----+

// |[Columbus,Ohio]| Yin|

// +---------------+----+

Hive 表

Spark SQL 还支持读取和写入存储在 Apache Hive 中的数据。 但是，由于 Hive 具有大量依赖关系，因此这些依赖关系不包含在默认 Spark 分发中。 如果在类路径中找到 Hive 依赖项，Spark 将自动加载它们。 请注意，这些 Hive 依赖关系也必须存在于所有工作节点上，因为它们将需要访问 Hive 序列化和反序列化库 (SerDes)，以访问存储在 Hive 中的数据。

通过将 hive-site.xml, core-site.xml（用于安全配置）和 hdfs-site.xml （用于 HDFS 配置）文件放在 conf/ 中来完成配置。

当使用 Hive 时，必须用 Hive 支持实例化 SparkSession，包括连接到持续的 Hive 转移，支持 Hive serdes 和 Hive 用户定义的功能。 没有现有 Hive 部署的用户仍然可以启用 Hive 支持。 当 hive-site.xml 未配置时，上下文会自动在当前目录中创建 metastore_db，并创建由 spark.sql.warehouse.dir 配置的目录，该目录默认为Spark应用程序当前目录中的 spark-warehouse 目录 开始了 请注意，自从2.0.0以来，hive-site.xml 中的 hive.metastore.warehouse.dir 属性已被弃用。 而是使用 spark.sql.warehouse.dir 来指定仓库中数据库的默认位置。 您可能需要向启动 Spark 应用程序的用户授予写权限。å

• Scala
• Java
• Python
• R

import java.io.File

import org.apache.spark.sql.Row

import org.apache.spark.sql.SparkSession

case class Record(key: Int, value: String)

// warehouseLocation points to the default location for managed databases and tables

val warehouseLocation = new File("spark-warehouse").getAbsolutePath

val spark = SparkSession

  .builder()

  .appName("Spark Hive Example")

  .config("spark.sql.warehouse.dir", warehouseLocation)

  .enableHiveSupport()

  .getOrCreate()

import spark.implicits._

import spark.sql

sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive")

sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")

// Queries are expressed in HiveQL

sql("SELECT * FROM src").show()

// +---+-------+

// |key|  value|

// +---+-------+

// |238|val_238|

// | 86| val_86|

// |311|val_311|

// ...

// Aggregation queries are also supported.

sql("SELECT COUNT(*) FROM src").show()

// +--------+

// |count(1)|

// +--------+

// |    500 |

// +--------+

// The results of SQL queries are themselves DataFrames and support all normal functions.

val sqlDF = sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")

// The items in DataFrames are of type Row, which allows you to access each column by ordinal.

val stringsDS = sqlDF.map {

  case Row(key: Int, value: String) => s"Key: $key, Value: $value"

}

stringsDS.show()

// +--------------------+

// |               value|

// +--------------------+

// |Key: 0, Value: val_0|

// |Key: 0, Value: val_0|

// |Key: 0, Value: val_0|

// ...

// You can also use DataFrames to create temporary views within a SparkSession.

val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, s"val_$i")))

recordsDF.createOrReplaceTempView("records")

// Queries can then join DataFrame data with data stored in Hive.

sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()

// +---+------+---+------+

// |key| value|key| value|

// +---+------+---+------+

// |  2| val_2|  2| val_2|

// |  4| val_4|  4| val_4|

// |  5| val_5|  5| val_5|

// ...

指定 Hive 表的存储格式

创建 Hive 表时，需要定义如何 从/向 文件系统 read/write 数据，即 “输入格式” 和 “输出格式”。 您还需要定义该表如何将数据反序列化为行，或将行序列化为数据，即 “serde”。 以下选项可用于指定存储格式 (“serde”, “input format”, “output format”)，例如，CREATE TABLE src(id int) USING hive OPTIONS(fileFormat 'parquet')。 默认情况下，我们将以纯文本形式读取表格文件。 请注意，Hive 存储处理程序在创建表时不受支持，您可以使用 Hive 端的存储处理程序创建一个表，并使用 Spark SQL 来读取它。

使用 OPTIONS 定义的所有其他属性将被视为 Hive serde 属性。

与不同版本的 Hive Metastore 进行交互

Spark SQL 的 Hive 支持的最重要的部分之一是与 Hive metastore 进行交互，这使得 Spark SQL 能够访问 Hive 表的元数据。 从 Spark 1.4.0 开始，使用 Spark SQL 的单一二进制构建可以使用下面所述的配置来查询不同版本的 Hive 转移。 请注意，独立于用于与转移点通信的 Hive 版本，内部 Spark SQL 将针对 Hive 1.2.1 进行编译，并使用这些类进行内部执行（serdes，UDF，UDAF等）。

以下选项可用于配置用于检索元数据的 Hive 版本：

JDBC 连接其它数据库

Spark SQL 还包括可以使用 JDBC 从其他数据库读取数据的数据源。此功能应优于使用 JdbcRDD。 这是因为结果作为 DataFrame 返回，并且可以轻松地在 Spark SQL 中处理或与其他数据源连接。 JDBC 数据源也更容易从 Java 或 Python 使用，因为它不需要用户提供 ClassTag。（请注意，这不同于 Spark SQL JDBC 服务器，允许其他应用程序使用 Spark SQL 运行查询）。

要开始使用，您需要在 Spark 类路径中包含特定数据库的 JDBC driver 程序。 例如，要从 Spark Shell 连接到 postgres，您将运行以下命令:

bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar

可以使用 Data Sources API 将来自远程数据库的表作为 DataFrame 或 Spark SQL 临时视图进行加载。 用户可以在数据源选项中指定 JDBC 连接属性。用户 和 密码通常作为登录数据源的连接属性提供。 除了连接属性外，Spark 还支持以下不区分大小写的选项:

• Scala
• Java
• Python
• R
• Sql

// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods

// Loading data from a JDBC source

val jdbcDF = spark.read

  .format("jdbc")

  .option("url", "jdbc:postgresql:dbserver")

  .option("dbtable", "schema.tablename")

  .option("user", "username")

  .option("password", "password")

  .load()

val connectionProperties = new Properties()

connectionProperties.put("user", "username")

connectionProperties.put("password", "password")

val jdbcDF2 = spark.read

  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

// Saving data to a JDBC source

jdbcDF.write

  .format("jdbc")

  .option("url", "jdbc:postgresql:dbserver")

  .option("dbtable", "schema.tablename")

  .option("user", "username")

  .option("password", "password")

  .save()

jdbcDF2.write

  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

// Specifying create table column data types on write

jdbcDF.write

  .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")

  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

故障排除

• JDBC driver 程序类必须对客户端会话和所有执行程序上的原始类加载器可见。 这是因为 Java 的 DriverManager 类执行安全检查，导致它忽略原始类加载器不可见的所有 driver 程序，当打开连接时。一个方便的方法是修改所有工作节点上的compute_classpath.sh 以包含您的 driver 程序 JAR。
• 一些数据库，例如 H2，将所有名称转换为大写。 您需要使用大写字母来引用 Spark SQL 中的这些名称。

性能调优

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

在内存中缓存数据

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

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

其他配置选项

以下选项也可用于调整查询执行的性能。这些选项可能会在将来的版本中被废弃，因为更多的优化是自动执行的。

分布式 SQL 引擎

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

运行 Thrift JDBC/ODBC 服务器

这里实现的 Thrift JDBC/ODBC 服务器对应于 Hive 1.2 中的 HiveServer2。 您可以使用 Spark 或 Hive 1.2.1 附带的直线脚本测试 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> \

  ...

or system properties:

./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

使用 beeline 方式连接到 JDBC/ODBC 服务器:

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

Beeline 将要求您输入用户名和密码。 在非安全模式下，只需输入机器上的用户名和空白密码即可。 对于安全模式，请按照 beeline 文档 中的说明进行操作。

配置Hive是通过将 hive-site.xml, core-site.xml 和 hdfs-site.xml 文件放在 conf/ 中完成的。

您也可以使用 Hive 附带的 beeline 脚本。

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

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>

运行 Spark SQL CLI

Spark SQL CLI 是在本地模式下运行 Hive 转移服务并执行从命令行输入的查询的方便工具。 请注意，Spark SQL CLI 不能与 Thrift JDBC 服务器通信。

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

./bin/spark-sql

配置 Hive 是通过将 hive-site.xml, core-site.xml 和 hdfs-site.xml 文件放在 conf/ 中完成的。 您可以运行 ./bin/spark-sql --help 获取所有可用选项的完整列表。

迁移指南

从 Spark SQL 2.1 升级到 2.2

• Spark 2.1.1 介绍了一个新的配置 key: spark.sql.hive.caseSensitiveInferenceMode. 它的默认设置是 NEVER_INFER, 其行为与 2.1.0 保持一致. 但是，Spark 2.2.0 将此设置的默认值更改为 “INFER_AND_SAVE”，以恢复与底层文件 schema（模式）具有大小写混合的列名称的 Hive metastore 表的兼容性。使用 INFER_AND_SAVE 配置的 value, 在第一次访问 Spark 将对其尚未保存推测 schema（模式）的任何 Hive metastore 表执行 schema inference（模式推断）. 请注意，对于具有数千个 partitions（分区）的表，模式推断可能是非常耗时的操作。如果不兼容大小写混合的列名，您可以安全地将spark.sql.hive.caseSensitiveInferenceMode 设置为 NEVER_INFER，以避免模式推断的初始开销。请注意，使用新的默认INFER_AND_SAVE 设置，模式推理的结果被保存为 metastore key 以供将来使用。因此，初始模式推断仅发生在表的第一次访问。

从 Spark SQL 2.0 升级到 2.1

• Datasource tables（数据源表）现在存储了 Hive metastore 中的 partition metadata（分区元数据）. 这意味着诸如 ALTER TABLE PARTITION ... SET LOCATION 这样的 Hive DDLs 现在使用 Datasource API 可用于创建 tables（表）.
  • 遗留的数据源表可以通过 MSCK REPAIR TABLE 命令迁移到这种格式。建议迁移遗留表利用 Hive DDL 的支持和提供的计划性能。
  • 要确定表是否已迁移，当在表上发出 DESCRIBE FORMATTED 命令时请查找 PartitionProvider: Catalog 属性.
• Datasource tables（数据源表）的 INSERT OVERWRITE TABLE ... PARTITION ... 行为的更改。
  • 在以前的 Spark 版本中，INSERT OVERWRITE 覆盖了整个 Datasource table，即使给出一个指定的 partition. 现在只有匹配规范的 partition 被覆盖。
  • 请注意，这仍然与 Hive 表的行为不同，Hive 表仅覆盖与新插入数据重叠的分区。

从 Spark SQL 1.6 升级到 2.0

• SparkSession 现在是 Spark 新的切入点, 它替代了老的 SQLContext 和 HiveContext。注意 : 为了向下兼容，老的 SQLContext 和 HiveContext 仍然保留。可以从 SparkSession 获取一个新的 catalog 接口 — 现有的访问数据库和表的 API，如 listTables，createExternalTable，dropTempView，cacheTable 都被移到该接口。

• Dataset API 和 DataFrame API 进行了统一。在 Scala 中，DataFrame 变成了 Dataset[Row] 类型的一个别名，而 Java API 使用者必须将 DataFrame 替换成 Dataset<Row>。Dataset 类既提供了强类型转换操作（如 map，filter 以及 groupByKey）也提供了非强类型转换操作（如 select 和 groupBy）。由于编译期的类型安全不是 Python 和 R 语言的一个特性，Dataset 的概念并不适用于这些语言的 API。相反，DataFrame仍然是最基本的编程抽象, 就类似于这些语言中单节点 data frame 的概念。

• Dataset 和 DataFrame API 中 unionAll 已经过时并且由 union 替代。
• Dataset 和 DataFrame API 中 explode 已经过时，作为选择，可以结合 select 或 flatMap 使用 functions.explode() 。

• Dataset 和 DataFrame API 中 registerTempTable 已经过时并且由 createOrReplaceTempView 替代。

• 对 Hive tables CREATE TABLE ... LOCATION 行为的更改.
  • 从 Spark 2.0 开始，CREATE TABLE ... LOCATION 与 CREATE EXTERNAL TABLE ... LOCATION 是相同的，以防止意外丢弃用户提供的 locations（位置）中的现有数据。这意味着，在用户指定位置的 Spark SQL 中创建的 Hive 表始终是 Hive 外部表。删除外部表将不会删除数据。 用户不能指定 Hive managed tables（管理表）的位置. 请注意，这与Hive行为不同。
  • 因此，这些表上的 “DROP TABLE” 语句不会删除数据。

从 Spark SQL 1.5 升级到 1.6

• 从 Spark 1.6 开始，默认情况下服务器在多 session（会话）模式下运行。这意味着每个 JDBC/ODBC 连接拥有一份自己的 SQL 配置和临时函数注册。缓存表仍在并共享。如果您希望以旧的单会话模式运行 Thrift server，请设置选项 spark.sql.hive.thriftServer.singleSession 为true。您既可以将此选项添加到 spark-defaults.conf，或者通过 --conf 将它传递给 start-thriftserver.sh。

   ./sbin/start-thriftserver.sh \

     --conf spark.sql.hive.thriftServer.singleSession=true \

     ...

   

• 从 1.6.1 开始，在 sparkR 中 withColumn 方法支持添加一个新列或更换 DataFrame 同名的现有列。

• 从 Spark 1.6 开始，LongType 强制转换为 TimestampType 期望是秒，而不是微秒。这种更改是为了匹配 Hive 1.2 的行为，以便从 numeric（数值）类型进行更一致的类型转换到 TimestampType。更多详情请参阅 SPARK-11724 。

从 Spark SQL 1.4 升级到 1.5

• 使用手动管理的内存优化执行，现在是默认启用的，以及代码生成表达式求值。这些功能既可以通过设置 spark.sql.tungsten.enabled 为 false 来禁止使用。
• Parquet 的模式合并默认情况下不再启用。它可以通过设置 spark.sql.parquet.mergeSchema 到 true 以重新启用。
• 字符串在 Python 列的 columns（列）现在支持使用点（.）来限定列或访问嵌套值。例如 df['table.column.nestedField']。但是，这意味着如果你的列名中包含任何圆点，你现在必须避免使用反引号（如 table.column.with.dots.nested）。
• 在内存中的列存储分区修剪默认是开启的。它可以通过设置 spark.sql.inMemoryColumnarStorage.partitionPruning 为 false 来禁用。
• 无限精度的小数列不再支持，而不是 Spark SQL 最大精度为 38 。当从 BigDecimal 对象推断模式时，现在使用（38，18）。在 DDL 没有指定精度时，则默认保留 Decimal(10, 0)。
• 时间戳现在存储在 1 微秒的精度，而不是 1 纳秒的。
• 在 sql 语句中，floating point（浮点数）现在解析为 decimal。HiveQL 解析保持不变。
• SQL / DataFrame 函数的规范名称现在是小写（例如 sum vs SUM）。
• JSON 数据源不会自动加载由其他应用程序（未通过 Spark SQL 插入到数据集的文件）创建的新文件。对于 JSON 持久表（即表的元数据存储在 Hive Metastore），用户可以使用 REFRESH TABLE SQL 命令或 HiveContext 的 refreshTable 方法，把那些新文件列入到表中。对于代表一个 JSON dataset 的 DataFrame，用户需要重新创建 DataFrame，同时 DataFrame 中将包括新的文件。
• PySpark 中 DataFrame 的 withColumn 方法支持添加新的列或替换现有的同名列。

从 Spark SQL 1.3 升级到 1.4

DataFrame data reader/writer interface

基于用户反馈，我们创建了一个新的更流畅的 API，用于读取 (SQLContext.read) 中的数据并写入数据 (DataFrame.write), 并且旧的 API 将过时（例如，SQLContext.parquetFile, SQLContext.jsonFile）.

针对 SQLContext.read ( Scala, Java, Python ) 和 DataFrame.write ( Scala, Java, Python ) 的更多细节，请看 API 文档.

DataFrame.groupBy 保留 grouping columns（分组的列）

根据用户的反馈， 我们更改了 DataFrame.groupBy().agg() 的默认行为以保留 DataFrame 结果中的 grouping columns（分组列）. 为了在 1.3 中保持该行为，请设置 spark.sql.retainGroupColumns 为 false.

• Scala
• Java
• Python

// In 1.3.x, in order for the grouping column "department" to show up,

// it must be included explicitly as part of the agg function call.

df.groupBy("department").agg($"department", max("age"), sum("expense"))

// In 1.4+, grouping column "department" is included automatically.

df.groupBy("department").agg(max("age"), sum("expense"))

// Revert to 1.3 behavior (not retaining grouping column) by:

sqlContext.setConf("spark.sql.retainGroupColumns", "false")

DataFrame.withColumn 上的行为更改

之前 1.4 版本中，DataFrame.withColumn() 只支持添加列。该列将始终在 DateFrame 结果中被加入作为新的列，即使现有的列可能存在相同的名称。从 1.4 版本开始，DataFrame.withColumn() 支持添加与所有现有列的名称不同的列或替换现有的同名列。

请注意，这一变化仅适用于 Scala API，并不适用于 PySpark 和 SparkR。

从 Spark SQL 1.0-1.2 升级到 1.3

在 Spark 1.3 中，我们从 Spark SQL 中删除了 “Alpha” 的标签，作为一部分已经清理过的可用的 API 。从 Spark 1.3 版本以上，Spark SQL 将提供在 1.X 系列的其他版本的二进制兼容性。这种兼容性保证不包括被明确标记为不稳定的（即 DeveloperApi 类或 Experimental） API。

重命名 DataFrame 的 SchemaRDD

升级到 Spark SQL 1.3 版本时，用户会发现最大的变化是，SchemaRDD 已更名为 DataFrame。这主要是因为 DataFrames 不再从 RDD 直接继承，而是由 RDDS 自己来实现这些功能。DataFrames 仍然可以通过调用 .rdd 方法转换为 RDDS 。

在 Scala 中，有一个从 SchemaRDD 到 DataFrame 类型别名，可以为一些情况提供源代码兼容性。它仍然建议用户更新他们的代码以使用 DataFrame来代替。Java 和 Python 用户需要更新他们的代码。

Java 和 Scala APIs 的统一

此前 Spark 1.3 有单独的Java兼容类（JavaSQLContext 和 JavaSchemaRDD），借鉴于 Scala API。在 Spark 1.3 中，Java API 和 Scala API 已经统一。两种语言的用户可以使用 SQLContext 和 DataFrame。一般来说论文类尝试使用两种语言的共有类型（如 Array 替代了一些特定集合）。在某些情况下不通用的类型情况下，（例如，passing in closures 或 Maps）使用函数重载代替。

此外，该 Java 的特定类型的 API 已被删除。Scala 和 Java 的用户可以使用存在于 org.apache.spark.sql.types 类来描述编程模式。

隔离隐式转换和删除 dsl 包（仅Scala）

许多 Spark 1.3 版本以前的代码示例都以 import sqlContext._ 开始，这提供了从 sqlContext 范围的所有功能。在 Spark 1.3 中，我们移除了从 RDDs 到 DateFrame 再到 SQLContext 内部对象的隐式转换。用户现在应该写成 import sqlContext.implicits._.

此外，隐式转换现在只能使用方法 toDF 来增加由 Product（即 case classes or tuples）构成的 RDD，而不是自动应用。

当使用 DSL 内部的函数时（现在使用 DataFrame API 来替换）, 用户习惯导入 org.apache.spark.sql.catalyst.dsl. 相反，应该使用公共的 dataframe 函数 API: import org.apache.spark.sql.functions._.

针对 DataType 删除在 org.apache.spark.sql 包中的一些类型别名（仅限于 Scala）

Spark 1.3 移除存在于基本 SQL 包的 DataType 类型别名。开发人员应改为导入类 org.apache.spark.sql.types。

UDF 注册迁移到 sqlContext.udf 中 (Java & Scala)

用于注册 UDF 的函数，不管是 DataFrame DSL 还是 SQL 中用到的，都被迁移到 SQLContext 中的 udf 对象中。

• Scala
• Java

sqlContext.udf.register("strLen", (s: String) => s.length())

Python UDF 注册保持不变。

Python DataTypes 不再是 Singletons（单例的）

在 Python 中使用 DataTypes 时，你需要先构造它们（如：StringType()），而不是引用一个单例对象。

与 Apache Hive 的兼容

Spark SQL 在设计时就考虑到了和 Hive metastore，SerDes 以及 UDF 之间的兼容性。目前 Hive SerDes 和 UDF 都是基于 Hive 1.2.1 版本，并且Spark SQL 可以连接到不同版本的Hive metastore（从 0.12.0 到 1.2.1，可以参考 与不同版本的 Hive Metastore 交互）

在现有的 Hive Warehouses 中部署

Spark SQL Thrift JDBC server 采用了开箱即用的设计以兼容已有的 Hive 安装版本。你不需要修改现有的 Hive Metastore , 或者改变数据的位置和表的分区。

所支持的 Hive 特性

Spark SQL 支持绝大部分的 Hive 功能，如:

• Hive query（查询）语句, 包括:
  • SELECT
  • GROUP BY
  • ORDER BY
  • CLUSTER BY
  • SORT BY
• 所有 Hive 操作, 包括:
  • 关系运算符 (=, ⇔, ==, <>, <, >, >=, <=, 等等)
  • 算术运算符 (+, -, *, /, %, 等等)
  • 逻辑运算符 (AND, &&, OR, ||, 等等)
  • 复杂类型的构造
  • 数学函数 (sign, ln, cos, 等等)
  • String 函数 (instr, length, printf, 等等)
• 用户定义函数 (UDF)
• 用户定义聚合函数 (UDAF)
• 用户定义 serialization formats (SerDes)
• 窗口函数
• Joins
  • JOIN
  • {LEFT|RIGHT|FULL} OUTER JOIN
  • LEFT SEMI JOIN
  • CROSS JOIN
• Unions
• Sub-queries（子查询）
  • SELECT col FROM ( SELECT a + b AS col from t1) t2
• Sampling
• Explain
• Partitioned tables including dynamic partition insertion
• View
• 所有的 Hive DDL 函数, 包括:
  • CREATE TABLE
  • CREATE TABLE AS SELECT
  • ALTER TABLE
• 大部分的 Hive Data types（数据类型）, 包括:
  • TINYINT
  • SMALLINT
  • INT
  • BIGINT
  • BOOLEAN
  • FLOAT
  • DOUBLE
  • STRING
  • BINARY
  • TIMESTAMP
  • DATE
  • ARRAY<>
  • MAP<>
  • STRUCT<>

未支持的 Hive 函数

以下是目前还不支持的 Hive 函数列表。在 Hive 部署中这些功能大部分都用不到。

主要的 Hive 功能

• Tables 使用 buckets 的 Tables: bucket 是 Hive table partition 中的 hash partitioning. Spark SQL 还不支持 buckets.

Esoteric Hive 功能

• UNION 类型
• Unique join
• Column 统计信息的收集: Spark SQL does not piggyback scans to collect column statistics at the moment and only supports populating the sizeInBytes field of the hive metastore.

Hive Input/Output Formats

• File format for CLI: For results showing back to the CLI, Spark SQL only supports TextOutputFormat.
• Hadoop archive

Hive 优化

有少数 Hive 优化还没有包含在 Spark 中。其中一些（比如 indexes 索引）由于 Spark SQL 的这种内存计算模型而显得不那么重要。另外一些在 Spark SQL 未来的版本中会持续跟踪。

• Block 级别的 bitmap indexes 和虚拟 columns (用于构建 indexes)
• 自动为 join 和 groupBy 计算 reducer 个数 : 目前在 Spark SQL 中, 你需要使用 “SET spark.sql.shuffle.partitions=[num_tasks];” 来控制 post-shuffle 的并行度.
• 仅 Meta-data 的 query: 对于只使用 metadata 就能回答的查询，Spark SQL 仍然会启动计算结果的任务.
• Skew data flag: Spark SQL 不遵循 Hive 中 skew 数据的标记.
• STREAMTABLE hint in join: Spark SQL 不遵循 STREAMTABLE hint.
• 对于查询结果合并多个小文件: 如果输出的结果包括多个小文件, Hive 可以可选的合并小文件到一些大文件中去，以避免溢出 HDFS metadata. Spark SQL 还不支持这样.

参考

数据类型

Spark SQL 和 DataFrames 支持下面的数据类型:

• Numeric types
  • ByteType: Represents 1-byte signed integer numbers. The range of numbers is from -128 to 127.
  • ShortType: Represents 2-byte signed integer numbers. The range of numbers is from -32768 to 32767.
  • IntegerType: Represents 4-byte signed integer numbers. The range of numbers is from -2147483648 to 2147483647.
  • LongType: Represents 8-byte signed integer numbers. The range of numbers is from -9223372036854775808 to 9223372036854775807.
  • FloatType: Represents 4-byte single-precision floating point numbers.
  • DoubleType: Represents 8-byte double-precision floating point numbers.
  • DecimalType: Represents arbitrary-precision signed decimal numbers. Backed internally by java.math.BigDecimal. A BigDecimalconsists of an arbitrary precision integer unscaled value and a 32-bit integer scale.
• String type
  • StringType: Represents character string values.
• Binary type
  • BinaryType: Represents byte sequence values.
• Boolean type
  • BooleanType: Represents boolean values.
• Datetime type
  • TimestampType: Represents values comprising values of fields year, month, day, hour, minute, and second.
  • DateType: Represents values comprising values of fields year, month, day.
• Complex types
  • ArrayType(elementType, containsNull): Represents values comprising a sequence of elements with the type of elementType. containsNull is used to indicate if elements in a ArrayType value can have null values.
  • MapType(keyType, valueType, valueContainsNull): Represents values comprising a set of key-value pairs. The data type of keys are described by keyType and the data type of values are described by valueType. For a MapType value, keys are not allowed to have nullvalues. valueContainsNull is used to indicate if values of a MapType value can have null values.
  • StructType(fields): Represents values with the structure described by a sequence of StructFields (fields).
    • StructField(name, dataType, nullable): Represents a field in a StructType. The name of a field is indicated by name. The data type of a field is indicated by dataType. nullable is used to indicate if values of this fields can have null values.

• Scala
• Java
• Python
• R

import org.apache.spark.sql.types._

NaN Semantics

当处理一些不符合标准浮点数语义的 float 或 double 类型时，对于 Not-a-Number(NaN) 需要做一些特殊处理. 具体如下:

• NaN = NaN 返回 true.
• 在 aggregations（聚合）操作中，所有的 NaN values 将被分到同一个组中.
• 在 join key 中 NaN 可以当做一个普通的值.
• NaN 值在升序排序中排到最后，比任何其他数值都大.

我们一直在努力

apachecn/spark-doc-zh