Spark RDD、DataFrame、DataSet比较

news2024/12/23 5:40:21

在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.coloradmin.cn/o/1315095.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

【华为数据之道学习笔记】5-4 数据入湖方式

数据入湖遵循华为信息架构&#xff0c;以逻辑数据实体为粒度入湖&#xff0c;逻辑数据实体在首次入湖时应该考虑信息的完整性。原则上&#xff0c;一个逻辑数据实体的所有属性应该一次性进湖&#xff0c;避免一个逻辑实体多次入湖&#xff0c;增加入湖工作量。 数据入湖的方式…

Codeforces Round 914 (Div. 2) A~E

A.Forked!&#xff08;思维&#xff09; 题意&#xff1a; 给出骑士的跳跃能力 ( x , y ) (x, y) (x,y) 以及国王和皇后的位置&#xff0c;问有多少个位置可以让骑士可以直接攻击到国王和皇后。 分析&#xff1a; 棋盘非常大 ( 1 0 8 1 0 8 ) (10^{8} \times 10^{8}) (1…

基于ssm物流管理系统论文

摘 要 本物流管理系统设计目标是实现物流的信息化管理&#xff0c;提高管理效率&#xff0c;使得物流管理作规范化、科学化、高效化。 本文重点阐述了物流管理系统的开发过程&#xff0c;以实际运用为开发背景&#xff0c;基于SSM框架&#xff0c;运用了Java编程语言和MYSQL数…

C++:命名空间

从今天正式开始对C的学习&#xff0c;这里只学习C对C的拓展&#xff0c;和C相同的部分在C语言专栏中都可以找到&#xff0c;我们先看一段C代码 #include<iostream> using namespace std; int main() {cout<<"hello world<<endl;return 0; } 同样也是打…

【Proteus仿真】【51单片机】电子门铃设计

文章目录 一、功能简介二、软件设计三、实验现象联系作者 一、功能简介 本项目使用Proteus8仿真51单片机控制器&#xff0c;使共阴数码管&#xff0c;按键、无源蜂鸣器等。 主要功能&#xff1a; 系统运行后&#xff0c;数码管默认显示第一种门铃音调&#xff0c;可通过K1键切…

力扣第一题-两数之和[简单]

题目描述 给定一个整数数组 nums 和一个整数目标值 target&#xff0c;请你在该数组中找出 和为目标值 target 的那 两个 整数&#xff0c;并返回它们的数组下标。 你可以假设每种输入只会对应一个答案。但是&#xff0c;数组中同一个元素在答案里不能重复出现。 你可以按任…

陈年葡萄酒里面有什么奥秘?

自古希腊和古罗马以来&#xff0c;陈年葡萄酒就令美食家着迷。很自然&#xff0c;许多人会认为酒越陈越香。只有一个问题:这不是真的。世界上绝大多数的葡萄酒一旦你购买了它们就会在一两年之内被品尝掉。葡萄酒是一种活的食物&#xff0c;当你打开软木塞或拧开金属螺旋盖时&am…

2023年底总结丨5大好用的局域网监控软件

不知不觉间2023年又到结尾了&#xff0c;今年我们服务过很多想要电脑监控软件的客服&#xff0c;也服务了很多想要加密软件的客户。 这一年&#xff0c;我们走得不疾不徐&#xff0c;走得稳而坚定&#xff1b;这一年&#xff0c;我们累积服务超过万计客户&#xff1b;这一年&a…

什么是质量工程?

质量工程是在精益、敏捷、DevOps等实践基础上进一步发展出的系统化赋能软件开发流程、构建既快又好的开发框架的实践。原文: What is Quality Engineering 我们基于古法语"definer"的定义&#xff0c;通过澄清边界、要素和相互作用的方式来构建某个概念&#xff0c;本…

二维差分详解

前言 上一期我们分享了一维差分的使用方法&#xff0c;这一期我们将接着上期的内容带大家了解二位差分的使用方法&#xff0c;话不多说&#xff0c;LET’S GO!&#xff08;上一期链接&#xff09; 二维差分 二维差分我们可以用于对矩阵区间进行多次操作的题。 二维差分我们还…

