美文网首页Hadoop大数据Hadoop系
一篇文章搞懂DataSet、DataFrame、RDD-《每日五

一篇文章搞懂DataSet、DataFrame、RDD-《每日五

作者: 老蒙 | 来源:发表于2017-05-31 17:51 被阅读1418次
    《每日五分钟搞定大数据》公众号:大叔据每周不定时更新

    点击看《每日五分钟搞定大数据》完整思维导图以及所有文章目录

    1. 三者共性:

    1、RDD、DataFrame、Dataset全都是spark平台下的分布式弹性数据集,为处理超大型数据提供便利
    2、三者都有惰性机制,执行trainform操作时不会立即执行,遇到Action才会执行
    3、三者都会根据spark的内存情况自动缓存运算,这样即使数据量很大,也不用担心会内存溢出
    4、三者都有partition的概念,如

    var predata=data.repartition(24).mapPartitions{
          PartLine => {
            PartLine.map{
              line =>
                 println(“转换操作”)
                                }
                             }
    } 
    

    这样对每一个分区进行操作时,就跟在操作数组一样,不但数据量比较小,而且可以方便的将map中的运算结果拿出来,如果直接用map,map中对外面的操作是无效的,如

    val rdd=spark.sparkContext.parallelize(Seq(("a", 1), ("b", 1), ("a", 1)))
        var flag=0
        val test=rdd.map{line=>
          println("运行")
          flag+=1
          println(flag)
          line._1
        }
    println(test.count)
    println(flag)
        /**
        运行
        1
        运行
        2
        运行
        3
        3
        0
       * */
    

    不使用partition时,对map之外的操作无法对map之外的变量造成影响
    5、三者有许多共同的函数,如filter,排序等
    6、在对DataFrame和Dataset进行操作许多操作都需要这个包进行支持

    import SparkSession.implicits._
    

    7、DataFrame和Dataset均可使用模式匹配获取各个字段的值和类型,为了提高稳健性,最好后面有一个_通配操作,这里提供了DataFrame一个解析字段的方法
    DataFrame:

    testDF.map{
          case Row(col1:String,col2:Int)=>
            println(col1);println(col2)
            col1
          case _=>
            ""
        }
    

    Dataset:

    case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型
        testDS.map{
          case Coltest(col1:String,col2:Int)=>
            println(col1);println(col2)
            col1
          case _=>
            ""
        }
    

    2. 对比:

    2.1 RDD:

    1、RDD一般和spark mlib同时使用
    2、RDD不支持sparksql操作

    2.2 DataFrame和Dataset相同点:

    1、DataFrame与Dataset一般与spark ml同时使用
    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()
    

    利用这样的保存方式,可以方便的获得字段名和列的对应,而且分隔符(delimiter)可以自由指定

    2.3 DataFrame和Dataset不同点:

    这里主要对比Dataset和DataFrame,因为Dataset和DataFrame拥有完全相同的成员函数,区别只是每一行的数据类型不同
    1)DataFrame也可以叫Dataset[Row],dataframe每一行的类型是Row(不解析的话无法得知每一行的字段名和对应的字段类型)
    拿出dataframe行中特定字段的方法有两个:

    getAS方法
    testDF.foreach{
      line =>
        val col1=line.getAs[String]("col1")
        val col2=line.getAs[String]("col2")
    }
    
    模式匹配
    testDF.map{
          case Row(col1:String,col2:Int)=>
            println(col1);println(col2)
            col1
          case _=>
            ""
        }
    

    2) 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]就能比较好的解决问题

    3. 转化:

    RDD、DataFrame、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封装成Row

    import 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又需要针对各个字段处理时极为方便

    参考资料:http://www.cnblogs.com/starwater/p/6841807.html

    公众号:大叔据 。

    评论不能及时回复可直接加公众号提问或交流,知无不答,谢谢 。


    欢迎关注大叔

    相关文章

      网友评论

      本文标题:一篇文章搞懂DataSet、DataFrame、RDD-《每日五

      本文链接:https://www.haomeiwen.com/subject/bjyjfxtx.html