点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

AI 炼丹日志-31- 千呼万唤始出来 GPT-5 发布！“快的模型 + 深度思考模型 + 实时路由”，持续打造实用 AI 工具指南！📐🤖

💻 Java 篇正式开启！（300 篇）

目前 2025 年 09 月 08 日更新到：Java-118 深入浅出 MySQL ShardingSphere 分片剖析：SQL 支持范围、限制与优化实践 MyBatis 已完结，Spring 已完结，Nginx 已完结，Tomcat 已完结，分布式服务正在更新！深入浅出助你打牢基础！

📊 大数据板块已完成多项干货更新（300 篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！大数据-278 Spark MLib - 基础介绍 机器学习算法 梯度提升树 GBDT 案例 详解

章节内容

上节完成的内容如下：

• SparkSQL 介绍

• SparkSQL 特点

• SparkSQL 数据抽象

• SparkSQL 数据类型

• 

SparkSession

在 Spark2.0 之前

• SQLContext 是创建 DataFrame 和 执行 SQL 的入口

• HiveContext 通过 HiveSQL 语句操作 Hive 数据，兼 Hive 操作，HiveContext 继承自 SQLContext

• 

在 Spark2.0 后

• 这些入口点统一到了 SparkSession，SparkSession 封装了 SQLContext 及 HiveContext

• 实现了 SQLContext 即 HiveContext 所有功能

• 通过 SparkSession 可以获取到 SparkContext

RDD（Resilient Distributed Dataset，弹性分布式数据集）

RDD 是 Spark 的核心数据抽象，它代表一个不可变的、可分区、可并行计算的数据集合。RDD 是 Spark 实现高效分布式计算的基础，它能够跨集群节点自动分区，并提供丰富的操作接口。

核心特点详解

1. 不可变性

RDD 一旦创建就不能被修改。任何转换操作（如 map、filter）都会生成一个新的 RDD，而不是修改原始 RDD。这种设计带来以下优势：

• 简化容错处理

• 支持并行计算

• 便于数据共享

• 示例：对一个存储用户数据的 RDD 进行过滤操作会生成仅包含符合条件用户的新 RDD

2. 弹性容错能力

RDD 通过以下机制实现容错：

• 血统（Lineage）：记录 RDD 的生成过程（即从哪些父 RDD 通过什么操作得到）

• 检查点（Checkpointing）：可选的持久化机制，将 RDD 数据保存到稳定存储

• 分区（Partitioning）：数据被划分为多个分区，每个分区可以在不同节点上并行计算

当节点故障时，Spark 可以根据血统信息重新计算丢失的分区，而不需要复制整个数据集。

3. 分布式计算

RDD 支持两种分区方式：

1. 哈希分区：根据键的哈希值分配数据

2. 范围分区：根据键的范围分配数据

典型的分区数量通常是集群 CPU 核心数的 2-4 倍，以充分利用并行计算能力。

4. 延迟计算机制

RDD 采用惰性求值策略，包括两类操作：

• 转换操作（Transformations）：如 map、filter、reduceByKey 等，只记录计算逻辑

• 行动操作（Actions）：如 count、collect、saveAsTextFile 等，触发实际计算

这种设计允许 Spark 优化整个计算流程，减少不必要的数据传输。

5. 类型安全与 API 特点

虽然 RDD 是类型化的（如 RDD[String]），但它的 API 是松散类型的：

• 编译时不会检查类型匹配

• 运行时才会发现类型错误

• 与 DataFrame/Dataset 相比，缺少查询优化能力

创建 RDD 的三种主要方式

1. 从集合创建：使用 SparkContext.parallelize()

val data = Array(1, 2, 3, 4, 5)val rdd = sc.parallelize(data)

2. 从外部存储系统：如 HDFS、S3、本地文件系统

val rdd = sc.textFile("hdfs://path/to/file")

3. 从现有 RDD 转换：通过转换操作生成新 RDD

val newRdd = rdd.map(x => x * 2)

