Spark RDD、DataFrame、DataSet比较

在Spark的学习当中，RDD、DataFrame、DataSet可以说都是需要着重理解的专业名词概念。尤其是在涉及到数据结构的部分，理解清楚这三者的共性与区别，非常有必要。


RDD，作为Spark的核心数据抽象，是Spark当中不可或缺的存在，而在SparkSQL中，Spark为我们提供了两个新的抽象，分别是DataFrame和DataSet。

RDD、DataFrame、DataSet三者的共性：
RDD、DataFrame、Dataset全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利。
三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算。
三者都会根据spark的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出。
三者都有partition的概念。三者有许多共同的函数，如filter，排序等。

DataFrame、DataSet和RDD有什么区别？

首先从版本的产生上来看：RDD(Spark1.0)—>Dataframe(Spark1.3)—>Dataset(Spark1.6)

RDD：
RDD一般和spark mlib同时使用。
RDD不支持sparksql操作。
DataFrame：
①与RDD和Dataset不同，DataFrame每一行的类型固定为Row，只有通过解析才能获取各个字段的值。
②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的困扰了。
③结构化数据处理非常方便，支持Avro，CSV，Elasticsearch数据等，也支持Hive，MySQL等传统数据表。
④兼容Hive，支持Hql、UDF
有schema和off-heap概念，DataFrame解决了RDD的缺点，但是却丢了RDD的优点。DataFrame不是类型安全的（只有编译后才能知道类型错误），API也不是面向对象风格的。
Dataset：
①DataSet集中了RDD的优点（强类型和可以用强大lambda函数）以及Spark SQL优化的执行引擎。DataSet可以通过JVM的对象进行构建，可以用函数式的转换（map/flatmap/filter）进行多种操作。
②DataSet结合了RDD和DataFrame的优点，并带来的一个新的概念Encoder。DataSet通过Encoder实现了自定义的序列化格式，使得某些操作可以在无需序列化情况下进行。另外Dataset还进行了包括Tungsten优化在内的很多性能方面的优化。
③Dataset<Row>等同于DataFrame（Spark 2.X）

RDD、DataFrame、DataSet的创建：

创建RDD
在Spark中创建RDD的方式主要分为2种：
1.读取内存数据创建RDD
2.读取文件创建RDD
3.通过其他RDD创建RDD

1、读取内存数据创建RDD
读取内存数据创建RDD，Spark主要提供了两个方法：parallelize和makeRDD。
使用makeRDD创建RDD的时候还可以指定分区数量。 

val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("CreateRDD"))
// 从内存中创建RDD，将内存中集合的数据作为处理的数据源
val seq = Seq[Int](elems = 1,2,3,4)
// parallelize方法创建RDD
// val rdd = sc.parallelize(seq)
// makeRDD方法创建RDD
// val rdd = sc.makeRDD(seq)
// 指定分区数量创建RDD
val rdd = sc.makeRDD(seq,3)
rdd.collect().foreach(println)
sc.stop()

2、读取文件创建RDD
读取文件创建RDD，Spark提供了textFile和wholeTextFiles方法：
textFile：以行为单位进行读取数据，
wholeTextFiles：以文件为单位读取数据，读取的结果为元组形式，第一个值为文件路径，第二个值为文件内容。 

val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("Rdd_File"))
// textFile方法读取文件创建RDD
// val rdd = sc.textFile(path = "test.txt")
// textFile方法也是可以指定分区数量的
// val rdd = sc.textFile(path = "test.txt", 3)
// wholeTextFiles方法读取多个文件创建RDD
val rdd = sc.wholeTextFiles(path = "test*.txt")
rdd.collect().foreach(println)
sc.stop()

3、通过其他RDD创建RDD 

val conf: SparkConf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[*]")
val sc = new SparkContext(conf)
val rdd: RDD[String] = sc.textFile("D:\\develop\\workspace\\bigdata2021\\spark2021\\input")
val flatRDD: RDD[String] = rdd.flatMap(_.split(" "))
sc.stop()
 

创建DataFrame

1、通过Seq生成

val spark = SparkSession
  .builder()
  .appName(this.getClass.getSimpleName).master("local")
  .getOrCreate()

