当前位置: 首页 > news >正文

Spark【Spark SQL(三)DataSet】

DataSet

         DataFrame 的出现,让 Spark 可以更好地处理结构化数据的计算,但存在一个问题:编译时的类型安全问题,为了解决它,Spark 引入了 DataSet API(DataFrame API 的扩展)。DataSet 是分布式的数据集合,它提供了强类型支持,也就是给 RDD 的每行数据都添加了类型约束。

        在 Spark 2.0 中,DataFrame 和 DataSet 被合并为 DataSet 。DataSet包含 DataFrame 的功能,DataFrame 表示为 DataSet[Row] ,即DataSet 的子集。

三种 API 的选择

        RDD 是DataFrame 和 DataSet 的底层,如果需要更多的控制功能(比如精确控制Spark 怎么执行一条查询),尽量使用 RDD。

        如果希望在编译时获得更高的类型安全,建议使用 DataSet。

        如果想统一简化 Spark 的API ,则使用 DataFrame 和 DataSet。

        基于 DataFrame API 和 DataSet API 开发的程序会被自动优化,开发人员不需要操作底层的RDD API 来手动优化,大大提高了开发效率。但是RDD API 对于非结构化数据的处理有独特的优势(比如文本数据流),而且更方便我们做底层的操作。

DataSet 的创建

1、使用createDataset()方法创建

def main(args: Array[String]): Unit = {//local代表本地单线程模式 local[*]代表本地多线程模式val spark = SparkSession.builder().appName("create dataset").master("local[*]").getOrCreate()//一定要导入它 不然无法创建DataSet对象import spark.implicits._val ds1 = spark.createDataset(1 to 5)ds1.show()val ds2 = spark.createDataset(spark.sparkContext.textFile("data/sql/people.txt"))ds2.show()spark.stop()}

运行结果:

+-----+
|value|
+-----+
|    1|
|    2|
|    3|
|    4|
|    5|
+-----++--------+
|   value|
+--------+
| Tom, 21|
|Mike, 25|
|Andy, 18|
+--------+

2、通过 toDS 方法生成

import org.apache.spark.sql.{Dataset, SparkSession}object DataSetCreate {//case类一定要写到main方法之外case class Person(name:String,age:Int)def main(args: Array[String]): Unit = {//local代表本地单线程模式 local[*]代表本地多线程模式val spark = SparkSession.builder().appName("create dataset").master("local[*]").getOrCreate()//一定要导入 SparkSession对象下的implicitsimport spark.implicits._val data = List(Person("Tom",21),Person("Andy",22))val ds: Dataset[Person] = data.toDS()ds.show()spark.stop()}
}

运行结果:

+----+---+
|name|age|
+----+---+
| Tom| 21|
|Andy| 22|
+----+---+

3、通过DataFrame 转换生成

需要注意:json中的数
object DataSetCreate{case class Person(name:String,age:Long,sex:String)
def main(args: Array[String]): Unit = {//local代表本地单线程模式 local[*]代表本地多线程模式val spark = SparkSession.builder().appName("create dataset").master("local[*]").getOrCreate()import spark.implicits._val df = spark.read.json("data/sql/people.json")val ds = df.as[Person]ds.show()spark.stop()}
}

RDD、DataFrame 和 DataSet 之间的相互转换

RDD <=> DataFrame

  1. RDD 转 DataFrame ,也就是上一篇博客中介绍的两种方法:
    1. 能创建case类,就直接映射出一个RDD[Person],然后调用toDF方法利用反射机制推断RDD模式。
    2. 无法创建case类,就使用编程方式定义RDD模式,使用 createDataFrame(rowRDD,schema) 指定rowRDD:RDD[Row]和schema:StructType。
  2. DataFrame 转 RDD,直接使用 rdd() 方法。
package com.study.spark.core.sqlimport org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}object TransForm {case class Person(name:String,age:Int)  //txt文件age字段可以用Int,但json文件尽量用Longdef main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("transform").master("local[*]").getOrCreate()import spark.implicits._//1.RDD和DataFrame之间互相转换//1.1 创建RDD对象val rdd: RDD[Person] = spark.sparkContext.textFile("data/sql/people.txt").map(_.split(",")).map(attr => Person(attr(0), attr(1).trim.toInt))rdd.foreach(println)/*Person(Andy,18)Person(Tom,21)Person(Mike,25)*///1.2 RDD转DataFrameval df = rdd.toDF()df.show()/*+----+---+|name|age|+----+---+| Tom| 21||Mike| 25||Andy| 18|+----+---+*///1.3 DataFrame转RDDval res: RDD[Row] = df.rdd/*[Andy,18][Tom,21][Mike,25]*/res.foreach(println)spark.stop()}
}

可以看到,RDD[Person]转为DataFrame后,再从DataFrame转回RDD就变成了RDD[Row] 类型了。


RDD <=> DataSet

RDD 和 DataSet 之间的转换比较简单:

  1. RDD 转 DataSet 直接使用case 类(比如Person),然后映射出 RDD[Person] ,直接调用 toDS方法。
  2. DataSet 转 RDD 直接调用 rdd方法即可。

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}object TransForm {case class Person(name:String,age:Int)  //txt文件age字段可以用Int,但json文件尽量用Longdef main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("transform").master("local[*]").getOrCreate()import spark.implicits._//1.RDD和DataSet之间互相转换//1.1 创建RDD对象val rdd: RDD[Person] = spark.sparkContext.textFile("data/sql/people.txt").map(_.split(",")).map(attr => Person(attr(0), attr(1).trim.toInt))rdd.foreach(println)/*Person(Andy,18)Person(Tom,21)Person(Mike,25)*///1.2 RDD转DataSetval ds = rdd.toDS()ds.show()/*+----+---+|name|age|+----+---+| Tom| 21||Mike| 25||Andy| 18|+----+---+*///1.3 DataFrame转RDDval res: RDD[Person] = ds.rddres.foreach(println)/*Person(Andy,18)Person(Tom,21)Person(Mike,25)*/spark.stop()}
}

