Spark2.x RDD, Dataframe Dataset API操作

喜欢ヅ旅行 2023-07-06 06:27 109阅读 0赞

Spark提供了三种主要的与数据相关的API：`RDD`、`DataFrame`、`Dataset`

> **RDD (Spark1.0) —> Dataframe(Spark1.3) —> Dataset(Spark1.6)**

![format_png][]

image

## RDD ##

RDD是Spark提供的最主要的一个抽象概念（Resilient Distributed Dataset），它是一个element的collection，分区化的位于集群的节点中，支持并行处理。

## DataFrame ##

Spark从1.3版本开始引入DataFrame，它克服了RDD的最主要的挑战。DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame是一个分布式的数据collection，而且将数据按照列名进行组织。在概念上它与关系型的数据库的表或者R/Python语言中的DataFrame类似。与之一起提供的还有，Spark引入了catalyst优化器，它可以优化查询。

## Dataset ##

Dataset是一个由特定领域的对象组成的强类型（typedrel）集合，可以使用函数（DSL）或关系运算（SQL）进行并行的转换操作。每个Dataset还有一个称为"DataFrame"的无类型（untypedrel）视图。

Dataset API是对DataFrame的一个扩展，使得可以支持类型安全的检查，并且对类结构的对象支持程序接口。它是强类型的，不可变collection，并映射成一个相关的schema。Dataset API的核心是一个被称为Encoder的概念。它是负责对JVM的对象以及表格化的表达（tabular representation）之间的相互转化。表格化的表达在存储时使用了Spark内置的Tungsten二进制形式，允许对序列化数据操作并改进了内存使用。在Spark 1.6版本之后，支持自动化生成Encoder，可以对广泛的primitive类型（比如String，Integer，Long等）、Scala的case class以及Java Bean自动生成对应的Encoder。

## RDD、DataFrame和Dataset三者的比较 ##

![format_png 1][]

img

### RDD和DataFrame的区别 ###

DataFrame与RDD的主要区别在于，DataFrame带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。使得Spark SQL得以洞察更多的结构信息，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。

RDD由于无从得知所存数据元素的具体内部结构，Spark Core只能在stage层面进行简单、通用的流水线优化。 DataFrame底层是以RDD为基础的分布式数据集。

### RDD和Dataset的区别 ###

Dataset与RDD类似，但是它不使用Java序列化或Kryo，而是使用专用的Encoder编码器来序列化对象以便通过网络进行处理或传输。虽然Encoder编码器和标准序列化都负责将对象转换为字节，但Encoder编码器是动态生成的代码，并使用一种格式，允许Spark执行许多操作，如过滤，排序和散列，而无需将字节反序列化为对象。

### Dataset和DataFrame的区别 ###

**区别**

1.  Dataset是强类型typedrel的，会在**编译**的时候进行类型检测；而DataFrame是弱类型untypedrel的，在**执行**的时候进行类型检测；
2.  Dataset是通过Encoder进行序列化，支持动态的生成代码，直接在bytes的层面进行排序，过滤等的操作；而DataFrame是采用可选的java的标准序列化或是kyro进行序列化

**联系**

1.  在spark2.x，DataFrame和Dataset的api进行了统一
2.  在语法角度，DataFrame是Dataset中每一个元素为Row类型的特殊情况`type DataFrame = Dataset[Row]`
3.  DataFrame和Dataset实质上都是一个逻辑计划，并且是懒加载的，都包含着scahema信息，只有到数据要读取的时候，才会将逻辑计划进行分析和优化，并最终转化为RDD
4.  二者由于api是统一的，所以都可以采用DSL和SQL方式进行开发，都可以通过sparksession对象进行创建或者是通过transform转化操作得到

## 数据格式转换 ##

### RDD DataFrame相互转化 ###

/**
      * 手动指定schema
      */
    
    // 定义schema
    val schema = StructType(Array(
      StructField("name", StringType, nullable = true),
      StructField("age", IntegerType, nullable = true)
    ))
    
    
    val rowRDD = rdd.map(x => Row(x._1, x._2))
    // 将personRDD转化为DataFrame
    var df = spark.createDataFrame(rowRDD, schema)
    df.show(false)
    
    // 隐式转换
    import spark.implicits._
    
    /**
      * Tuple RDD转换成DataFrame
      */
    
    // 未指定列，列默认名为`_1`,`_2`
    df = rdd.toDF()
    df.show(false)
    // 指定列名
    rdd.toDF("name", "age")
    df.show(false)
    
    /**
      * Case class RDD转换成DataFrame
      */
    
    // 将rdd转化为DataFrame
    val personRDD = rdd.map(x => Person(x._1, x._2))
    df = personRDD.toDF("name", "age")
    df.show(false)
    
    /**
      * DataFrame转RDD
      */
    
    val rdd1 = df.rdd.map(row => (row.get(0), row.get(1)))
    rdd1.collect().foreach(println)

### RDD Dataset相互转化 ###

/**
      * RDD转DataSet
      */
    
    /**
      * Tuple RDD转Dataset
      */
    var dataset = rdd.toDS()
    dataset.show(false)
    dataset = spark.createDataset(rdd)
    dataset.show(false)
    
    /**
      * Case class RDD转Dataset
      */
    var dataset1 = personRDD.toDS()
    dataset1.show(false)
    dataset1 = spark.createDataset(personRDD)
    dataset1.show(false)
    
    /**
      * Dataset转RDD
      */
    val rdd2 = dataset.rdd
    rdd2.collect().foreach(println)

### DataFrame Dataset相互转化 ###

/**
      * Dataset转DataFrame
      */
    val df2 = dataset.toDF()
    df2.show(false)
    
    /**
      * DataFrame转Dataset
      */
    val dataset2 = df.as[Person]
    dataset2.show(false)

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zNDIzNjY1LWFhZjY5Yzc4MGU2ZWU4YmY_aW1hZ2VNb2dyMi9hdXRvLW9yaWVudC9zdHJpcHxpbWFnZVZpZXcyLzIvdy85OTA?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8zNDIzNjY1LWM0YzI0Y2Q4YzI2NzJjNTA_aW1hZ2VNb2dyMi9hdXRvLW9yaWVudC9zdHJpcHxpbWFnZVZpZXcyLzIvdy84NjM?x-oss-process=image/format,png