[spark] RDD, DataFrame和DataSet是什么?如何相互转化

2023-12-27 20:50:21

文章目录

是什么

在 Apache Spark 中,RDD(Resilient Distributed Dataset)、DataFrame 和 Dataset 是三个不同的数据抽象层,各自有不同的特点和用途。

  1. RDD(Resilient Distributed Dataset):

    • RDD 是 Spark 最基本的抽象,表示分布在集群节点上的不可变、可弹性恢复的数据集。
    • RDD 提供了一系列转换操作(transformation)和行动操作(action)来处理数据。转换操作会创建一个新的 RDD,而行动操作会返回结果或将数据写入外部存储。
    • RDD 编程风格是面向函数的,并且对类型的要求较弱,适用于较低层次的数据处理和计算。
  2. DataFrame:

    • DataFrame 是在 Spark 1.3 引入的更高层次的抽象,构建在 RDD 之上,提供了更丰富的结构化数据处理能力。
    • DataFrame 是以类似关系型数据库表的形式组织的分布式数据集合,每列都有名称和数据类型。
    • Spark 提供了一系列的 DataFrame 操作,包括 SQL 查询、过滤、聚合等,而且可以利用 Catalyst 引擎进行优化执行计划。
    • DataFrame 通常比 RDD 更易于使用,特别是在执行结构化查询时。
  3. Dataset:

    • Dataset 是 Spark 1.6 引入的抽象,它是 DataFrame 的扩展,提供了类型安全的编程接口。
    • Dataset 支持强类型的编程语言(如 Scala 和 Java),并在编译时进行类型检查,提供更好的性能和更好的错误检测。
    • Dataset 同时提供了 DataFrame 的结构化查询能力,可以通过编程接口或者 SQL 查询进行操作。

简而言之,RDD 是最基本的抽象,DataFrame 是对结构化数据的更高层次抽象,而 Dataset 是在 DataFrame 基础上提供了类型安全性的扩展。在实际使用中,通常优先选择使用 DataFrame 或 Dataset,因为它们更适合进行结构化数据处理和利用 Spark 的优化能力。

如何转化

在 Apache Spark 中,可以通过一些方法进行 RDD、DataFrame 和 Dataset 之间的转化。以下是一些示例 Scala 代码,演示了如何进行这些转化:

  1. RDD 转 DataFrame:
import org.apache.spark.sql.{SparkSession, Row}
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

val spark = SparkSession.builder
  .appName("RDD to DataFrame")
  .master("local")
  .getOrCreate()

// 创建一个示例 RDD
val rdd = spark.sparkContext.parallelize(Seq(
  Row("John", 25),
  Row("Alice", 30),
  Row("Bob", 28)
))

// 定义结构信息
val schema = StructType(Seq(
  StructField("name", StringType, nullable = true),
  StructField("age", IntegerType, nullable = true)
))

// 创建 DataFrame
val df = spark.createDataFrame(rdd, schema)

// 打印 DataFrame
df.show()

// 关闭 SparkSession
spark.stop()
  1. DataFrame 转 RDD:
import org.apache.spark.sql.{SparkSession, Row}

val spark = SparkSession.builder
  .appName("DataFrame to RDD")
  .master("local")
  .getOrCreate()

// 创建一个示例 DataFrame
val df = spark.createDataFrame(Seq(
  ("John", 25),
  ("Alice", 30),
  ("Bob", 28)
)).toDF("name", "age")

// 转换为 RDD
val rdd = df.rdd

// 打印 RDD
rdd.foreach(println)

// 关闭 SparkSession
spark.stop()
  1. DataFrame 转 Dataset:
import org.apache.spark.sql.{SparkSession, Encoder}
import org.apache.spark.sql.expressions.Encoder

case class Person(name: String, age: Int)

val spark = SparkSession.builder
  .appName("DataFrame to Dataset")
  .master("local")
  .getOrCreate()

// 创建一个示例 DataFrame
val df = spark.createDataFrame(Seq(
  ("John", 25),
  ("Alice", 30),
  ("Bob", 28)
)).toDF("name", "age")

// 转换为 Dataset
val ds = df.as[Person]

// 打印 Dataset
ds.show()

// 关闭 SparkSession
spark.stop()
  1. Dataset 转 DataFrame:
import org.apache.spark.sql.{SparkSession, Encoder}
import org.apache.spark.sql.expressions.Encoder

case class Person(name: String, age: Int)

val spark = SparkSession.builder
  .appName("Dataset to DataFrame")
  .master("local")
  .getOrCreate()

// 创建一个示例 Dataset
val ds = Seq(
  Person("John", 25),
  Person("Alice", 30),
  Person("Bob", 28)
).toDS()

// 转换为 DataFrame
val df = ds.toDF()

// 打印 DataFrame
df.show()

// 关闭 SparkSession
spark.stop()

这些示例代码演示了在 Spark 中如何进行 RDD、DataFrame 和 Dataset 之间的基本转化。根据实际需求和数据处理场景,选择适当的数据抽象进行操作。

文章来源:https://blog.csdn.net/qq_44810930/article/details/135252797
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。