一、引言

在上一篇博客中，我们深入学习了Spark Core与RDD。RDD的强大在于其极高的灵活性——它允许我们用函数式编程处理任何类型的数据。但是，随着大数据分析逐渐走向业务化，我们面对的数据80%以上都是结构化的（例如数据库中的表、CSV文件、JSON日志）。

此时，RDD的劣势显现了出来：

1. 缺乏数据结构信息（Schema）：Spark框架本身不知道RDD内部元素的类型结构，无法进行底层优化。

2. 开发门槛高：对于数据分析师来说，写SQL远比写Scala/Python函数要方便。

为了解决这些问题，Spark SQL应运而生。它不仅提供了更友好的编程接口（DataFrame/DataSet），还内置了一个强大的Catalyst 优化器，能将SQL语句自动优化，执行效率甚至远超手写的RDD代码。今天，我们就来彻底掌握Spark SQL。

二、核心概念：DataFrame与Dataset

2.1 什么是DataFrame？

与RDD类似，DataFrame也是一个分布式的数据容器。但不同的是，DataFrame 带有 Schema（表结构）信息。它更像传统数据库中的一张二维表格，不仅有数据，还记录了每一列的名称和类型。

RDD 与 DataFrame 的区别：

• RDD[Person]：Spark 知道里面装的是 Person 对象，但不知道 Person 内部有什么属性。

• DataFrame：Spark 清晰地知道这张表包含name（String类型）和age（Int类型）两列。

有了 Schema，Spark SQL 就可以像关系型数据库那样，进行列裁剪、谓词下推等高级优化。

2.2 两种编程风格

Spark SQL 支持两种编程风格：

1. SQL 风格：将 DataFrame 注册为临时视图，直接书写标准 SQL。

2. DSL 风格（Table API）：调用 DataFrame 提供的 API（如select,where,groupBy）进行链式编程。

三、Spark SQL 编程实战

3.1 环境准备与 DataFrame 创建

在 Spark 2.0 之后，SparkSession取代了SparkContext成为新的入口。

from pyspark.sql import SparkSession if __name__ == '__main__': # 1. 创建 SparkSession spark = SparkSession.builder \ .master("local[*]") \ .appName("SparkSQL_Basic") \ .getOrCreate() # 2. 读取 JSON 文件生成 DataFrame (JSON自带Schema) df = spark.read.json(r"D:\data\students.json") # 3. 打印表结构 (Schema) df.printSchema() # 4. 展示数据 (默认显示20行) df.show() spark.stop()

3.2 SQL 风格与 DSL 风格

结合给定的数据，我们来演示如何用两种方式查询数据。

数据准备 (student.csv)：

id,name,kill,country 1,张三,10,A国 2,李四,13,A国 3,王五,83,B国 4,赵六,32,B国 5,孙七,33,C国

代码演示：

# ----------------- 方式一：SQL 风格 ----------------- # 注册临时视图 df.createOrReplaceTempView("students") # 写 SQL 查询 sql_df = spark.sql(""" SELECT country, COUNT(1) as num, MAX(kill) as max_kill FROM students GROUP BY country """) sql_df.show() # ----------------- 方式二：DSL 风格 ----------------- # 使用 API 链式调用 import pyspark.sql.functions as F dsl_df = df.select("country", "kill") \ .where(df.kill > 10) \ .groupBy("country") \ .agg(F.count("kill").alias("num"), F.max("kill").alias("max_kill")) dsl_df.show()

3.3 复杂实战：窗口函数与行列转换

Spark SQL 完全兼容 Hive 的窗口函数和复杂类型操作。结合a.txt（WordCount）和x2.txt（多列拆分、分组TopN），我们来看 DSL 的强大之处。

案例一：WordCount (爆炸函数)

df_lines = spark.read.text("a.txt") # 使用 split 和 explode 进行行转列（扁平化） word_df = df_lines.select( F.explode(F.split("value", " ")).alias("word") ) # 分组统计 word_count = word_df.groupBy("word").count().orderBy(F.desc("count")) word_count.show()

案例二：分组TopN (窗口函数)
需求：求出每个国家杀敌数排行前三的英雄。

from pyspark.sql.window import Window # 定义窗口：按国家分区，按杀敌数降序 window_spec = Window.partitionBy("country").orderBy(F.desc("kill")) # 添加排名列 df_rank = df.withColumn("rnk", F.row_number().over(window_spec)) # 过滤出排名小于等于3的 top3_df = df_rank.where(df_rank.rnk <= 3).drop("rnk") top3_df.show()