典型应用场景

• 迭代式算法（如机器学习）

• 交互式数据挖掘

• ETL 流水线处理

• 流式数据处理（通过微批处理实现）

• 图计算（结合 GraphX 库）

DataFrame

DataFrame 是一种基于 RDD 的分布式数据集，它具有命名的列。它是 Spark SQL 中的核心数据结构，为处理结构化数据提供了高效的接口。

主要特点

1. 结构化数据

DataFrame 是一个二维表格数据结构，具有以下特点：

• 由行(Row)和列(Column)组成

• 每列都有明确的名称和数据类型

• 类似于关系数据库中的表或 Pandas 的 DataFrame

• 支持多种数据源，包括 JSON、CSV、Parquet、JDBC 等

示例：一个表示员工信息的 DataFrame 可能包含以下列：

• id: Integer

• name: String

• department: String

• salary: Double

2. 优化引擎

DataFrame 受益于 Spark SQL 引擎的优化功能：

• Catalyst 优化器：自动执行查询优化

• 谓词下推

• 列剪裁

• 常量折叠

• 连接重排序

• Tungsten 执行引擎：提供高效的二进制内存表示

• 自动生成高效的执行计划

• 比直接使用 RDD 性能更高

3. 丰富的 API

DataFrame 提供多种操作方式：

• SQL 风格操作：

  df.select("name", "salary").filter(df.salary > 5000)

• DSL 操作：

  df.groupBy("department").agg({"salary": "avg"})

• 支持各种转换操作：

• 过滤(filter)

• 投影(select)

• 聚合(groupBy/agg)

• 排序(orderBy)

• 连接(join)

• 窗口函数(window)

4. 类型安全

与 RDD 相比，DataFrame 的类型检查特性：

• 编译时不进行类型检查

• 运行时才会验证数据类型

• 可能导致运行时错误

• 可使用 Dataset API 获得编译时类型检查

应用场景

DataFrame 特别适合以下场景：

1. 结构化数据处理

2. 数据仓库查询

3. ETL 流程

4. 数据分析与可视化

5. 机器学习数据准备

性能优势

通过 DataFrame API 编写代码通常比直接使用 RDD API 更高效，因为：

1. 优化器可以理解操作语义

2. 减少数据序列化/反序列化开销

3. 采用列式存储格式

4. 启用代码生成技术

DataSet

DataSet 是 Spark 1.6 引入的一个新的数据抽象，它结合了 RDD 的强类型优势和 DataFrame 的优化能力。

特点：

• 类型安全：DataSet 是强类型的，它利用编译时类型检查，确保在编译时检测类型错误。

• 优化和性能：DataSet 受益于 Catalyst 优化器和 Tungsten 执行引擎，提供与 DataFrame 相同的优化能力，同时保留了类型安全性。

• 更丰富的 API：DataSet 提供了 RDD 的大部分 API，如 map、filter 等，同时也支持 SQL 查询。

• 统一 API：DataSet API 统一了 RDD 和 DataFrame，提供了一种更具表现力和安全性的编程模型。

DataFrame & Dataset 创建

不要刻意区分： DF & DS，DF 是一种特殊的 DS：ds.transformation => ds

由 Range 生成 Dataset

在 spark-shell 中进行测试

val numDS = spark.range(5, 100, 5)// orderBy 转换操作 numDS.orderBy(desc("id")).show(5)// 统计信息numDS.describe().show// 显示 Schema 信息numDS.printSchema// 使用RDD执行同样的操作numDS.rdd.map(_.toInt).stats// 检查分区数numDS.rdd.getNumPartitions

运行测试的过程如下图所示：

有集合生成 Dataset

Dataset = RDD[case class]，在 spark-shell 中进行测试

case class Person(name: String, age: Int, height: Int)
// 注意 Seq 中元素的类型val seq1 = Seq(Person("Jack", 28, 184), Person("Tom", 10, 144), Person("Andy", 16, 165))
val ds1 = spark.createDataset(seq1)ds1.printSchemads1.show

