在 Spark SQL 中创建 DataFrame 有多种方式，DataFrame 与 RDD 在抽象层次和优化机制上有显著区别。

DataFrame 的创建方式

1. 从结构化数据文件创建

// 从 JSON 文件
val dfJson = spark.read.json("path/to/data.json")

// 从 CSV 文件
val dfCsv = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv("path/to/data.csv")

// 从 Parquet 文件
val dfParquet = spark.read.parquet("path/to/data.parquet")

2. 从 RDD 转换创建

// 从 RDD[Row] 创建
case class Person(name: String, age: Int)
val peopleRDD = sc.parallelize(Seq(
  Person("Alice", 25),
  Person("Bob", 30)
))

import spark.implicits._
val dfFromRDD = peopleRDD.toDF()

// 或者显式指定 Schema
import org.apache.spark.sql.types._
val schema = StructType(Array(
  StructField("name", StringType, true),
  StructField("age", IntegerType, true)
))

val rowRDD = peopleRDD.map(p => Row(p.name, p.age))
val dfWithSchema = spark.createDataFrame(rowRDD, schema)

3. 从外部数据库创建

// 从 JDBC 数据源
val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://localhost/test")
  .option("dbtable", "users")
  .option("user", "username")
  .option("password", "password")
  .load()

4. 编程方式创建

// 使用 createDataFrame
val data = Seq(
  ("Alice", 25, "Engineer"),
  ("Bob", 30, "Manager")
)

val df = spark.createDataFrame(data).toDF("name", "age", "job")

// 使用 toDF 简化版
import spark.implicits._
val simpleDF = Seq(
  ("Alice", 25),
  ("Bob", 30)
).toDF("name", "age")

5. 从 Hive 表创建

// 直接读取 Hive 表
val hiveDF = spark.sql("SELECT * FROM my_database.users")

// 或者通过 catalog
val tableDF = spark.table("my_database.sales")

DataFrame 与 RDD 的核心区别

抽象层次对比

具体差异对比表

性能对比示例

RDD 方式（较低效）

// RDD 实现过滤和聚合
val rdd = sc.textFile("data.txt")
val result = rdd
  .map(line => line.split(","))
  .filter(fields => fields(2).toInt > 25)  // 运行时解析
  .map(fields => (fields(0), 1))
  .reduceByKey(_ + _)  // Shuffle 操作

DataFrame 方式（高效优化）

// DataFrame 实现相同逻辑
val df = spark.read.option("header", "true").csv("data.csv")
val result = df
  .filter($"age" > 25)  // 谓词下推优化
  .groupBy("name")
  .count()  // 聚合优化

Catalyst 优化器工作流程

// 原始查询
val query = df.filter($"age" > 25).select($"name", $"salary")

// Catalyst 优化过程：
// 1. 分析阶段：解析列名和类型
// 2. 逻辑优化：谓词下推、列裁剪
// 3. 物理计划：选择最优执行策略
// 4. 代码生成：生成高效字节码

实际选择建议

使用 RDD 的情况：

• 需要极细粒度的控制
• 处理非结构化数据
• 实现自定义的分布式算法
• 与遗留代码集成

使用 DataFrame 的情况：

• 结构化数据处理
• SQL 查询和关系型操作
• 需要性能优化的场景
• 与 BI 工具集成

最佳实践： 优先使用 DataFrame/Dataset API，仅在必要时降级到 RDD API，以获得最佳的开发效率和运行时性能。