关于“Python”的核心知识点整理大全23

目录 ​编辑 第&#xff11;0 章 文件和异常 10.1 从文件中读取数据 10.1.1 读取整个文件 pi_digits.txt file_reader.py 10.1.2 文件路径 10.1.3 逐行读取 file_reader.py 10.1.4 创建一个包含文件各行内容的列表 10.1.5 使用文件的内容 pi_string.py 往期快速传…

【从零开始学习--设计模式--装饰者模式】

返回首页 前言 感谢各位同学的关注与支持&#xff0c;我会一直更新此专题&#xff0c;竭尽所能整理出更为详细的内容分享给大家&#xff0c;但碍于时间及精力有限&#xff0c;代码分享较少&#xff0c;后续会把所有代码示例整理到github&#xff0c;敬请期待。 此章节介绍装…

关于“Python”的核心知识点整理大全14

目录 ​编辑 7.2.2 让用户选择何时退出 parrot.py 7.2.3 使用标志 7.2.4 使用 break 退出循环 cities.py 7.2.5 在循环中使用 continue counting.py 7.2.6 避免无限循环 counting.py 7.3 使用 while 循环来处理列表和字典 7.3.1 在列表之间移动元素 confirmed_user…

虚拟网络管理解决方案

随着业务规模的扩大&#xff0c;其网络的规模和复杂性也成比例地扩大&#xff0c;企业级组织和中小型企业需要大规模网络来满足不断增长的业务需求&#xff0c;然而&#xff0c;大规模网络需要大量的物理组件、定期维护和配置&#xff0c;所有这些都是有代价的&#xff0c;因此…

无mac在线申请hbuilderx打包ios证书的方法

hbuilderx是一个跨平台的开发工具&#xff0c;可以开发android和ios的app应用。打包hbuilderx应用需要hbuilderx打包证书。但是很多使用hbuilderx开发的程序员&#xff0c;并没有mac电脑&#xff0c;而申请ios的证书&#xff0c;hbuilderx官网的教程却是需要mac电脑的&#xff…

竞赛保研 opencv 图像识别 指纹识别 - python

0 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 &#x1f6a9; 基于机器视觉的指纹识别系统 &#x1f947;学长这里给一个题目综合评分(每项满分5分) 难度系数&#xff1a;3分工作量&#xff1a;3分创新点&#xff1a;4分 该项目较为新颖&#xff0c;适…

C++面向对象编程(OOP)-浅拷贝与深拷贝

本文主要使用详细的代码比较说明深拷贝和浅拷贝的区别&#xff0c;延伸讲到构造函数。并提供深拷贝和浅拷贝的对比代码。 目录 1 深拷贝和浅拷贝引入原因 2 深拷贝个浅拷贝基本介绍 3 浅拷贝的弊端 4 拷贝构造函数 5 类中有指针的浅拷贝 6 类中有指针的深拷贝 1 深拷贝和浅…

Zabbix监控系统部署与管理

zabbix介绍 zabbix是⼀个基于 Web 界面的提供分布式系统监视以及网络监视功能的企业级的免费开源解决⽅案。zabbix能监视各种⽹络参数&#xff0c;保证服务器系统的安全运营&#xff1b;并提供灵活的通知机制以让系统管理员快速定位/解决存在的各种问题。 zabbix构成 zabbix由…

从纸笔到屏幕:我的CS笔记记录体验分享

前言 三年大学生活里&#xff0c;我花了很多时间在记录笔记上&#xff0c;也因为现有种类繁多的各种学习方式&#xff0c;和朋友一起走了很多弯路。纸笔&#xff0c;OneNote&#xff0c;Typora…… 想总结分享一下自己大学期间的学习笔记记录方式&#xff08;主要针对计算机学…

后端项目全局异常处理-使用RuntimeException自定义异常异常分类简单举例

接上篇&#xff1a;后端项目操作数据库-中枢组件Service调用Mapper 自定义异常&#xff1a; 手动抛出异常&#xff0c;为了后续统一捕获&#xff0c;需要异常自定义&#xff1b; 如&#xff1a;当使用抛出异常的方式表示“操作失败”时&#xff0c;为了后续统一捕获&#xff0c…