可以看到,相比RDD和DataFrame互相转换,RDD和DataSet转换的过程中,不会有数据类型的变化,而DataFrame转RDD的过程就会把我们定义的case类转为Row对象。

DataFrame <=> DataSet

  1. DataFrame 转 DataSet 先使用case类,然后直接使用 as[case 类] 方法。
  2. DataSet 转 DataFrame 直接使用 toDF 方法。
import org.apache.spark.sql.{DataFrame, Row, SparkSession}object TransForm {case class Person(name:String,age:Long,sex:String)  //txt文件age字段可以用Int,但json文件尽量用Longdef main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("transform").master("local[*]").getOrCreate()import spark.implicits._//1.DataFrame和DataSet之间互相转换//1.1 创建DataFrame对象val df = spark.read.json("data/sql/people.json")df.show()/*+---+----------+---+|age|      name|sex|+---+----------+---+| 30|   Michael| 男|| 19|      Andy| 女|| 19|    Justin| 男|| 20|Bernadette| 女|| 23|  Gretchen| 女|| 27|     David| 男|| 33|    Joseph| 女|| 27|     Trish| 女|| 33|      Alex| 女|| 25|       Ben| 男|+---+----------+---+*///1.2 DataFrame转DataSetval ds = df.as[Person]ds.show()/*+---+----------+---+|age|      name|sex|+---+----------+---+| 30|   Michael| 男|| 19|      Andy| 女|| 19|    Justin| 男|| 20|Bernadette| 女|| 23|  Gretchen| 女|| 27|     David| 男|| 33|    Joseph| 女|| 27|     Trish| 女|| 33|      Alex| 女|| 25|       Ben| 男|+---+----------+---+*///1.3 DataSet转DataFrameval res: DataFrame = ds.toDF()res.show()/*+---+----------+---+|age|      name|sex|+---+----------+---+| 30|   Michael| 男|| 19|      Andy| 女|| 19|    Justin| 男|| 20|Bernadette| 女|| 23|  Gretchen| 女|| 27|     David| 男|| 33|    Joseph| 女|| 27|     Trish| 女|| 33|      Alex| 女|| 25|       Ben| 男|+---+----------+---+*/spark.stop()}
}

DataSet 实现 WordCount

