Spark Core组件:RDD、DataFrame和DataSet

  1. 云栖社区>
  2. 博客>
  3. 正文

Spark Core组件:RDD、DataFrame和DataSet

wsc449 2018-01-17 09:11:26 浏览727
展开阅读全文

1. 介绍

spark生态系统中,Spark Core,包括各种Spark的各种核心组件,它们能够对内存和硬盘进行操作,或者调用CPU进行计算。
spark core定义了RDD、DataFrame和DataSet

这里写图片描述

spark最初只有RDD,DataFrame在Spark 1.3中被首次发布,DataSet在Spark1.6版本中被加入。

2. RDD

RDD:Spark的核心概念是RDD (resilientdistributed dataset),指的是一个只读的,可分区的分布式数据集,这个数据集的全部或部分可以缓存在内存中,在多次计算间重用。

优点:
编译时类型安全
编译时就能检查出类型错误
面向对象的编程风格
直接通过类名点的方式来操作数据
缺点:
序列化和反序列化的性能开销
无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化.
GC的性能开销
频繁的创建和销毁对象, 势必会增加GC
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

object Run {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val sqlContext = new SQLContext(sc)

    /**
      * id      age
      * 1       30
      * 2       29
      * 3       21
      */
    case class Person(id: Int, age: Int)
    val idAgeRDDPerson = sc.parallelize(Array(Person(1, 30), Person(2, 29), Person(3, 21)))

    // 优点1
    // idAge.filter(_.age > "") // 编译时报错, int不能跟String比

    // 优点2
    idAgeRDDPerson.filter(_.age > 25) // 直接操作一个个的person对象
  }
}

3. DataFrame

在Spark中,DataFrame是一种以RDD为基础的分布式数据集,类似于传统数据库中的二维表格。DataFrame与RDD的主要区别在于,前者带有schema元信息,即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得Spark SQL得以洞察更多的结构信息,从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化,最终达到大幅提升运行时效率的目标。反观RDD,由于无从得知所存数据元素的具体内部结构,Spark Core只能在stage层面进行简单、通用的流水线优化。

这里写图片描述

DataFrame引入了schema和off-heap

schema : RDD每一行的数据, 结构都是一样的.
这个结构就存储在schema中。 Spark通过schame就能够读懂数据, 因此在通信和IO时就只需要序列化和反序列化数据,而结构的部分就可以省略了。 off-heap : 意味着JVM堆以外的内存,这些内存直接受操作系统管理(而不是JVM)。Spark能够以二进制的形式序列化数据(不包括结构)到off-heap中,当要操作数据时,就直接操作off-heap内存。由于Spark理解schema,所以知道该如何操作。

off-heap就像地盘,schema就像地图, Spark有地图又有自己地盘了, 就可以自己说了算了, 不再受JVM的限制,也就不再收GC的困扰了。通过schema和off-heap,DataFrame解决了RDD的缺点,但是却丢了RDD的优点。 DataFrame不是类型安全的, API也不是面向对象风格的。

import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object Run {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val sqlContext = new SQLContext(sc)
    /**
      * id      age
      * 1       30
      * 2       29
      * 3       21
      */
    val idAgeRDDRow = sc.parallelize(Array(Row(1, 30), Row(2, 29), Row(4, 21)))

    val schema = StructType(Array(StructField("id", DataTypes.IntegerType), StructField("age", DataTypes.IntegerType)))

    val idAgeDF = sqlContext.createDataFrame(idAgeRDDRow, schema)
    // API不是面向对象的
    idAgeDF.filter(idAgeDF.col("age") > 25) 
    // 不会报错, DataFrame不是编译时类型安全的
    idAgeDF.filter(idAgeDF.col("age") > "") 
  }
}

4. DataSet

Dataset是一个强类型的特定领域的对象,这种对象可以函数式或者关系操作并行地转换。每个Dataset也有一个被称为一个DataFrame的类型化视图,这种DataFrame是Row类型的Dataset,即Dataset[Row]

Dataset是“懒惰”的,只在执行行动操作时触发计算。本质上,数据集表示一个逻辑计划,该计划描述了产生数据所需的计算。当执行行动操作时,Spark的查询优化程序优化逻辑计划,并生成一个高效的并行和分布式物理计划。

DataSet结合了RDD和DataFrame的优点,,并带来的一个新的概念Encoder 当序列化数据时,Encoder产生字节码与off-heap进行交互,能够达到按需访问数据的效果, 而不用反序列化整个对象。 Spark还没有提供自定义Encoder的API,但是未来会加入。

下面看DataFrame和DataSet在2.0.0-preview中的实现

下面这段代码, 在1.6.x中创建的是DataFrame
// 上文DataFrame示例中提取出来的
val idAgeRDDRow = sc.parallelize(Array(Row(1, 30), Row(2, 29), Row(4, 21)))

val schema = StructType(Array(StructField("id", DataTypes.IntegerType), StructField("age", DataTypes.IntegerType)))

val idAgeDF = sqlContext.createDataFrame(idAgeRDDRow, schema)
但是同样的代码在2.0.0-preview中, 创建的虽然还叫DataFrame

// sqlContext.createDataFrame(idAgeRDDRow, schema) 方法的实现, 返回值依然是DataFrame
def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame = {
sparkSession.createDataFrame(rowRDD, schema)
}
但是其实却是DataSet, 因为DataFrame被声明为Dataset[Row]

package object sql {
  // ...省略了不相关的代码

  type DataFrame = Dataset[Row]
}
因此当我们从1.6.x迁移到2.0.0的时候, 无需任何修改就直接用上了DataSet.

下面是一段DataSet的示例代码