val df = spark.createDataFrame(Seq(
  ("ming", 20, 15552211521L),
  ("hong", 19, 13287994007L),
  ("zhi", 21, 15552211523L)
)) toDF("name", "age", "phone")

df.show()

2、读取Json文件生成

json文件内容
{"name":"ming","age":20,"phone":15552211521}
{"name":"hong", "age":19,"phone":13287994007}
{"name":"zhi", "age":21,"phone":15552211523}


    val dfJson = spark.read.format("json").load("/Users/shirukai/Desktop/HollySys/Repository/sparkLearn/data/student.json")
    dfJson.show()

3、读取csv文件生成

csv文件
name,age,phone
ming,20,15552211521
hong,19,13287994007
zhi,21,15552211523

val dfCsv = spark.read.format("csv").option("header", true).load("/Users/shirukai/Desktop/HollySys/Repository/sparkLearn/data/students.csv")
dfCsv.show()

4、通过Json格式的RDD生成（弃用）

    val sc = spark.sparkContext
    import spark.implicits._
    val jsonRDD = sc.makeRDD(Array(
      "{\"name\":\"ming\",\"age\":20,\"phone\":15552211521}",
      "{\"name\":\"hong\", \"age\":19,\"phone\":13287994007}",
      "{\"name\":\"zhi\", \"age\":21,\"phone\":15552211523}"
    ))

    val jsonRddDf = spark.read.json(jsonRDD)
    jsonRddDf.show()

5、通过Json格式的DataSet生成

val jsonDataSet = spark.createDataset(Array(
  "{\"name\":\"ming\",\"age\":20,\"phone\":15552211521}",
  "{\"name\":\"hong\", \"age\":19,\"phone\":13287994007}",
  "{\"name\":\"zhi\", \"age\":21,\"phone\":15552211523}"
))
val jsonDataSetDf = spark.read.json(jsonDataSet)

jsonDataSetDf.show()

6、通过csv格式的DataSet生成

   val scvDataSet = spark.createDataset(Array(
      "ming,20,15552211521",
      "hong,19,13287994007",
      "zhi,21,15552211523"
    ))
    spark.read.csv(scvDataSet).toDF("name","age","phone").show()

7、动态创建schema

    val schema = StructType(List(
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("phone", LongType, true)
    ))
    val dataList = new util.ArrayList[Row]()
    dataList.add(Row("ming",20,15552211521L))
    dataList.add(Row("hong",19,13287994007L))
    dataList.add(Row("zhi",21,15552211523L))
    spark.createDataFrame(dataList,schema).show()

8、通过jdbc创建

    //第八种：读取数据库（mysql）
    val options = new util.HashMap[String,String]()
    options.put("url", "jdbc:mysql://localhost:3306/spark")
    options.put("driver","com.mysql.jdbc.Driver")
    options.put("user","root")
    options.put("password","hollysys")
    options.put("dbtable","user")

    spark.read.format("jdbc").options(options).load().show()

创建Dateset

1、通过createDataset（seq,list,rdd）

import org.apache.spark.SparkContext
import org.apache.spark.sql.{Dataset, SparkSession}

object CreateDataset {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local[4]").appName(this.getClass.getName).getOrCreate()
    //   需要导入隐式转换
    import spark.implicits._

    val sc: SparkContext = spark.sparkContext
    //通过seq创建Dataset
    val seqDs: Dataset[Int] =spark.createDataset(1 to 10)
    //通过list创建Dataset
    val listDs: Dataset[(String, Int)] = spark.createDataset(List(("a",1),("b",2),("c",3)))
    //通过rdd创建Dataset
    val rddDs: Dataset[(String, Int, Int)] = spark.createDataset(sc.parallelize(List(("a",1,2),("b",2,3),("c",3,4))))

    seqDs.show()
    listDs.show()
    rddDs.show()
  }

}

2、通过case class

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

import scala.collection.mutable

object CreateDataSetByCaseClass {

  case class Point(label:String,x:Double,y:Double)
  case class Category(id:Long,name:String)

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local[4]").appName(this.getClass.getName).getOrCreate()
    //   需要导入隐式转换
    import spark.implicits._

