摘要：RDD 和 Dataset 是 Spark 两大核心抽象，它们之间的互操作是连接"低级 API"与"高级 API"的桥梁。本文从 6 条转换路径、Encoder 底层序列化/反序列化机制（InternalRow-Row-CaseClass）、ExpressionEncoder 表达式树、spark.implicits 隐式转换链、以及选型决策五个维度，配合 2 张架构图 + 完整代码实例，彻底掌握 RDD↔Dataset 互操作。

关键词：RDD, Dataset, toDS, toDF, createDataFrame, Encoder, InternalRow, spark.implicits, mapPartitions

---

一、开篇：为什么要互操作？

RDD 提供了精细的分区控制和函数式 API，Dataset 提供了 Catalyst 优化、Tungsten 引擎和类型安全。实际生产中常常需要"取两者之长"。

互操作 6 条路径速览
RDD → Dataset:
  A1. rdd.toDS()              — 需要隐式 Encoder
  A2. rdd.toDF("c1","c2")     — 简单列名
  A3. spark.createDataFrame(rdd, schema) — 手动 Schema
Dataset → RDD:
  B1. ds.rdd    → RDD[T]      类型安全
  B2. df.rdd    → RDD[Row]    通用行
  B3. ds.mapPartitions{...}.rdd — 手动转换

---

二、互操作架构全景

2.1 RDD → Dataset

import spark.implicits._  // 必需！

case class Person(id: Long, name: String, age: Int)
val rdd: RDD[Person] = sc.parallelize(Seq(
  Person(1, "Alice", 30), Person(2, "Bob", 25)))

// A1: .toDS() — 最简洁
val ds: Dataset[Person] = rdd.toDS()

// A2: .toDF() — Tuple RDD 转 DataFrame
val tupleRdd = rdd.map(p => (p.id, p.name, p.age))
val df = tupleRdd.toDF("id", "name", "age")

// A3: createDataFrame — 手动 Schema
val rowRdd = rdd.map(p => Row(p.id, p.name, p.age))
val schema = StructType(Seq(
  StructField("id", LongType),
  StructField("name", StringType),
  StructField("age", IntegerType)))
val df = spark.createDataFrame(rowRdd, schema)

2.2 Dataset → RDD

// B1: ds.rdd — 类型安全
val rdd1: RDD[Person] = ds.rdd

// B2: df.rdd — 返回 RDD[Row]
val rdd2: RDD[Row] = df.rdd
rdd2.map(row => (row.getLong(0), row.getString(1)))

// B3: mapPartitions — 内部转换
val rdd3: RDD[Person] = ds.mapPartitions { iter =>
  iter.map { row =>
    Person(row.id, row.name.toUpperCase, row.age + 1)
  }
}.rdd

---

三、底层映射链：InternalRow-Row-CaseClass

ExpressionEncoder 如何工作

// Serializer (Person → InternalRow)
CreateNamedStruct(
  Literal("id"),   Invoke(obj, "id", LongType),
  Literal("name"), Invoke(obj, "name", StringType),
  Literal("age"),  Invoke(obj, "age", IntType))

// Deserializer (InternalRow → Person)
newInstance(classOf[Person],
  GetColumnByOrdinal(0, LongType),    // id
  GetColumnByOrdinal(1, StringType),  // name
  GetColumnByOrdinal(2, IntType))     // age

spark.implicits 隐式转换链

// 隐式类 rddToDatasetHolder
implicit def rddToDatasetHolder[T](rdd: RDD[T])
  (implicit encoder: Encoder[T]): DatasetHolder[T]

// .toDS() 调用链:
rdd.toDS()  // RDD 没有 toDS 方法
  → rddToDatasetHolder(rdd)(encoder)  // 隐式包装
  → holder.toDS()                      // DatasetHolder.toDS
  → spark.createDataset(rdd)(encoder)  // 最终创建 Dataset

---

四、选型决策

用 Dataset 的时候:
  SQL 分析 · filter/map 链式操作 · groupBy 聚合
  Catalyst 谓词下推 · 列裁剪 · WSCG 加速

切回 RDD 的时候:
  精细控制分区数/分区函数
  非结构化/半结构化数据预处理

推荐模式:
  RDD(预处理) → Dataset(Catalyst优化) → RDD(精细控制)

---

五、总结

1. 6 条转换路径：RDD→DS(toDS/toDF/createDataFrame) + DS→RDD(rdd/mapPartitions)。

2. Encoder 核心：ExpressionEncoder 通过 Catalyst 表达式树生成 Serializer/Deserializer，比 Kryo 快 10x。

3. 混合架构：RDD(预处理) → Dataset(Catalyst) → RDD(精细控制)，取两者之长。

---

作者：starzy
博客：blog.starzy.cn
GitHub：starzy1990.github.io
专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践