Spark SQL是Spark提供的分布式SQL查询引擎，通过Spark SQL，我们可以使用SQL语法来查询非关系型数据、结构化数据、CSV文件等。Spark SQL目前支持Hive查询语法和Spark SQL语法，也允许用户进行自定义函数、聚合函数等操作。

Spark学习笔记之Spark SQL的具体使用

简介

Spark SQL是Spark提供的分布式SQL查询引擎，通过Spark SQL，我们可以使用SQL语法来查询非关系型数据、结构化数据、CSV文件等。Spark SQL目前支持Hive查询语法和Spark SQL语法，也允许用户进行自定义函数、聚合函数等操作。

安装

要使用Spark SQL，我们需要先安装Spark。安装方法可以参考Spark官网提供的安装教程。

使用

初始化

我们先通过SparkSession初始化Spark SQL环境：

import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder()
  .appName("Spark SQL example")
  .master("local[*]")
  .getOrCreate()

import spark.implicits._

读取数据

Spark SQL支持多种数据源格式，比如CSV、JSON、文本等。下面我们以读取CSV文件为例：

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

其中，option("header", "true")表示第一行为列名，option("inferSchema", "true")表示自动推断列的数据类型。

执行查询

接下来我们可以使用Spark SQL进行查询。Spark SQL支持Hive查询语法，也支持Spark SQL语法。这里我们以Spark SQL语法为例：

df.select($"col1", $"col2", $"col3".alias("newCol"))
  .filter($"col1" > 10)
  .groupBy($"col2")
  .agg(avg($"col3"), max($"col3"))
  .show()

这个查询语句的意思是：选取“col1”、“col2”、“col3”三列，对“col1”大于10的行进行筛选，按照“col2”进行分组，并计算“col3”的平均值和最大值。

自定义函数

Spark SQL允许用户自定义函数。下面是一个简单的例子：

import org.apache.spark.sql.functions._

val square: Int => Int = x => x * x

spark.udf.register("square", square)

df.selectExpr("col1", "square(col1) as col1_squared", "col2")

这个自定义函数的意思是，将输入的整数进行平方运算。我们可以使用udf.register()方法注册该函数。