初识Spark SQL

一、引言在上一篇博客中我们深入学习了Spark Core与RDD。RDD的强大在于其极高的灵活性——它允许我们用函数式编程处理任何类型的数据。但是随着大数据分析逐渐走向业务化我们面对的数据80%以上都是结构化的例如数据库中的表、CSV文件、JSON日志。此时RDD的劣势显现了出来缺乏数据结构信息SchemaSpark框架本身不知道RDD内部元素的类型结构无法进行底层优化。开发门槛高对于数据分析师来说写SQL远比写Scala/Python函数要方便。为了解决这些问题Spark SQL应运而生。它不仅提供了更友好的编程接口DataFrame/DataSet还内置了一个强大的Catalyst 优化器能将SQL语句自动优化执行效率甚至远超手写的RDD代码。今天我们就来彻底掌握Spark SQL。二、核心概念DataFrame与Dataset2.1 什么是DataFrame与RDD类似DataFrame也是一个分布式的数据容器。但不同的是DataFrame 带有 Schema表结构信息。它更像传统数据库中的一张二维表格不仅有数据还记录了每一列的名称和类型。RDD 与 DataFrame 的区别RDD[Person]Spark 知道里面装的是 Person 对象但不知道 Person 内部有什么属性。DataFrameSpark 清晰地知道这张表包含nameString类型和ageInt类型两列。有了 SchemaSpark SQL 就可以像关系型数据库那样进行列裁剪、谓词下推等高级优化。2.2 两种编程风格Spark SQL 支持两种编程风格SQL 风格将 DataFrame 注册为临时视图直接书写标准 SQL。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(rD:\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.txtWordCount和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 函数作为 UDFUser 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保存模式SaveMode说明errorifexists默认文件存在则报错append追加写入overwrite覆盖原有数据ignore忽略不报错也不写入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(urlurl, tabledf_user, modeoverwrite, propertiesprops) # 从 MySQL 读取 spark.read.jdbc(urlurl, tabledf_user, propertiesprops).show()七、Spark 与 Hive 的深度整合在企业数仓中底层数据通常存储在 Hive 中。为了让 Spark 能直接读取 Hive 元数据我们需要整合两者。7.1 Spark on Hive vs Hive on SparkSpark on Hive使用 Spark 引擎执行 SQL但数据/元数据来自于 Hive。这是目前的主流用法。Hive on Spark将 Hive 的底层计算引擎从 MR 替换为 Spark作为 Hive 的一个优化手段。7.2 整合步骤配置环境将 Hive 的hive-site.xml拷贝到 Spark 的conf目录下。拷贝驱动将 MySQL 驱动包mysql-connector-java-x.x.x.jar拷贝到 Spark 的jars目录。启动 Hive 元数据服务hive --service metastore metastore.log 21 测试整合可以通过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 让大数据的处理回归到了最自然、最易用的“二维表”模型。核心模块关键内容核心抽象DataFrame带 Schema 的分布式数据、Dataset编程风格SQL 风格易用、DSL 风格灵活、易组合数据类型转换RDD 与 DataFrame 互转Schema 是关键高级特性自定义函数UDF、窗口函数分组TopN、行列转换explode数据源支持 Parquet、JSON、CSV、MySQL生态整合Spark on Hive复用元数据无缝对接数仓