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.mfbz.cn/a/242648.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

深算院YashanDB与长亮科技联合,推出国产数据库金融核心解决方案

近期&#xff0c;深圳计算科学研究院&#xff08;简称“深算院”&#xff09;携手深圳市长亮科技股份有限公司&#xff08;简称“长亮科技”&#xff09;重磅推出基于崖山数据库YashanDB的金融核心解决方案&#xff0c;为推动金融机构实现技术自主可控与数字化转型全面赋能。 …

HarmonyOS(二)—— 初识ArkTS开发语言(下)之ArkTS声明式语法和组件化基础

前言&#xff1a; 通过前面ArkTS开发语言&#xff08;上&#xff09;之TypeScript入门以及ArkTS开发语言&#xff08;中&#xff09;之ArkTS的由来和演进俩文我们知道了ArkTS的由来以及演进&#xff0c;知道了ArkTS具备了声明式语法和组件化特性&#xff0c;今天&#xff0c;搭…

光栅化渲染:可见性问题和深度缓冲区算法

在前面第二章中&#xff0c;我们了解到&#xff0c;在投影点&#xff08;屏幕空间中的点&#xff09;的第三个坐标中&#xff0c;我们存储原始顶点 z 坐标&#xff08;相机空间中点的 z 坐标&#xff09;&#xff1a; 当一个像素与多个三角形重叠时&#xff0c;查找三角形表面上…

【Let‘s Encrypt SSL】使用 acme.sh 给 Nginx 安装 Let’s Encrypt 提供的免费 SSL 证书

安装acme.sh 安装 acme.sh 并设置邮箱用来接受重要通知&#xff0c;如证书快过期未更新通知 curl https://get.acme.sh | sh -s emailmyexample.com执行命令后几秒就安装好了&#xff0c;如果半天没有反应请 CtrlC 后重新执行命令。acme.sh 安装在 ~/.acme.sh 目录下&#xf…

申请Let‘s Encrypt证书

Lets Encrypt官网&#xff1a;https://letsencrypt.org/ Certbot官网&#xff1a;https://certbot.eff.org/ 1、域名控制台添加域名&#xff0c;如test.example.com 2、在服务器添加nginx配置&#xff0c;test.example.com.conf文件内容如下 # test.example.com.conf配置serve…

NCNN 源码学习【三】:数据处理

一、Topic&#xff1a;数据处理 这次我们来一段NCNN应用代码中&#xff0c;除了推理外最重要的一部分代码&#xff0c;数据处理&#xff1a; ncnn::Mat in ncnn::Mat::from_pixels_resize(bgr.data, ncnn::Mat::PIXEL_BGR, bgr.cols, bgr.rows, 227, 227);const float mean_v…

无法打开源文件“opencv2/opencv.hpp“

如图报错&#xff0c;看见就非常高血压 解决方案&#xff1a; 1.打开项目属性 第二步&#xff0c;注意你如果跑的是Debug&#xff0c;那么你在项目属性里面设置的必须选择Debug模式&#xff0c;跑的Release模式&#xff0c;则你必须要设置相应的Release模式&#xff01;否则你…

【解决】Windows 11检测提示电脑不支持 TPM 2.0(注意从DTPM改为PTT)

win11升级&#xff0c;tpm不兼容 写在最前面1. 打开电脑健康状况检查2. 开启tpm3. 微星主板AMD平台开启TPM2.0解决电脑健康状况检查显示可以安装win11&#xff0c;但是系统更新里显示无法更新 写在最前面 我想在台式电脑上用win11的专注模式&#xff0c;但win10不支持 1. 打…

湖仓一体架构理论与实践汇总

湖仓一体架构理论与实践汇总 软件研发本质上属于“手工业”。软件研发在很大程度上还是依赖于个人的能力。当软件规模较小时&#xff0c;依赖“手工业”可以解决问题&#xff0c;但是当软件规模大了之后再依赖“手工业”就不行了。 软件的复杂度包含两个层面&#xff1a;软件…

Git篇---第七篇

系列文章目录 文章目录 系列文章目录前言一、如果分支是否已合并为master,你可以通过什么手段知道?二、 什么是SubGit?三、列举工作中常用的几个git命令?前言 前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家。点击跳转到网站,这篇文…

