Spark SQL 编程初级实践详解

介绍

Spark SQL 是 Apache Spark 计算框架下的一种数据处理模块，它提供了类似于 SQL 的语言接口，使得在 Spark 中处理结构化数据变得更加方便和高效。

本文将会详细介绍如何使用 Spark SQL 进行编程，包括数据的加载、SQL 的执行、结果的输出等操作。

数据加载

Spark SQL 支持加载多种数据格式的数据集，比如 CSV、JSON、Parquet 等。其中，CSV 是最常见的数据格式之一，我们在这里以 CSV 为例进行介绍。

import org.apache.spark.sql.SparkSession

val spark: SparkSession = SparkSession.builder
  .appName("load-csv-data")
  .master("local[*]")
  .getOrCreate

val csvPath = "file:///path/to/csv"
val df = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv(csvPath)

首先，我们需要创建一个 SparkSession 对象，用于加载数据集和执行 SQL 查询。进行数据集的加载需要调用 spark.read.csv() 方法，其中，参数 csvPath 表示数据文件的路径。

同时，我们需要在 read 方法中使用 option 方法，以配置读取 CSV 文件的行头（header）、自动类型推断（inferSchema）等参数。例如，option("header", "true") 表示 CSV 文件的第一行是行头；option("inferSchema", "true") 表示 Spark 应该自动推断列的类型。

SQL 执行

加载好数据集后，我们可以通过 SQL 接口对数据进行查询和分析。Spark SQL 提供了多种对 SQL 查询的支持方式，包括 spark.sql() 方法和 DataFrame.createOrReplaceTempView() 方法，下面会一一介绍。

spark.sql() 方法

通过 spark.sql() 方法可以执行一条 SQL 查询，并返回一个 DataFrame 结果集。例如：

val result = spark.sql("SELECT COUNT(*) FROM myTable WHERE age > 20")
result.show()

其中，myTable 是我们加载进来的数据集名字，result 是查询结果集。我们可以调用 show() 方法来查看结果。

DataFrame.createOrReplaceTempView() 方法

另一种方法是将 DataFrame 注册为一个临时表，然后通过 SQL 查询这张表。例如：

df.createOrReplaceTempView("people")
val result = spark.sql("SELECT COUNT(*) FROM people WHERE age > 20")
result.show()

其中，我们使用 createOrReplaceTempView() 方法将 DataFrame df 注册为一个名为 people 的临时表，然后基于这张表执行 SQL 查询。

结果输出

查询出的结果可以使用 show() 方法进行输出。如果需要将结果存储到磁盘中，可以使用 write 方法将 DataFrame 保存为 CSV、JSON、Parquet 等格式的文件。

// 输出结果
result.show()

// 存储DataFrame为JSON文件
val path = "file:///path/to/output"
result.write.json(path)

示例

下面给出两个例子，以帮助理解 Spark SQL 的编程实践。

示例 1

我们有一个包含学生信息的数据集，有以下几列数据：name、age、gender、grade，其中，name 和 gender 是字符串类型，age 和 grade 是数值类型。我们需要查询出年龄大于 18 岁的男学生的人数：

val spark: SparkSession = SparkSession.builder
  .appName("count-male-students")
  .master("local[*]")
  .getOrCreate

val csvPath = "file:///path/to/students.csv"
val df = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv(csvPath)

df.createOrReplaceTempView("students")
val result = spark.sql("SELECT COUNT(*) FROM students WHERE gender = 'M' AND age > 18")
result.show()

示例 2

我们有一个包含访问日志的数据集，有以下几列数据：timestamp、uid、url、status_code、ip，其中，timestamp、url 和 ip 是字符串类型，uid 是数值类型，status_code 是 HTTP 状态码。我们需要查询出每个 IP 地址被访问的次数，并且按照访问次数降序排列：

val spark: SparkSession = SparkSession.builder
  .appName("count-ips")
  .master("local[*]")
  .getOrCreate

val csvPath = "file:///path/to/access.log"
val df = spark.read
  .option("header", "false")
  .option("inferSchema", "true")
  .csv(csvPath)

df.createOrReplaceTempView("access_log")
val result = spark.sql("SELECT ip, COUNT(*) AS cnt FROM access_log GROUP BY ip ORDER BY cnt DESC")
result.show()