执行的结果：

再来一个测试：

val seq2 = Seq(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))val ds2 = spark.createDataset(seq2)ds2.printSchemads2.show

执行的结果：

由集合生成 DataFrame

DataFrame = RDD[Row] + Schema 继续进行测试：

val lst = List(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))val df1 = spark.createDataFrame(lst).withColumnRenamed("_1", "name1").withColumnRenamed("_2", "age1").withColumnRenamed("_3", "height1")df1.orderBy("age1").show(10)

执行的结果如下图所示：

RDD 转成 DataFrame

DataFrame = RDD[Row] + Schema

val arr = Array(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))val rdd1 = sc.makeRDD(arr).map(f => Row(f._1, f._2, f._3))
val schema = StructType(  StructField("name", StringType, false) ::  StructField("age", IntegerType, false) ::  StructField("height", IntegerType, false) ::  Nil)
val schema1 = (new StructType).add("name", "string", false).add("age", "int", false).add("height", "int", false)val rddToDF = spark.createDataFrame(rdd1, schema)rddToDF.orderBy(desc("name")).show(false)

执行的结果如下图：

RDD 转 Dataset

Dataset = RDD[case class]DataFrame = RDD[Row] + Schema

val arr = Array(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))val rdd1 = sc.makeRDD(arr)val ds2 = spark.createDataset(rdd1)ds2.show(10)

执行的结果如下图：

从文件创建 DataFrame

CSV 文件

我们生成了一个 CSV 文件，大致内容如下：

运行测试

val df1 = spark.read.csv("/opt/wzk/data/people1.csv")df1.printSchema()df1.show()

运行结果如下图所示：

三者转换

Spark SQL 提供了一个领域特定语言（DSL）以方便操作结构化数据，核心思想还是 SQL，仅仅是一个语法问题。

RDD 与 DataFrame 之间的转换

RDD 转换为 DataFrame

将 RDD 转换为 DataFrame 需要提供数据的模式信息。通常你会使用 toDF() 方法将 RDD 转换为 DataFrame。这里有两种主要方法：

• 使用隐式转换：需要导入 spark.implicits._，这允许你在不显式提供模式的情况下将常见的 RDD（如元组）转换为 DataFrame。

• 使用 StructType 定义模式：如果 RDD 的数据结构比较复杂，或者你需要精确控制 DataFrame 的模式，可以使用 StructType 和 Row。

DataFrame 转换为 RDD：

• 将 DataFrame 转换为 RDD 非常简单，只需调用 rdd 方法即可

DataFrame 与 DataSet 之间的转换

DataFrame 转换为 DataSet

• DataFrame 是无类型的，而 DataSet 是类型化的。为了将 DataFrame 转换为 DataSet，你需要定义一个对应的数据类型（通常是一个 case class）并使用 as[T] 方法

DataSet 转换为 DataFrame

• 将 DataSet 转换为 DataFrame 非常简单，只需调用 toDF() 方法即可

RDD 与 DataSet 之间的转换

RDD 转换为 DataSet

• 将 RDD 转换为 DataSet 需要将 RDD 的元素类型与 DataSet 的类型一致。与将 RDD 转换为 DataFrame 类似，通常使用隐式转换或显式提供模式信息

DataSet 转换为 RDD

• DataSet 本质上是类型化的 RDD，因此转换为 RDD 非常直接，只需调用 rdd 方法

最终汇总

• RDD 转换为 DataFrame：使用 toDF()，或使用 createDataFrame() 提供模式。

• DataFrame 转换为 RDD：使用 rdd 方法，转换后元素类型为 Row。

• DataFrame 转换为 DataSet：使用 as[T] 方法，需提供对应的 case class。

• DataSet 转换为 DataFrame：使用 toDF() 方法。

• RDD 转换为 DataSet：使用 toDS()，需提供对应的 case class。

• DataSet 转换为 RDD：使用 rdd 方法。