def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("create dataset").master("local[*]").getOrCreate()import spark.implicits._val res: Dataset[(String, Long)] = spark.read.text("data/word.txt").as[String].flatMap(_.split(" ")).groupByKey(_.toLowerCase).count()res.show()spark.stop()}

运行结果:

|   key|count(1)|
+------+--------+
|  fast|       1|
|    is|       3|
| spark|       2|
|better|       1|
|  good|       1|
|hadoop|       1|
+------+--------+

总结

        剩下来就是不断练习各种DataFrame和DataSet的操作、熟悉各种转换和行动操作。

相关文章:

Spark【Spark SQL(三)DataSet】

DataSet DataFrame 的出现&#xff0c;让 Spark 可以更好地处理结构化数据的计算&#xff0c;但存在一个问题&#xff1a;编译时的类型安全问题&#xff0c;为了解决它&#xff0c;Spark 引入了 DataSet API&#xff08;DataFrame API 的扩展&#xff09;。DataSet 是分布式的数…...

制作立体图像实用软件:3DMasterKit 10.7 Crack

3DMasterKit 软件专为创建具有逼真 3D 和运动效果的光栅图片而设计&#xff1a;翻转、动画、变形和缩放。 打印机、广告工作室、摄影工作室和摄影师将发现 3DMasterKit 是一种有用且经济高效的解决方案&#xff0c;可将其业务扩展到新的维度&#xff0c;提高生成的 3D 图像和光…...

高校 Web 站点网络安全面临的主要的威胁

校园网 Web 站点的主要安全威胁来源于计算机病毒、内部用户恶意攻击和 破坏、内部用户非恶意的错误操作和网络黑客入侵等。 2.1 计算机病毒 计算机病毒是指编制者在计算机程序中插入的破坏计算机功能或者数据&#xff0c; 影响计算机使用并且能够自我复制的一组计算机指令或…...

vue前端解决跨域

1,首先 axios请求&#xff0c;看后端接口路径&#xff0c;http://122.226.146.110:25002/api/xx/ResxxList&#xff0c;所以baseURL地址改成 ‘/api’ let setAxios originAxios.create({baseURL: /api, //这里要改掉timeout: 20000 // request timeout}); export default s…...

【Cicadaplayer】解码线程及队列实现

4.4分支https://github.com/alibaba/CicadaPlayer/blob/release/0.4.4/framework/codec/ActiveDecoder.h对外:送入多个包,获取一个帧 int send_packet(std::unique_ptr<IAFPacket> &packet, uint64_t timeOut) override;int getFrame(std::u...

把文件上传到Gitee的详细步骤

目录 第一步&#xff1a;创建一个空仓库 第二步&#xff1a;找到你想上传的文件所在的地址&#xff0c;打开命令窗口&#xff0c;git init 第三步&#xff1a;git add 想上传的文件 &#xff0c;git commit -m "给这次提交取个名字" 第四步&#xff1a;和咱们在第…...

基于keras中Lenet对于mnist的处理

文章目录 MNIST导入必要的包加载数据可视化数据集查看数据集的分布开始训练画出loss图画出accuracy图 使用数据外的图来测试图片可视化转化灰度图的可视化可视化卷积层的特征图第一层卷积 conv1 和 pool1第二层卷积 conv2 和 pool2 MNIST MNIST&#xff08;Modified National …...

Python爬虫 教程:IP池的使用

前言 嗨喽~大家好呀&#xff0c;这里是魔王呐 ❤ ~! python更多源码/资料/解答/教程等 点击此处跳转文末名片免费获取 一、简介 爬虫中为什么需要使用代理 一些网站会有相应的反爬虫措施&#xff0c;例如很多网站会检测某一段时间某个IP的访问次数&#xff0c;如果访问频率…...

Ansible之playbook剧本

一、playbook概述1.1 playbook 介绍1.2 playbook 组成部分 二、playbook 示例2.1 playbook 启动及检测2.2 实例一2.3 vars 定义、引用变量2.4 指定远程主机sudo切换用户2.5 when条件判断2.6 迭代2.7 Templates 模块1.先准备一个以 .j2 为后缀的 template 模板文件&#xff0c;设…...

unique_ptr的大小探讨

