当前位置：首页 > news >正文

Spark核心数据（RDD、DataFrame 和 Dataset） - 教程

news 2026/5/12 11:36:33

在 Spark 中，RDD、DataFrame 和 Dataset 是三种核心数据抽象，它们既一脉相承又各有侧重，以下从定义、特性、关系及语法支持等方面详细说明：

一、核心定义与特性

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

定义：Spark 最基础的分布式数据抽象，是不可变的、分区的对象集合。
核心特性：
- 无 schema：数据以对象（如 Python 元组、Java 对象）形式存在，不预设字段结构。
- 弹性容错：通过血缘关系（Lineage）记录依赖，某分区数据丢失可通过父 RDD 重算恢复。
- 分布式并行：数据自动分区，支持跨节点并行计算。
- 函数式操作：通过 map、filter 等转换操作处理数据，不直接修改原 RDD。

2. DataFrame

定义：分布式的二维表结构数据集合，包含行和列，且有明确的 schema（字段名和数据类型）。
核心特性：
- 有 schema：类似数据库表结构，预先定义列名和类型（如 id: int, name: string）。
- 优化执行：依赖 Catalyst 优化器，自动优化执行计划（如谓词下推、列裁剪）。
- 多数据源兼容：可直接读取 CSV、JSON、Parquet、Hive 表等结构化数据。
- 无类型安全：编译/运行前不检查字段类型，错误可能在执行时暴露（如访问不存在的列）。

3. Dataset（主要支持 Java/Scala，Python 不原生支持）

定义：DataFrame 与 RDD 的结合体，既保留 schema 信息，又支持类型安全（编译时类型检查）。
核心特性：
- 强类型绑定：数据与 Java/Scala 类（如 Case Class）绑定，操作时直接访问类属性。
- 双重接口：支持 RDD 风格的 lambda 函数操作，也支持 DataFrame 的 SQL 和 API 操作。
- 优化执行：同样基于 Catalyst 优化器，性能与 DataFrame 接近。
- 编译时检查：字段错误（如拼写错误、类型不匹配）在编译阶段即可发现。

二、三者之间的关系

底层依赖关系：
- DataFrame 和 Dataset 底层均基于 RDD 实现（可通过 .rdd 方法转换为 RDD）。
- Dataset 在 Spark 2.0 后与 DataFrame 统一，在 Scala 中 DataFrame = Dataset[Row]（Row 是无类型的行对象）。
演进关系：
- RDD 是 Spark 1.0 的核心抽象，解决了分布式计算的基础问题（容错、并行）。
- DataFrame 是 Spark 1.3 引入的结构化数据抽象，弥补 RDD 无 schema、无优化器的缺陷。
- Dataset 是 Spark 1.6 引入的类型安全抽象，在 Java/Scala 中增强了数据操作的安全性。
功能覆盖：
- RDD 功能最基础，支持所有数据类型（结构化、非结构化），但无优化。
- DataFrame 和 Dataset 专注于结构化数据，通过 schema 和优化器提升效率，Dataset 额外增加类型安全。

三、支持的语法（分语言示例）

1. RDD 语法（Python/Scala 通用风格）

# Python 示例
# 创建 RDD
rdd = sc.parallelize([(1, "Alice", 25), (2, "Bob", 30)])
# 转换操作：过滤年龄>28的记录，提取id和姓名
filtered_rdd = rdd.filter(lambda x: x[2] > 28)
result_rdd = filtered_rdd.map(lambda x: (x[0], x[1]))
# 行动操作：收集结果
print(result_rdd.collect())  # 输出：[(2, 'Bob')]

// Scala 示例
val rdd = sc.parallelize(Seq((1, "Alice", 25), (2, "Bob", 30)))
val result = rdd.filter(_._3 > 28).map(x => (x._1, x._2)).collect()

2. DataFrame 语法（SQL + API 风格）

# Python 示例
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
# 定义 schema 并创建 DataFrame
schema = StructType([
StructField("id", IntegerType()),
StructField("name", StringType()),
StructField("age", IntegerType())
])
df = spark.createDataFrame([(1, "Alice", 25), (2, "Bob", 30)], schema)
# 方式1：DataFrame API
result_df = df.filter(df.age > 28).select("id", "name")
# 方式2：SQL 操作（需注册临时视图）
df.createOrReplaceTempView("users")
result_df = spark.sql("SELECT id, name FROM users WHERE age > 28")
result_df.show()
# +---+----+
# | id|name|
# +---+----+
# |  2| Bob|
# +---+----+

3. Dataset 语法（Scala 类型安全风格）

// Scala 示例
// 定义样例类（与数据结构绑定）
case class User(id: Int, name: String, age: Int)
// 创建 Dataset
val ds = spark.createDataset(Seq(User(1, "Alice", 25), User(2, "Bob", 30)))
// 方式1：类型安全的 lambda 操作
val filteredDs = ds.filter(_.age > 28)  // 直接访问类属性，编译时检查
// 方式2：混合使用 DataFrame API
val resultDs = filteredDs.select("id", "name")
resultDs.show()
// +---+----+
// | id|name|
// +---+----+
// |  2| Bob|
// +---+----+

四、对比总结

维度	RDD	DataFrame	Dataset（Java/Scala）
数据结构	无 schema（对象集合）	有 schema（二维表）	有 schema（绑定类）
类型安全	支持（针对对象）	不支持（运行时检查）	支持（编译时检查）
优化器	无	有（Catalyst）	有（Catalyst）
操作接口	函数式 API（map 等）	SQL + DataFrame API	SQL + API + lambda
适用场景	非结构化数据处理	结构化数据+SQL分析	类型安全的结构化处理
Python 支持	完全支持	完全支持	不支持