    val sc: SparkContext = spark.sparkContext
    //通过Point的样例类创建一个seq，并将它转化为Dataset
    val points: Dataset[Point] = Seq(Point("bar",2.6,3.5),Point("foo",4.0,3.7)).toDS()
    //通过Category的样例类创建一个seq，并将它转化为Dataset
    val categories: Dataset[Category] = Seq(Category(1,"bar"),Category(2,"foo")).toDS()
    //进行join连接,注意这里需要传入三个”=“，这时一个方法
     points.join(categories,points("label")===categories("name")).show()

    //通过Point的样例类创建一个List，并将它转化为Dataset
    val points2: Dataset[Point] = List(Point("bar",2.6,3.5),Point("foo",4.0,3.7)).toDS()
    //通过Category的样例类创建一个List，并将它转化为Dataset
    val categories2: Dataset[Category] = List(Category(1,"bar"),Category(2,"foo")).toDS()
    //进行join连接,注意这里需要传入三个”=“，这时一个方法
    points2.join(categories2,points2("label")===categories2("name")).show()

    //通过Point的样例类创建一个RDD，并将它转化为Dataset
    val points3: Dataset[Point] = sc.parallelize(List(Point("bar",2.6,3.5),Point("foo",4.0,3.7))).toDS()
    //通过Category的样例类创建一个RDD，并将它转化为Dataset
    val categories3: Dataset[Category] = sc.parallelize(List(Category(1,"bar"),Category(2,"foo"))).toDS()
    points3.join(categories3,points3("label")===categories3("name")).show()
  }
}


RDD、DataFrame、DataSet三者之间的转换：

1.RDD与DataFrame转换

（1）toDF方法：将RDD转换为DataFrame；

## 创建RDD
val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1, "ww", 20), (2, "ss", 30), (3, "xx", 40)))
## 指定列名
val df: DataFrame = rdd.toDF("id", "name", "age")
## 不指定列名
val df1: DataFrame = rdd.toDF()
## 展示
df.show()
df1.show()

（2）rdd方法：将DataFrame转换为RDD。

val rowRDD: RDD[Row] = df.rdd
## 输出
rowRDD.collect().foreach(println)

2.DataFrame与DataSet转换

（1）as方法：将DataFrame转换为DataSet，使用 as[] 方法时需要指明数据类型或者采用样例类的方式；

## 引入隐式转换
import spark.implicits._
## 创建样例类(不能创建于main方法中)
case class User(id:Int,name:String,age:Int)
## 指定数据类型
val ds: Dataset[(Int,String,Int)] = df.as[(Int,String,Int)]
## 采用样例类
val ds1: Dataset[User] = df.as[User]
## 展示
ds.show()
ds1.show()

（2）toDF方法：将DataSet转换为DataFrame。

## 转换
val df2: DataFrame = ds.toDF()
## 展示
df2.show()

3.RDD与DataSet转换

（1）toDS方法：将RDD转换为DataSet，使用 toDS() 方法时可以先将数据包装为样例类的形式也可以直接以数据类型输出；

## 通过case将样例类User与数据进行匹配
val ds2: Dataset[User] = rdd.map {
  case (id, name, age) => {
    User(id, name, age)
  }
}.toDS()
## 直接转换
val ds3: Dataset[(Int, String, Int)]rdd.toDS()
## 展示
ds2.show()
ds3.show()

（2）rdd方法：将DataSet转换为RDD

## 转换
val userRDD: RDD[User] = ds1.rdd
## 输出
userRDD.collect().foreach(println)


编程要求
DD 转换成 DataFrame、Dataset： 
1、读取list数据创建 RDD； 
2、将 RDD转换为 DataFrame,并指定列名为("id","name","sex","age")； 
3、将 RDD转换为 DataSet,并以样例类的方式转换。
DataFrame 转换成 RDD、DataSet： 
1、读取staff.josn文件创建 DataFrame； 
2、将 DataFrame转换为 RDD； 
3、将 DataFrame转换为 DataSet。
DataSet 转换成 RDD、DataFrame： 
1、读取staff2.json文件创建 DataSet，并以Staff样例类的方法创建； 
2、将 DataSet转换为 DataFrame； 
3、将 DataSet转换为 RDD。

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, sql}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
object sparkSql_transform {
 