unique_ptr大小和删除器有很大关系&#xff0c;具体区别看如下代码的分析。不要让unique_ptr占用的空间太大&#xff0c;否则不会达到裸指针同样的效果。 #include <iostream> #include <memory> using namespace std;class Widget {int m_x;int m_y;int m_z;publ…...

人工智能TensorFlow PyTorch物体分类和目标检测合集【持续更新】

1. 基于TensorFlow2.3.0的花卉识别 基于TensorFlow2.3.0的花卉识别Android APP设计_基于安卓的花卉识别_lilihewo的博客-CSDN博客 2. 基于TensorFlow2.3.0的垃圾分类 基于TensorFlow2.3.0的垃圾分类Android APP设计_def model_load(img_shape(224, 224, 3)_lilihewo的博客-CS…...

ElementPlus·面包屑导航实现

面包屑导航 使用vue3中的UI框架elementPlus的 <el-breadcrumb> 实现面包屑导航 <template><!-- 面包屑 --><div class"bread-container" ><el-breadcrumb separator">"><el-breadcrumb-item :to"{ path:/ }&quo…...

【项目管理】PM vs PMO 18点区别

导读&#xff1a;项目经理跟PMO主要有哪些区别&#xff1f;首先从定义上了解&#xff0c;然后根据其他维度进行对比分析&#xff0c;基本可以了解这二者的区别&#xff0c;文中罗列18点区别供各位参考。 目录 1、定义 1.1 PMO 1.2 PM 2、两者区别 2.1 ROI 2.2 项目成功率…...

13 Python使用Json

概述 在上一节&#xff0c;我们介绍了如何在Python中使用xml&#xff0c;包括&#xff1a;SAX、DOM、ElementTree等内容。在这一节&#xff0c;我们将介绍如何在Python中使用Json。Json的英文全称为JavaScript Object Notation&#xff0c;中文为JavaScript对象表示法&#xff…...

PDFBOX和ASPOSE.PDF

一、aspose.pdf 文档 https://docs.aspose.com/pdf/java/ 1、按段落分段 /*** docx文本按段分段*/ public static void main(String[] args) {int i 1;try {// 打开文件流FileInputStream file new FileInputStream("I:\\范文.docx");// 创建 Word 文档对象XWPFDo…...

第51节:cesium 范围查询(含源码+视频)

结果示例: 完整源码: <template><div class="viewer"><el-button-group class="top_item"><el-button type=...

YOLOv5改进算法之添加CA注意力机制模块

目录 1.CA注意力机制 2.YOLOv5添加注意力机制 送书活动 1.CA注意力机制 CA&#xff08;Coordinate Attention&#xff09;注意力机制是一种用于加强深度学习模型对输入数据的空间结构理解的注意力机制。CA 注意力机制的核心思想是引入坐标信息&#xff0c;以便模型可以更好地…...

Jmeter系列-阶梯加压线程组Stepping Thread Group详解(6)

前言 tepping Thread Group是第一个自定义线程组但&#xff0c;随着版本的迭代&#xff0c;已经有更好的线程组代替Stepping Thread Group了【Concurrency Thread Group】&#xff0c;所以说Stepping Thread Group已经是过去式了&#xff0c;但还是介绍一下 Stepping Thread …...

图像的几何变换(缩放、平移、旋转)

图像的几何变换 学习目标 掌握图像的缩放、平移、旋转等了解数字图像的仿射变换和透射变换 1 图像的缩放 缩放是对图像的大小进行调整&#xff0c;即 使图像放大或缩小 cv2.resize(src,dsize,fx0,fy0,interpolationcv2.INTER_LINEAR) 参数&#xff1a; src :输入图像dsize…...

计算机网络第四章——网络层(上)

提示&#xff1a;朝碧海而暮苍梧,睹青天而攀白日 文章目录 网络层是路由器的最高层次&#xff0c;通过网络层就可以将各个设备连接到一起&#xff0c;从而实现这两个主机的数据通信和资源共享&#xff0c;之前学的数据链路层和物理层也是将两端连接起来&#xff0c;但是却没有网…...

Leetcode 3576. Transform Array to All Equal Elements

Leetcode 3576. Transform Array to All Equal Elements 1. 解题思路2. 代码实现 题目链接&#xff1a;3576. Transform Array to All Equal Elements 1. 解题思路 这一题思路上就是分别考察一下是否能将其转化为全1或者全-1数组即可。 至于每一种情况是否可以达到&#xf…...

抖音增长新引擎:品融电商,一站式全案代运营领跑者

抖音增长新引擎&#xff1a;品融电商&#xff0c;一站式全案代运营领跑者 在抖音这个日活超7亿的流量汪洋中&#xff0c;品牌如何破浪前行&#xff1f;自建团队成本高、效果难控&#xff1b;碎片化运营又难成合力——这正是许多企业面临的增长困局。品融电商以「抖音全案代运营…...

数据链路层的主要功能是什么

数据链路层&#xff08;OSI模型第2层&#xff09;的核心功能是在相邻网络节点&#xff08;如交换机、主机&#xff09;间提供可靠的数据帧传输服务&#xff0c;主要职责包括&#xff1a; &#x1f511; 核心功能详解&#xff1a; 帧封装与解封装 封装&#xff1a; 将网络层下发…...

unix/linux,sudo,其发展历程详细时间线、由来、历史背景

sudo 的诞生和演化,本身就是一部 Unix/Linux 系统管理哲学变迁的微缩史。来,让我们拨开时间的迷雾,一同探寻 sudo 那波澜壮阔(也颇为实用主义)的发展历程。 历史背景:su的时代与困境 ( 20 世纪 70 年代 - 80 年代初) 在 sudo 出现之前,Unix 系统管理员和需要特权操作的…...

AI+无人机如何守护濒危物种?YOLOv8实现95%精准识别

【导读】 野生动物监测在理解和保护生态系统中发挥着至关重要的作用。然而&#xff0c;传统的野生动物观察方法往往耗时耗力、成本高昂且范围有限。无人机的出现为野生动物监测提供了有前景的替代方案&#xff0c;能够实现大范围覆盖并远程采集数据。尽管具备这些优势&#xf…...

三分算法与DeepSeek辅助证明是单峰函数

前置 单峰函数有唯一的最大值&#xff0c;最大值左侧的数值严格单调递增&#xff0c;最大值右侧的数值严格单调递减。 单谷函数有唯一的最小值&#xff0c;最小值左侧的数值严格单调递减&#xff0c;最小值右侧的数值严格单调递增。 三分的本质 三分和二分一样都是通过不断缩…...

深入理解Optional:处理空指针异常

1. 使用Optional处理可能为空的集合 在Java开发中&#xff0c;集合判空是一个常见但容易出错的场景。传统方式虽然可行&#xff0c;但存在一些潜在问题&#xff1a; // 传统判空方式 if (!CollectionUtils.isEmpty(userInfoList)) {for (UserInfo userInfo : userInfoList) {…...

安卓基础(Java 和 Gradle 版本)

1. 设置项目的 JDK 版本 方法1&#xff1a;通过 Project Structure File → Project Structure... (或按 CtrlAltShiftS) 左侧选择 SDK Location 在 Gradle Settings 部分&#xff0c;设置 Gradle JDK 方法2&#xff1a;通过 Settings File → Settings... (或 CtrlAltS)…...

[QMT量化交易小白入门]-六十二、ETF轮动中简单的评分算法如何获取历史年化收益32.7%

本专栏主要是介绍QMT的基础用法,常见函数,写策略的方法,也会分享一些量化交易的思路,大概会写100篇左右。 QMT的相关资料较少,在使用过程中不断的摸索,遇到了一些问题,记录下来和大家一起沟通,共同进步。 文章目录 相关阅读1. 策略概述2. 趋势评分模块3 代码解析4 木头…...

Async-profiler 内存采样机制解析:从原理到实现

引言 在 Java 性能调优的工具箱中&#xff0c;async-profiler 是一款备受青睐的低开销采样分析器。它不仅能分析 CPU 热点&#xff0c;还能精确追踪内存分配情况。本文将深入探讨 async-profiler 实现内存采样的多种机制&#xff0c;结合代码示例解析其工作原理。 为什么需要内…...