SparkSQL基本数据抽象RDD/DataFrame/Dataset介绍[附操作代码]

文章目录

  • 1. SparkSQL
    • 1.1 总述
    • 1.2 数据格式
    • 1.3 转化关系
      • 1.3.1 RDD转DataFrame | Dataset
      • 1.3.2 DataFrame转Dataset
      • 1.3.3 DataFrame | Dataset转RDD
      • 1.3.4 Dataset转DataFrame
  • 2. DataFrame 数据导入
    • 2.1 准备工作
      • pom.xml
      • log4j.properties
    • 2.2 RDD转换DataFrame
      • 2.2.1 模式1
      • 2.2.2 模式2

1. SparkSQL

1.1 总述

SparkSQL是Spark在数据处理上的上层抽象。我们知道，Spark在大数据数据仓库管理中替代的是MapReduce，作为数据仓库的执行引擎，Spark相较于MapReduce在数据逻辑抽象和物理计算层面上最大的不同在于：

  1. Spark将数据本身管理抽象成为一个个RDD[Resiliant Distributed Dataset]，方便Spark进程在内存运算中进行分配、懒加载、shuffle等操作的管理，同时也方便数据在集群中以更大的粒度进行分发与交换，同时能够在序列化、反序列化与压缩的过程中节省资源与运算时间。这一点上Spark与MapReduce相比可以理解为是C面向过程编程的设计思想与C++面向对象的设计思想之间的差别，Spark将数据封装为对象，虽然二者同样都是对文件进行逐行分解，但是RDD显然对内存的利用效率更高

  2. Spark能够提供上层API记录数据的元信息[metadata]，并为检索这一类信息提供封装好的API。我们知道，大数据存储往往涉及到大量半结构化甚至是非结构化的数据，在Hadoop集群中甚至是Amazon S3云存储服务器中，大量的日志资料与埋点数据都是以大文件的形式进行存储的。分布式文件系统顺序写、追加写的特征使得我们很难对数据进行以行为单位的增删改操作，因此对于源数据的加工将极大依赖于执行引擎。Hive借助MR进行服务的提供，HBase能够借助自身编写的引擎提供部分受限符合SQL规范的DSL，而Spark通过SparkSQL几乎能够在Spark的执行引擎下提供几乎全量SQL规范下的spark SQL DSL

1.2 数据格式

Spark SQL 对数据的封装主要体现在三个内置格式上：

  1. RDD
  2. DataFrame = Dataset[Row]
  3. Dataset

搞清楚这三者的关系，实际上只需要牢记：RDD是最为底层的数据管理结构，DataFrame和Dataset都是记录了列关系的数据管理结构

D a t a F r a m e = R D D + S c h e m a DataFrame = RDD+ Schema DataFrame=RDD+Schema
其中Schema是一个StructType对象，StructType记录着所有数据StructField(key,value)的List对象。RDD在DataFrame中常常以case class的形式进行存储，Dataset与DataFrame的不同之处就在于这个case class对于Dataset来说不是一个具体的class而是一个spark内置定义的Row对象，Spark能够根据Row对象中存储的信息动态推断出字段的数据类型
D a t a F r a m e = D a t a s e t [ R o w ] DataFrame = Dataset[Row] DataFrame=Dataset[Row]
因此，DataFrame和Dataset都握有相应的RDD，我们均可以通过二者的无参函数字面量rdd获取相应的RDD对象

1.3 转化关系

∗ * ∗ 注：以下所有代码均默认运行在伪分布式hadoop集群-单机spark模式之下

1.3.1 RDD转DataFrame | Dataset

简单来说，就是 toDF() 以及 toDS() 两个方法,

1.3.2 DataFrame转Dataset

简单来说，就是as[Bean]方法，由于DataFrame会将Bean直接泛化成为Row对象，因此DataFrame转Dataset时需要显式指定Bean的相关类型，而反过来就直接使用 toDF() 即可
这个Bean实际上就是case class

1.3.3 DataFrame | Dataset转RDD

由于DataFrame | Dataset都握有相应的RDD对象，我们只需调用无参函数字面量rdd即可

1.3.4 Dataset转DataFrame

如前所述，使用 toDF() 即可，但是这个操作会丢失掉Bean的相应值而变成Row，当此DataFrame再转换回Dataset时，其Schema将会变为Row对象而不是之前的Bean对象。

2. DataFrame 数据导入

在这里，我们不使用spark-shell进行操作，而是直接通过自定义java程序连接spark集群提交spark任务

2.1 准备工作

pom.xml

首先， 我们需要构建相应的pom文件坐标，需要注意的是，如果我们使用spark连接MySQL，我们需要导入mysql-connector，如果我们需要连接hive，除去导入hive-metastore包外，还要同步导入spark-hive连接包