  case class Message()
  def main(args: Array[String]): Unit = {
 
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
    val spark =SparkSession.builder().config(sparkConf).getOrCreate()
    import spark.implicits._
 
    val list = List((202201, "Mark", "female", 23), (202202, "Peter", "female", 24), (202203, "Anna", "male", 21))
 
    val path1 = "/data/workspace/myshixun/step1/data/staff.json"
    val path2 = "/data/workspace/myshixun/step1/data/staff2.json"
 
      /********* Begin *********/
 
    /********* RDD 转换成 DataFrame、DataSet *********/
    // 读取list数据创建RDD
    val rdd:RDD[(Int,String,String,Int)]=spark.sparkContext.makeRDD(list)
 
    // 将RDD转换为DataFrame,指定列名为（"id","name","sex","age"）,并打印输出
    val df:DataFrame=rdd.toDF("id","name","sex","age")
    df.show()
 
    // 将RDD转换为DataSet,以样例类的方式转换,并打印输出
    val ds=rdd.map{line=>Staff(line._1,line._2,line._3,line._4)}.toDS()
    ds.show()
 
    /********* DataFrame 转换成 RDD、DataSet *********/
 
    // 读取staff.josn文件创建DataFrame
    val df1: DataFrame = spark.read.json(path1)
 
    // 将DataFrame转换为RDD,并打印输出
    val rdd1=df1.rdd
    rdd1.collect().foreach(println)
 
    // 将DataFrame转换为DataSet,并打印输出
    val ds1=df1.as[Staff]
    ds1.show()
 
    /********* DataSet 转换成 RDD、DataFrame *********/
    // 读取staff2.json文件创建DataSet，并以Staff样例类的方法创建
    val ds2: Dataset[Staff] = spark.read.json(path2).as[Staff]
    
    // 将DataSet转换为DataFrame，并打印输出
    val df2=ds2.toDF
    df2.show()
 
    // 将DataSet转换为RDD，并打印输出
    val rdd2=ds2.rdd
    rdd2.collect().foreach(println)
   
      /********* End *********/
 
    // TODO 关闭环境
    spark.close()
 
  }
  // Staff样例类
  case class Staff(id: BigInt,name: String,sex: String,age: BigInt) 
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/279281.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

synchronized关键字的使用和原理

synchronized关键字的使用和原理

synchronized关键字的使用和原理 synchronized&#xff1a;对象锁&#xff0c;保证了临界区内代码的原子性&#xff0c;采用互斥的方式让同一时刻至多只有一个线程能持有对象锁&#xff0c;其它线程获取这个对象锁时会阻塞&#xff0c;保证拥有锁的线程可以安全的执行临界区内…
阅读更多...
【SpringBoot篇】Interceptor拦截器 | 拦截器和过滤器的区别

【SpringBoot篇】Interceptor拦截器 | 拦截器和过滤器的区别

文章目录 &#x1f339;概念⭐作用 &#x1f384;快速入门⭐入门案例代码实现 &#x1f6f8;拦截路径&#x1f354;拦截器interceptor和过滤器filter的区别&#x1f386;登录校验 &#x1f339;概念 拦截器&#xff08;Interceptor&#xff09;是一种软件设计模式&#xff0c;…
阅读更多...
【NSX-T】10. 搭建NSX-T环境 —— 使用 BGP 配置 Tier-0 网关

【NSX-T】10. 搭建NSX-T环境 —— 使用 BGP 配置 Tier-0 网关

目录 10.1 创建上行链路网段10.2 创建 Tier-0 网关&#xff08;1&#xff09;设置 Interface 信息&#xff08;2&#xff09;设置 BGP添加 BGP 邻居 &#xff08;3&#xff09;设置 BGP 路由重分发设置路由重新分发 10.3 连接 Tier-0 和 Tier-1 网关10.4 使用网络拓扑验证 Tier…
阅读更多...
《人工智能导论》知识思维导图梳理【第7章节】

《人工智能导论》知识思维导图梳理【第7章节】

文章目录 说明一 专家系统二 机器学习2.1 机器学习定义2.2 机器学习过程2.2.1 工作流程2.2.2 模型评估 2.3 机器学习分类在这里插入图片描述 机器学习部分md内容机器学习1 机器学习定义机器学习是从数据中自动分析获得模型&#xff0c;并利用模型对未知数据进行预测机器学习&am…
阅读更多...
jpa 修改信息拦截

jpa 修改信息拦截

实现目标springbootJPA 哪个人&#xff0c;修改了哪个表的哪个字段&#xff0c;从什么值修改成什么值 import jakarta.persistence.*; import jakarta.servlet.http.HttpServletRequest; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; im…
阅读更多...
嵌入式开发、C++后端开发、C++音视频开发怎么选择？

嵌入式开发、C++后端开发、C++音视频开发怎么选择？

嵌入式开发、C后端开发和C音视频开发的选择问题 近年来&#xff0c;随着互联网和物联网的快速发展&#xff0c;嵌入式开发、C后端开发和C音视频开发等职业领域也逐渐受到广泛关注。 对于有志于从事这些领域的人来说&#xff0c;如何做出选择呢&#xff1f;下面将从前景、薪酬和…
阅读更多...
数据结构 之map/set练习

数据结构 之map/set练习

文章目录 1. 只出现一次的数字算法原理&#xff1a;代码&#xff1a; 2. 随机链表的复制算法原理&#xff1a;代码&#xff1a; 3. 宝石与石头算法原理&#xff1a;代码&#xff1a; 4. 坏键盘打字算法原理&#xff1a;代码&#xff1a; 5. 前K个高频单词算法原理&#xff1a;代…
阅读更多...
kakfa实战指引-实时海量流式数据处理

kakfa实战指引-实时海量流式数据处理

前言 我们最终决定从头开始构建一些东西。我们的想法是&#xff0c;与其专注于保存成堆的数据&#xff0c;如关系数据库、键值存储、搜索索引或缓存&#xff0c;不如专注于将数据视为不断发展和不断增长的流&#xff0c;并围绕这个想法构建一个数据系统——实际上是一个数据架…
阅读更多...
日本科学家们使用AI技术首次创造了世界上的心灵影像

日本科学家们使用AI技术首次创造了世界上的心灵影像

每周跟踪AI热点新闻动向和震撼发展 想要探索生成式人工智能的前沿进展吗&#xff1f;订阅我们的简报&#xff0c;深入解析最新的技术突破、实际应用案例和未来的趋势。与全球数同行一同&#xff0c;从行业内部的深度分析和实用指南中受益。不要错过这个机会&#xff0c;成为AI领…
阅读更多...
圆通速递查询，圆通速递单号查询，一键复制查询好的物流信息

圆通速递查询，圆通速递单号查询，一键复制查询好的物流信息

批量查询圆通速递单号的物流信息&#xff0c;并将查询好的物流信息一键复制出来。 所需工具&#xff1a; 一个【快递批量查询高手】软件 圆通速递单号若干 操作步骤&#xff1a; 步骤1&#xff1a;运行【快递批量查询高手】软件&#xff0c;第一次使用的朋友记得先注册&…
阅读更多...
网站监控/定时任务/网站网址URL状态监控神器

网站监控/定时任务/网站网址URL状态监控神器

源码介绍&#xff1a; 这是一款在线监控网址的源码&#xff0c;对于有多个网站的站长来说还是非常有用的。也可以做为一项收费服务对外提供。这个程序没啥问题&#xff0c;就是UI有点简单&#xff0c;不影响使用。 网站监控/定时任务/网站网址URL状态监控神器&#xff0c;可以…
阅读更多...
【python】Debian安装miniconda、spyder、tushare

【python】Debian安装miniconda、spyder、tushare

1. miniconda 安装 — 动手学深度学习 2.0.0 documentation中有安装Miniconda的一些说明。 Miniconda — miniconda documentation是Miniconda网站&#xff0c;里面也有安装说明。 Debian安装按照linux安装即可&#xff1a; mkdir -p ~/miniconda3 wget https://repo.anaco…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023