import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object Test {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("test").setMaster("local") // 调试的时候一定不要用local[*]
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._

    val idAgeRDDRow = sc.parallelize(Array(Row(1, 30), Row(2, 29), Row(4, 21)))

    val schema = StructType(Array(StructField("id", DataTypes.IntegerType), StructField("age", DataTypes.IntegerType)))

    // 在2.0.0-preview中这行代码创建出的DataFrame, 其实是DataSet[Row]
    val idAgeDS = sqlContext.createDataFrame(idAgeRDDRow, schema)

    // 在2.0.0-preview中, 还不支持自定的Encoder, Row类型不行, 自定义的bean也不行
    // 官方文档也有写通过bean创建Dataset的例子,但是我运行时并不能成功
    // 所以目前需要用创建DataFrame的方法, 来创建DataSet[Row]
    // sqlContext.createDataset(idAgeRDDRow)

    // 目前支持String, Integer, Long等类型直接创建Dataset
    Seq(1, 2, 3).toDS().show()
    sqlContext.createDataset(sc.parallelize(Array(1, 2, 3))).show()
  }
}

5. RDD和DataFrame比较

这里写图片描述

DataFrame与RDD相同之处,都是不可变分布式弹性数据集。不同之处在于,DataFrame的数据集都是按指定列存储,即结构化数据。类似于传统数据库中的表。
DataFrame的设计是为了让大数据处理起来更容易。DataFrame允许开发者把结构化数据集导入DataFrame,并做了higher-level的抽象; DataFrame提供特定领域的语言(DSL)API来操作你的数据集。

上图直观地体现了DataFrame和RDD的区别。左侧的RDD[Person]虽然以Person为类型参数,但Spark框架本身不了解Person类的内部结构。而右侧的DataFrame却提供了详细的结构信息,使得Spark SQL可以清楚地知道该数据集中包含哪些列,每列的名称和类型各是什么。DataFrame多了数据的结构信息,即schema。RDD是分布式的Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外,更重要的特点是提升执行效率、减少数据读取以及执行计划的优化,比如filter下推、裁剪等。

6. RDD和DataSet比较

DataSet以Catalyst逻辑执行计划表示,并且数据以编码的二进制形式被存储,不需要反序列化就可以执行sorting、shuffle等操作。

DataSet创立需要一个显式的Encoder,把对象序列化为二进制,可以把对象的scheme映射为Spark SQl类型,然而RDD依赖于运行时反射机制。

通过上面两点,DataSet的性能比RDD的要好很多

7. DataFrame和DataSet比较

Dataset可以认为是DataFrame的一个特例,主要区别是Dataset每一个record存储的是一个强类型值而不是一个Row。因此具有如下三个特点:
1.DataSet可以在编译时检查类型
2.是面向对象的编程接口。用wordcount举例:
3.后面版本DataFrame会继承DataSet,DataFrame是面向Spark SQL的接口。
DataFrame和DataSet可以相互转化,df.as[ElementType]这样可以把DataFrame转化为DataSet,ds.toDF()这样可以把DataSet转化为DataFrame。
//DataFrame

// Load a text file and interpret each line as a java.lang.String
val ds = sqlContext.read.text("/home/spark/1.6/lines").as[String]
val result = ds
  .flatMap(_.split(" "))               // Split on whitespace
  .filter(_ != "")                     // Filter empty words
  .toDF()                              // Convert to DataFrame to perform aggregation / sorting
  .groupBy($"value")                   // Count number of occurences of each word
  .agg(count("*") as "numOccurances")
  .orderBy($"numOccurances" desc)      // Show most common words first

//DataSet,完全使用scala编程,不要切换到DataFrame

val wordCount =
  ds.flatMap(_.split(" "))
    .filter(_ != "")
    .groupBy(_.toLowerCase()) // Instead of grouping on a column expression (i.e. $"value") we pass a lambda function
    .count()

8. 应用场景

什么时候用RDD?使用RDD的一般场景:
你需要使用low-level的transformation和action来控制你的数据集;
你得数据集非结构化,比如,流媒体或者文本流;
你想使用函数式编程来操作你得数据,而不是用特定领域语言(DSL)表达;
你不在乎schema,比如,当通过名字或者列处理(或访问)数据属性不在意列式存储格式;
你放弃使用DataFrame和Dataset来优化结构化和半结构化数据集
RDD在Apache Spark 2.0中惨遭抛弃?
答案当然是 NO !
通过后面的描述你会得知:Spark用户可以在RDD,DataFrame和Dataset三种数据集之间无缝转换,而是只需使用超级简单的API方法。
什么时候使用DataFrame或者Dataset?
你想使用丰富的语义,high-level抽象,和特定领域语言API,那你可DataFrame或者Dataset;
你处理的半结构化数据集需要high-level表达, filter,map,aggregation,average,sum ,SQL 查询,列式访问和使用lambda函数,那你可DataFrame或者Dataset;
你想利用编译时高度的type-safety,Catalyst优化和Tungsten的code生成,那你可DataFrame或者Dataset;
你想统一和简化API使用跨Spark的Library,那你可DataFrame或者Dataset;
如果你是一个R使用者,那你可DataFrame或者Dataset;
如果你是一个Python使用者,那你可DataFrame或者Dataset;
你可以无缝的把DataFrame或者Dataset转化成一个RDD,只需简单的调用 .rdd:

// select specific fields from the Dataset, apply a predicate
// using the where() method, convert to an RDD, and show first 10
// RDD rows

val deviceEventsDS = ds.select($"device_name", $"cca3", $"c02_level").where($"c02_level" > 1300)
// convert to RDDs and take the first 10 rows

val eventsRDD = deviceEventsDS.rdd.take(10)

网友评论

登录后评论
0/500
评论
wsc449
+ 关注