四、RDD 与 DataFrame 的相互转换

在复杂的业务场景中，经常需要结合 RDD 的灵活性（如非结构化清洗）和 DataFrame 的结构化查询优势。

4.1 RDD 转 DataFrame

有两种方式：反射机制和自定义 Schema。

rdd = spark.sparkContext.textFile("person.txt") rdd2 = rdd.map(lambda v: v.split(" ")).map(lambda v: (v[0], int(v[1]))) # 方式一：toDF (直接指定列名) df = rdd2.toDF(["name", "age"]) # 方式二：通过 StructType 自定义 Schema（适合复杂结构） from pyspark.sql.types import StructType, StructField, StringType, IntegerType schema = StructType([ StructField("name", StringType(), True), StructField("age", IntegerType(), True) ]) df2 = spark.createDataFrame(rdd2, schema)

4.2 DataFrame 转 RDD

直接调用.rdd属性。此时返回的 RDD 每一行是一个Row对象，可以通过row['列名']访问。

五、自定义函数（UDF）

当内置函数无法满足需求时，我们可以注册 Python 函数作为 UDF（User Defined Function）。

# 定义一个将 1/2 转为 男/女 的函数 def convert_sex(sex): return "男" if sex == 1 else "女" # 注册 UDF 到 SparkSession spark.udf.register("to_sex", convert_sex, StringType()) # 1. 在 SQL 中使用 spark.sql("select name, to_sex(sex) as sex from t_user").show() # 2. 在 DSL 中使用 df.select("name", convert_sex(df.sex).alias("sex")).show()

六、数据的加载与保存 (Load / Save)

Spark SQL 支持极其丰富的数据源，包括 Parquet、JSON、CSV、MySQL 等。

6.1 通用模式与 SaveMode

在写入数据时，需要特别注意SaveMode（保存模式）：

6.2 读写 Parquet (企业级首选)

Parquet 是一种列式存储格式，自带 Schema 且压缩率极高，是大数据数仓的绝对主流。

# 写入（采用覆盖模式） df.write.mode("overwrite").parquet("output/parquet_data") # 读取 spark.read.parquet("output/parquet_data").show()

6.3 读写 JDBC (MySQL)

注意坑点：在使用 Python 的 Spark SQL 连接 MySQL 时，必须将mysql-connector-java-x.x.x.jar驱动包放入pyspark/jars目录下。

url = "jdbc:mysql://192.168.100.110:3306/test props = { "user": "root", "password": "123456", "driver": "com.mysql.cj.jdbc.Driver" } # 写入 MySQL df.write.jdbc(url=url, table="df_user", mode="overwrite", properties=props) # 从 MySQL 读取 spark.read.jdbc(url=url, table="df_user", properties=props).show()

七、Spark 与 Hive 的深度整合

在企业数仓中，底层数据通常存储在 Hive 中。为了让 Spark 能直接读取 Hive 元数据，我们需要整合两者。

7.1 Spark on Hive vs Hive on Spark

• Spark on Hive：使用 Spark 引擎执行 SQL，但数据/元数据来自于 Hive。这是目前的主流用法。

• Hive on Spark：将 Hive 的底层计算引擎从 MR 替换为 Spark，作为 Hive 的一个优化手段。

7.2 整合步骤

1. 配置环境：将 Hive 的hive-site.xml拷贝到 Spark 的conf目录下。
2. 拷贝驱动：将 MySQL 驱动包mysql-connector-java-x.x.x.jar拷贝到 Spark 的jars目录。
3. 启动 Hive 元数据服务：

   hive --service metastore >> metastore.log 2>&1 &

4. 测试整合：可以通过spark-sql命令行或 Python 代码。

Python 代码读取 Hive 表：

from pyspark.sql import SparkSession if __name__ == '__main__': spark = (SparkSession.builder .master("local[*]") .appName("SparkOnHive") # 指向 Hive Metastore 服务地址 .config("hive.metastore.uris", "thrift://master:9083") # 开启 Hive 支持 .enableHiveSupport() .getOrCreate()) # 直接查询 Hive 中的表 spark.sql("select * from user1").show() spark.stop()

八、总结

Spark SQL 让大数据的处理回归到了最自然、最易用的“二维表”模型。