Spark SQL 教程翻译快速开始

Posted 2021-08-27 顧棟

文章目录

• Getting Started
• • Starting Point: SparkSession
  • Creating DataFrames
  • Untyped Dataset Operations (aka DataFrame Operations)
  • Running SQL Queries Programmatically
  • Global Temporary View
  • Creating Datasets
  • Interoperating with RDDs
  • • Programmatically Specifying the Schema
  • Aggregations
  • • Untyped User-Defined Aggregate Functions
    • Type-Safe User-Defined Aggregate Functions

Getting Started

Starting Point: SparkSession

Spark 中所有功能的入口点是 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._

在 Spark 存储库中的examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala中找到完整的示例代码。
Spark 2.0 中的 SparkSession 为 Hive 功能提供内置支持，包括使用 HiveQL 编写查询的能力、访问 Hive UDF 以及从 Hive 表读取数据的能力。 要使用这些功能，您不需要拥有现有的 Hive 设置。

Creating DataFrames

使用 SparkSession，应用程序可以从现有 RDD、Hive 表或 Spark 数据源创建数据帧。

例如，以下内容基于 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中找到完整的示例代码。

Untyped Dataset Operations (aka DataFrame Operations)

DataFrames 为 Scala、Java、Python 和 R 中的结构化数据操作提供了一种特定领域的语言。

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

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

// 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 文档。

除了简单的列引用和表达式，Datasets 还拥有丰富的函数库，包括字符串操作、日期算术、常见数学运算等。 完整列表可在 DataFrame 函数参考中找到。

Running SQL Queries Programmatically

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 存储库中的examples/src/main/scala/org/apache/spark/examples/sql/SparkSQLExample.scala中找到完整的示例代码。

Global Temporary View

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

// 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中找到完整的示例代码。

Creating Datasets

数据集类似于 RDD，但是，它们不使用 Java 序列化或 Kryo，而是使用专门的编码器来序列化对象以进行处理或通过网络传输。 虽然编码器和标准序列化都负责将对象转换为字节，但编码器是动态生成的代码，使用的格式允许 Spark 执行许多操作，如过滤、排序和散列，而无需将字节反序列化回对象。

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

Interoperating with RDDs

Spark SQL 支持两种不同的方法将现有的 RDD 转换为数据集。 第一种方法使用反射来推断包含特定类型对象的 RDD 的模式。 当您在编写 Spark 应用程序时已经知道架构时，这种基于反射的方法会产生更简洁的代码并且效果很好。

创建数据集的第二种方法是通过编程接口，该接口允许您构建架构，然后将其应用于现有 RDD。 虽然此方法更加冗长，但它允许您在列及其类型直到运行时才知道时构造数据集。

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

Programmatically Specifying the Schema

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

1. 从原始RDD创建一个行的RDD；
2. 创建由与步骤 1 中创建的 RDD 中的 Rows 结构匹配的 StructType 表示的模式。
3. 通过 SparkSession 提供的 createDataFrame 方法将 schema 应用到 Rows 的 RDD。

For example:

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

内置的 DataFrames 函数提供了常见的聚合，例如 count()、countDistinct()、avg()、max()、min() 等。虽然这些函数是为 DataFrames 设计的，但 Spark SQL 也有类型安全的版本用于 其中一些在 Scala 和 Java 中用于处理强类型数据集。 此外，用户不限于预定义的聚合函数，可以创建自己的聚合函数。

Untyped User-Defined Aggregate Functions

用户必须扩展 UserDefinedAggregateFunction 抽象类来实现自定义的无类型聚合函数。 例如，用户定义的平均值可能如下所示：

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.types._

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

强类型数据集的用户定义聚合围绕聚合器抽象类。 例如，类型安全的用户定义平均值可能如下所示

import org.apache.spark.sql.{Encoder, Encoders, SparkSession}
import org.apache.spark.sql.expressions.Aggregator

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