需要注意的是，由于spark中使用了slf4j的接口包，我们需要同步导入一个slf4j-nop的实现包，日志系统才能够正常运行

最后，为了scala文件能够正常编译，我们在build栏目下同步导入sbt支持包

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>spark-test</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><spark.version>3.5.0</spark.version><scala.version>2.13.8</scala.version><hive.version>3.1.3</hive.version></properties><dependencies><!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.13</artifactId><version>${spark.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.13</artifactId><version>${spark.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.scala-lang/scala-library --><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version><scope>provided</scope></dependency><!-- https://mvnrepository.com/artifact/org.scala-lang/scala-reflect --><dependency><groupId>org.scala-lang</groupId><artifactId>scala-reflect</artifactId><version>${scala.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-nop --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-nop</artifactId><version>2.0.12</version><scope>test</scope></dependency><!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java --><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.27</version></dependency><!-- spark hive compilation --><!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.12</artifactId><version>${hive.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.hive/hive-metastore --><dependency><groupId>org.apache.hive</groupId><artifactId>hive-metastore</artifactId><version>${hive.version}</version></dependency></dependencies><build><finalName>${project.artifactId}</finalName><outputDirectory>target/classes</outputDirectory><testOutputDirectory>target/test-classes</testOutputDirectory><sourceDirectory>src/main/scala</sourceDirectory><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><!--scala原始在sbt(类似java maven)上做开发,现可以用这个插件来在maven中进行开发--><groupId>org.scala-tools</groupId><artifactId>maven-scala-plugin</artifactId><version>2.15.2</version><executions><execution><id>scala-compile-first</id><goals><goal>compile</goal></goals><configuration><includes><include>**/*.scala</include></includes><scalaVersion>2.13.8</scalaVersion><args><arg>-target:jvm-1.8</arg></args></configuration></execution></executions></plugin></plugins></build>
</project>

log4j.properties

其次，我们设置log4j

log4j.rootLogger=INFO, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n# Set the default spark-shell log level to ERROR. When running the spark-shell, the
# og level for this class is used to overwrite the root logger’s log level, so that
# the user can have different defaults for the shell and regular Spark apps.log4j.logger.org.apache.spark.repl.Main=INFO
log4j.logger.org.apache.spark.sql=INFO
#Settings to quiet third party logs that are too verboselog4j.logger.org.spark_project.jetty=WARN
log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=WARN
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=WARN
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=WARN
log4j.logger.org.apache.parquet=WARN
log4j.logger.parquet=WARN# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive supportlog4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR

2.2 RDD转换DataFrame

下面介绍两种RDD转换DataFrame的方式

2.2.1 模式1

首先是第一种非Schema模式，这个方法要求我们首先要把RDD处理成为Bean的模式

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}import java.nio.file.Pathsobject WithTextFile {// prepare SparkSession resource service objectval spark: SparkSession = SparkSession.builder.master("local").appName("RDDtoDF").getOrCreateprivate val array: Array[(String, Int)]= Array(("Zhangsan", 19), ("Lisi", 21), ("Hanmeimei", 23))// prepare case class for DataFrame Schemacase class Person(name: String, age: Long)// introduce implicit implementation of transformation: encoderimport spark.implicits._/*** if input data is structured in explicit form* we could create one case class to store schema** @param path* @return*/def structuredToDF(array:Array[(String, Int)]): DataFrame = {val rdd: RDD[Person] = spark.sparkContext.parallelize(array).map(_.split(",")).map(f => Person(f(0), f(1).trim.toInt))rdd.toDF()}

2.2.2 模式2

其次是第二种Schema模式，这个方法要求我们自定义schema，并使用spark.createDataFrame(rdd, schema)方法进行DataFrame的创建

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}import java.nio.file.Pathsobject WithTextFile {// prepare SparkSession resource service objectval spark: SparkSession = SparkSession.builder.master("local").appName("RDDtoDF").getOrCreateprivate val array: Array[(String, Int)]= Array(("Zhangsan", 19), ("Lisi", 21), ("Hanmeimei", 23))// prepare case class for DataFrame Schemacase class Person(name: String, age: Long)// introduce implicit implementation of transformation: encoderimport spark.implicits._/*** if input data should be transferred into more complex form* we need to provide schema for its loading process** @param path* @return*/def customizedToDF(array:Array[(String, Int)]): DataFrame = {// schema should be like:// name:String, age:intval schema: StructType = StructType(List(StructField("name", StringType),StructField("age", IntegerType)))val rdd = spark.sparkContext.parallelize(array).map(_.split(",")).map(f => Row(f(0).toString.trim, f(1).toString.trim.toInt))spark.createDataFrame(rdd, schema)}

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

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

相关文章

设计模式，装修模式，Php代码演示，优缺点，注意事项

设计模式，装修模式，Php代码演示，优缺点，注意事项

装饰模式&#xff08;Decorator Pattern&#xff09;是一种结构型设计模式&#xff0c;它允许动态地向一个现有对象添加新的功能或行为&#xff0c;而不改变其原始结构。在 PHP 中&#xff0c;可以使用类的继承和组合来实现装饰模式。下面是一个简单的 PHP 装饰模式示例代码&am…
阅读更多...
Android MediaPlayer

Android MediaPlayer

MediaPlayer 类是媒体框架最重要的组成部分之一。此类的对象能够获取、解码以及播放音频和视频&#xff0c;而且只需极少量设置。它支持多种不同的媒体源&#xff0c;例如&#xff1a; • 本地资源 • 内部 URI&#xff0c;例如您可能从内容解析器那获取的 URI • 外部网址…
阅读更多...
debian12，linux-image-6.6.13+bpo-amd64内核nvidia显卡驱动失效

debian12，linux-image-6.6.13+bpo-amd64内核nvidia显卡驱动失效

问题 更新linux内核linux-image-6.6.13bpo-amd64和linux-headers-6.6.13bpo-common后无法进入图形化界面&#xff0c;nvidia驱动版本535.154.05&#xff0c;nvidia官方驱动 日志 /var/log/Xorg.1.log [ 3.834] (--) Log file renamed from "/var/log/Xorg.pid-11…
阅读更多...
QA：ubuntu22.04.4桌面版虚拟机鼠标丢失的解决方法

QA：ubuntu22.04.4桌面版虚拟机鼠标丢失的解决方法

前言 在Windows11中的VMWare Workstation17.5.1 Pro上安装了Ubuntu22.04.4&#xff0c;在使用过程中发现&#xff0c;VM虚拟机的鼠标的光标会突然消失&#xff0c;但鼠标其他正常&#xff0c;就是光标不见了&#xff0c;下面是解决办法。 内容 如下图&#xff0c;输入mouse&a…
阅读更多...
【智能算法】天鹰优化算法（AO）原理及实现

【智能算法】天鹰优化算法（AO）原理及实现

目录 1.背景2.算法原理2.1算法思想2.2算法过程 3.结果展示4.参考文献 1.背景 2021年&#xff0c;L Abualigah等人受到天鹰猎食过程启发&#xff0c;提出了天鹰优化算法&#xff08;Aquila Optimizer&#xff0c;AO&#xff09;。 2.算法原理 2.1算法思想 AO模拟天鹰 4 种不…
阅读更多...
Vue系列——数据对象

Vue系列——数据对象

<!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta name"viewport" content"widthdevice-width, initial-scale1.0"><title>el:挂载点</title> </head> <body&g…
阅读更多...
表情生成器微信小程序版

表情生成器微信小程序版

1.纯前端输出&#xff0c;无需后台&#xff0c;无需域名&#xff0c;速度杠杠快&#xff01; 2.完美支持微信端和抖音端&#xff1b; 3.双端均支持配置开启流量主广告&#xff0c;包括&#xff1a;激励视频广告、插屏广告、banner广告、原生广告、封面广告等&#xff1b; 4.…
阅读更多...
刷题日记——国家名称排序

刷题日记——国家名称排序

7.国家名称排序 分析 一开始打算用二维的字符数组来操作&#xff0c;但是数组指针玩不太明白&#xff0c;于是改用结构体&#xff0c;结构体country里面仅一个成员name&#xff08;字符数组&#xff09;&#xff0c;这样就有两种解题方法&#xff1a; 方法一&#xff1a;使用…
阅读更多...
SpringBoot 整合Redis第1篇

SpringBoot 整合Redis第1篇

SpringBoot是一个开发框架&#xff0c;Redis是一个高性能的键值存储数据库&#xff0c; 常用于缓存、会话管理、消息队列等应用场景。 定义 Redis是什么&#xff1f; 它是一个存储层级&#xff0c; 在实际项目中&#xff0c;位于关系数据库之上&#xff0c; 类似Android分为5…
阅读更多...
ubuntu2204配置zabbix6.4高可用

ubuntu2204配置zabbix6.4高可用

zabbix6.4-HA 配置keepalived配置haproxy数据库高可用配置zabbix-server配置proxy配置客户端agent 本实验VMware搭建zabbix6.4高可用集群&#xff0c;搭配haproxykeepalived。 master&#xff0c;node节点搭建haproxykeepalibed主备并配置vip地址 三台控制节点搭建数据库高可用…
阅读更多...
什么是计算机视觉？计算机视觉：从基础到前沿

什么是计算机视觉？计算机视觉：从基础到前沿

引言 计算机视觉&#xff0c;作为人工智能的一个重要分支&#xff0c;致力于赋予机器“看”的能力——即从图像或视频中理解和解释视觉信息的能力。这项技术的发展为自动驾驶汽车、面部识别、机器人导航等多种应用开辟了道路&#xff0c;正在逐步改变我们的工作和生活方式。本…
阅读更多...
相关性差异性分析

相关性差异性分析

相关性差异性分析是一种用于比较不同组之间相关性差异的统计分析方法。在这种分析中&#xff0c;我们通常会比较两个或多个群体、样本或实验条件之间的相关性程度是否存在显著差异。 例如&#xff0c;假设我们有两组数据&#xff0c;分别代表两种不同的治疗方法对患者病情恢复…
阅读更多...
人工智能-线性回归学习笔记

人工智能-线性回归学习笔记

导数 先了解导数&#xff0c;也就是梯度 解释梯度&#xff1a;yf(x) x2(随机取一个值)&#xff0c;然后x变化了△x(趋向于0)&#xff0c;y值变化了△y&#xff0c;△y除以△x 就是x2的梯度。当x等于一个值&#xff0c;对应的梯度为0&#xff0c; 代表了y到了该函数的极…
阅读更多...
Linux内核之Binder驱动container_of进阶用法(三十四)

Linux内核之Binder驱动container_of进阶用法(三十四)

简介&#xff1a; CSDN博客专家&#xff0c;专注Android/Linux系统&#xff0c;分享多mic语音方案、音视频、编解码等技术&#xff0c;与大家一起成长&#xff01; 优质专栏&#xff1a;Audio工程师进阶系列【原创干货持续更新中……】&#x1f680; 优质专栏&#xff1a;多媒…
阅读更多...
小小狠招:巧妙使用HANA数据库的jdbc driver

小小狠招:巧妙使用HANA数据库的jdbc driver

SAP旗下的HANA数据库&#xff0c;实际上是分为两个系列进行发布&#xff0c;一种是基于本地部署的称之为HANA Platform。另一种是面向Cloud平台的&#xff0c;称之为HANA Cloud。 在实际使用当用&#xff0c;因为两者基本上共用同一代码库&#xff0c;除个别地方略有差异以外&…
阅读更多...
车载电子电器架构 —— 电气架构释放检查

车载电子电器架构 —— 电气架构释放检查

车载电子电器架构 —— 电气架构释放检查 我是穿拖鞋的汉子，魔都中坚持长期主义的汽车电子工程师。 老规矩，分享一段喜欢的文字，避免自己成为高知识低文化的工程师： 屏蔽力是信息过载时代一个人的特殊竞争力，任何消耗你的人和事，多看一眼都是你的不对。非必要不费力证明…
阅读更多...
Sql注入---基础

Sql注入---基础

文章目录 前言一、pandas是什么&#xff1f;二、使用步骤 1.引入库2.读入数据总结 一.Sql注入概述 攻击者通过构造恶意的SQL查询语句&#xff0c;将其注入到应用程序的数据库查询中&#xff0c;以执行未经授权的操作或者获取敏感信息。 假设如下场景&#xff0c;当你想要知道对…
阅读更多...
【c++】简单的日期计算器

【c++】简单的日期计算器

&#x1f525;个人主页&#xff1a;Quitecoder &#x1f525;专栏&#xff1a;c笔记仓 朋友们大家好啊&#xff0c;在我们学习了默认成员函数后&#xff0c;我们本节内容来完成知识的实践&#xff0c;来实现一个简易的日期计算器 目录 头文件声明函数函数的实现1.全缺省默认构…
阅读更多...
每日一博 - 关于日志记录的最佳实践

每日一博 - 关于日志记录的最佳实践

文章目录 概述选择合适的日志等级打印函数的入参、出参打印日志对象要做判空处理&#xff0c;避免阻断流程推荐使用 Slf4j不用e.printStackTrace()打印日志低级别的日志输出&#xff0c;必须进行日志级别开关判断不打印重复日志打印全部的异常信息&#xff0c;方便定位问题核心…
阅读更多...
cocos2.x =＞ node 属性修改

cocos2.x =＞ node 属性修改

简介 与节点属性相关的几个核心变量_trs、_matrix、_worldMatrix、_localMatDirty、_worldMatDirty。 _trs&#xff1a;存储节点的position、rotation、scale _matrix&#xff1a;存储节点的缩放、位移、旋转三者合一的变化矩陈&#xff08;仿射矩陈&#xff09; _worldMat…
阅读更多...
最新文章

Copyright @ 2022~2023