StatusBar、NavigationBar窗口显示在Activity下面之aosp14窗口类bug线索征集

背景&#xff1a; hi&#xff0c;粉丝朋友们&#xff1a; 从上次帮助国际学员解决了一个分屏有黑屏的bug后&#xff0c;相关blog和解决方法 https://blog.csdn.net/learnframework/article/details/134708393 解决方法看b站视频&#xff1a; https://www.bilibili.com/video/B…

常用的建表但范式、反规范化

规范化&#xff1a; 规范化是用于数据库设计的一系列原理和技术&#xff0c;它可以减少表中数据的冗余&#xff0c;增加数据完整性和一致性。通常有很多范式。 第一范式&#xff08;1NF&#xff09;&#xff1a; 常用的三种范式&#xff1a; 表中的字段都是不可再分割的原子属…

Linux---切换目录命令

1. 切换目录命令的使用 命令说明cd 目录切换到指定目录cd ~切换到当前用户的主目录cd ..切换到上一级目录cd .切换到当前目录cd -切换到上一次目录 注意: cd命令切换目录时&#xff0c;这个目录必须存在。cd 后面不写目录等价于cd ~ cd 目录效果图: cd ~效果图: cd ..效果图…

git 实用命令杂记

使用解决冲突的方式合并&#xff0c;将避免简单的自动合并 git merge origin/dev --strategyresolve清理本地已经合并到 dev 的分支 git branch --merged | grep -v dev | xargs -n 1 git branch -d分支清理 Git 之删除本地无用分支_dearfulan 的博客 - CSDN 博客_git 删除本…

同义词替换论文降重工具 快码论文

大家好&#xff0c;今天来聊聊同义词替换论文降重工具&#xff0c;希望能给大家提供一点参考。 以下是针对论文重复率高的情况&#xff0c;提供一些修改建议和技巧&#xff1a; 标题&#xff1a;同义词替换论文降重工具&#xff1a;原理、应用与优势 一、引言 在学术研究中&am…

数据入表 | 详解数据资产会计核算与企业应对

从2015年《促进大数据发展行动纲要》到2022年《数据20条》到2023年8月份出台了《企业数据资源相关会计处理暂行规定》&#xff0c;可见国家层面对数据的重视和探索如何进一步挖掘数据价值&#xff0c;发挥数据的应用潜力。一石激起千层浪&#xff0c;面对如此重要的规定&#x…

高通平台开发系列讲解(外设篇)高通平台EMMC适配说明

文章目录 一、EMMC的内部框图说明二、EMMC 设备树配置三、EMMC 内核配置四、EMMC 源码沉淀、分享、成长,让自己和他人都能有所收获!😄 📢本篇章主要图解高通平台 EMMC适配说明。 eMMC(嵌入式多媒体卡)是一种集成了闪存存储器和控制器的存储芯片,通常用于嵌入式设备中,…

WPF-一个简单登录界面

一个简单登录界面 文章目录 一个简单登录界面一、效果展示二、准备代码 一、效果展示 二、准备代码 创建一个WPF工程&#xff0c;创建名为 Login5 的WPF项目。 添加Nuget包 MaterialDesignThemes 界面的整体布局和样式代码 <Window x:Class"Login5.MainWindow&quo…

01详解Gateway服务网关的功能,实现,分类.工作流程

Gateway服务网关 网关功能 Gateway网关是是所有微服务的统一入口, 网关的核心功能特性主要体现在请求路由,权限控制,限流三部分 路由: 由于网关不能处理业务,所以网关需要根据某种规则(断言)把请求转发给匹配的主机或者接口上,这个转发的过程就叫做路由负载均衡: 当路由的目…

听觉动态范围是什么?

什么是动态范围&#xff0c;它如何影响您的听力&#xff1f; 简单地说&#xff0c;动态范围就是您可用的听力范围。它是从你能听到的最柔和的声音&#xff08;你的听力阈值&#xff09;到声音非常响亮的地方&#xff08;不舒服的响度水平&#xff09;的分贝&#xff08;dB&…
最新文章