RDD、DataFrame、Dataset介绍
2017-08-27 20:33
357 查看
rdd
优点:
编译时类型安全
编译时就能检查出类型错误
面向对象的编程风格
直接通过类名点的方式来操作数据
缺点:
序列化和反序列化的性能开销
无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化.
GC的性能开销
频繁的创建和销毁对象, 势必会增加GC
DataFrame:
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也不是面向对象风格的.
1、RDD和Dataset不同,DataFrame每一行的类型固定为Row,只有通过解析才能获取各个字段的值
3、DataFrame与Dataset支持一些特别方便的保存方式,比如保存成csv,可以带上表头,这样每一列的字段名一目了然
Dataset:
这里主要对比Dataset和DataFrame,因为Dataset和DataFrame拥有完全相同的成员函数,区别只是每一行的数据类型不同
DataFrame也可以叫Dataset[Row],每一行的类型是Row,不解析,每一行究竟有哪些字段,各个字段又是什么类型都无从得知,只能用上面提到的getAS方法或者共性中的第七条提到的模式匹配拿出特定字段
而Dataset中,每一行是什么类型是不一定的,在自定义了case class之后可以很自由的获得每一行的信息
Dataset在需要访问列中的某个字段时是非常方便的,然而,如果要写一些适配性很强的函数时,如果使用Dataset,行的类型又不确定,可能是各种case class,无法实现适配,这时候用DataFrame即Dataset[Row]就能比较好的解决问题
DataSet结合了RDD和DataFrame的优点, 并带来的一个新的概念Encoder
当序列化数据时, Encoder产生字节码与off-heap进行交互, 能够达到按需访问数据的效果, 而不用反序列化整个对象. Spark还没有提供自定义Encoder的API, 但是未来会加入.
下面这段代码, 在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的示例代码
这个转换很简单
val rdd1=testDF.rdd
val rdd2=testDS.rdd
一般用元组把一行的数据写在一起,然后在toDF中指定字段名
可以注意到,定义每一行的类型(case class)时,已经给出了字段名和类型,后面只要往case class里面添加值即可
这个也很简单,因为只是把case class封装成Row
import spark.implicits._
val testDF = testDS.toDF
这种方法就是在给出每一列的类型后,使用as方法,转成Dataset,这在数据类型是DataFrame又需要针对各个字段处理时极为方便
在使用一些特殊的操作时,一定要加上 import spark.implicits._ 不然toDF、toDS无法使用
优点:
编译时类型安全
编译时就能检查出类型错误
面向对象的编程风格
直接通过类名点的方式来操作数据
缺点:
序列化和反序列化的性能开销
无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化.
GC的性能开销
频繁的创建和销毁对象, 势必会增加GC
val sparkconf = new SparkConf().setMaster("local").setAppName("test").set("spark.port.maxRetries","1000") val spark = SparkSession.builder().config(sparkconf).getOrCreate() val rdd=spark.sparkContext.parallelize(Seq(("a", 1), ("b", 1), ("a", 1)))
DataFrame:
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") > "") } }
1、RDD和Dataset不同,DataFrame每一行的类型固定为Row,只有通过解析才能获取各个字段的值
testDF.foreach{ line => val col1=line.getAs[String]("col1") val col2=line.getAs[String]("col2") }2、DataFrame与Dataset均支持sparksql的操作,比如select,groupby之类,还能注册临时表/视窗,进行sql语句操作,如
dataDF.createOrReplaceTempView("tmp") spark.sql("select ROW,DATE from tmp where DATE is not null order by DATE").show(100,false)
3、DataFrame与Dataset支持一些特别方便的保存方式,比如保存成csv,可以带上表头,这样每一列的字段名一目了然
//保存 val saveoptions = Map("header" -> "true", "delimiter" -> "\t", "path" -> "hdfs://172.xx.xx.xx:9000/test") datawDF.write.format("com.databricks.spark.csv").mode(SaveMode.Overwrite).options(saveoptions).save() //读取 val options = Map("header" -> "true", "delimiter" -> "\t", "path" -> "hdfs://172.xx.xx.xx:9000/test") val datarDF= spark.read.options(options).format("com.databricks.spark.csv").load()
Dataset:
这里主要对比Dataset和DataFrame,因为Dataset和DataFrame拥有完全相同的成员函数,区别只是每一行的数据类型不同
DataFrame也可以叫Dataset[Row],每一行的类型是Row,不解析,每一行究竟有哪些字段,各个字段又是什么类型都无从得知,只能用上面提到的getAS方法或者共性中的第七条提到的模式匹配拿出特定字段
而Dataset中,每一行是什么类型是不一定的,在自定义了case class之后可以很自由的获得每一行的信息
case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型 /** rdd ("a", 1) ("b", 1) ("a", 1) * */ val test: Dataset[Coltest]=rdd.map{line=> Coltest(line._1,line._2) }.toDS test.map{ line=> println(line.col1) println(line.col2) }
Dataset在需要访问列中的某个字段时是非常方便的,然而,如果要写一些适配性很强的函数时,如果使用Dataset,行的类型又不确定,可能是各种case class,无法实现适配,这时候用DataFrame即Dataset[Row]就能比较好的解决问题
DataSet结合了RDD和DataFrame的优点, 并带来的一个新的概念Encoder
当序列化数据时, Encoder产生字节码与off-heap进行交互, 能够达到按需访问数据的效果, 而不用反序列化整个对象. Spark还没有提供自定义Encoder的API, 但是未来会加入.
下面这段代码, 在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)
}
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() } }
但是其实却是DataSet, 因为DataFrame被声明为Dataset[Row]
package object sql {
// ...省略了不相关的代码
type DataFrame = Dataset[Row]
}
因此当我们从1.6.x迁移到2.0.0的时候, 无需任何修改就直接用上了DataSet.
下面是一段DataSet的示例代码
DataFrame/Dataset转RDD:
这个转换很简单val rdd1=testDF.rdd
val rdd2=testDS.rdd
RDD转DataFrame:
import spark.implicits._ val testDF = rdd.map {line=> (line._1,line._2) }.toDF("col1","col2")
一般用元组把一行的数据写在一起,然后在toDF中指定字段名
RDD转Dataset:
import spark.implicits._ case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型 val testDS = rdd.map {line=> Coltest(line._1,line._2) }.toDS
可以注意到,定义每一行的类型(case class)时,已经给出了字段名和类型,后面只要往case class里面添加值即可
Dataset转DataFrame:
这个也很简单,因为只是把case class封装成Rowimport spark.implicits._
val testDF = testDS.toDF
DataFrame转Dataset:
import spark.implicits._ case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型 val testDS = testDF.as[Coltest]
这种方法就是在给出每一列的类型后,使用as方法,转成Dataset,这在数据类型是DataFrame又需要针对各个字段处理时极为方便
特别注意:
在使用一些特殊的操作时,一定要加上 import spark.implicits._ 不然toDF、toDS无法使用
相关文章推荐
- spark2.0版本RDD、DataFrame、DataSet介绍
- Spark核心API发展史:RDD、DataFrame、DataSet
- DataFrame、RDD、DataSet的区别
- RDD,DataFrame与DataSet区别
- 进一步理解DataFrame, Dataset, RDD
- Spark视频王家林大神第1课: 30分钟彻底理解Spark核心API发展史:RDD、DataFrame、DataSet
- RDD,DataFrame 和 Dataset区别以及各自用途
- 谈谈RDD、DataFrame、Dataset的区别和各自的优势
- RDD,DataFrame与DataSet区别
- RDD、DataFrame、Dataset的区别和各自的优势
- spark RDD,DataFrame,DataSet 介绍
- RDD 、 DataFrame 和 DataSet 详解
- Spark RDD、DataFrame、DataSet区别和联系
- Spark机器学习第4课及第5课:深入理解RDD、DataFrame、DataSet、Structured Streaming
- spark sql定义RDD、DataFrame与DataSet
- rdd,dataframe,dataset之间的区别
- BigData进阶--细说RDD,DataFrame,Dataset的区别
- RDD、DataFrame、Dataset三者三者之间转换
- RDD,DataFrame与DataSet
- Spark 2.0介绍:从RDD API迁移到DataSet API