Spark（19）：SparkSQL中数据的加载和保存

news2024/9/23 11:17:46

目录

0. 相关文章链接

1. 通用的加载和保存方式

1.1. 加载数据

1.2. 保存数据

2. Parquet

2.1. 加载数据

2.2. 保存数据

3. JSON

4. CSV

5. MySQL

5.1. 导入依赖

5.2. 读取数据

5.3. 写入数据

6. Hive

6.1. SparkSQL连接Hive

6.2. 内嵌的 HIVE

6.3. 外部的 HIVE

6.4. 运行 Spark SQL CLI

6.5. 运行 Spark beeline

6.6. 代码操作Hive

0. 相关文章链接

 Spark文章汇总 

1. 通用的加载和保存方式

        SparkSQL 提供了通用的保存数据和数据加载的方式。这里的通用指的是使用相同的API，根据不同的参数读取和保存不同格式的数据， SparkSQL 默认读取和保存的文件格式为 parquet。

1.1. 加载数据

spark.read.load 是加载数据的通用方法

scala> spark.read.
csv format jdbc json load option options orc parquet schema table text textFile

如果读取不同格式的数据，可以对不同的数据格式进行设定

scala> spark.read.format("…")[.option("…")].load("…")
  • format("…")：指定加载的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"。 
  • load("…")：在"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"格式下需要传入加载数据的路径。 
  • option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable 。

我们前面都是使用 read API 先把文件加载到 DataFrame 然后再查询，其实，我们也可以直接在文件上进行查询: 文件格式.`文件路径`。

scala>spark.sql("select * from json.`/opt/module/data/user.json`").show

1.2. 保存数据

df.write.save 是保存数据的通用方法

scala>df.write.
csv jdbc json orc parquet textFile… …

如果保存不同格式的数据，可以对不同的数据格式进行设定

scala>df.write.format("…")[.option("…")].save("…")
  • format("…")：指定保存的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"。 
  • save ("…")：在"csv"、"orc"、"parquet"和"textFile"格式下需要传入保存数据的路径。 
  • option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable 

保存操作可以使用 SaveMode, 用来指明如何处理数据，使用 mode()方法来设置。有一点很重要: 这些 SaveMode 都是没有加锁的, 也不是原子操作。

SaveMode 是一个枚举类，其中的常量包括：

df.write.mode("append").json("/opt/module/data/output")

2. Parquet

        Spark SQL 的默认数据源为 Parquet 格式。 Parquet 是一种能够有效存储嵌套数据的列式存储格式。数据源为 Parquet 文件时， Spark SQL 可以方便的执行所有的操作，不需要使用 format。修改配置项 spark.sql.sources.default，可修改默认数据源格式。

2.1. 加载数据

scala> val df = spark.read.load("examples/src/main/resources/users.parquet")
scala> df.show

2.2. 保存数据

scala> var df = spark.read.json("/opt/module/data/input/people.json")
//保存为 parquet 格式
scala> df.write.mode("append").save("/opt/module/data/output")

3. JSON

        Spark SQL 能够自动推测 JSON 数据集的结构，并将它加载为一个 Dataset[Row]. 可以通过SparkSession.read.json()去加载 JSON 文件。

注意：Spark 读取的 JSON 文件不是传统的 JSON 文件，每一行都应该是一个 JSON 串。格式如下：

{"name":"Michael"}
{"name":"Andy"， "age":30}
[{"name":"Justin"， "age":19},{"name":"Justin"， "age":19}]
  • 步骤一：导入隐式转换
import spark.implicits._
  • 步骤二：加载 JSON 文件
val path = "/opt/module/spark-local/people.json"
val peopleDF = spark.read.json(path)
  • 步骤三：创建临时表
peopleDF.createOrReplaceTempView("people")
  • 步骤四：数据查询
val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
teenagerNamesDF.show()

+------+
| name |
+------+
|Justin|
+------+

4. CSV

Spark SQL 可以配置 CSV 文件的列表信息，读取 CSV 文件,CSV 文件的第一行设置为数据列，代码如下：

spark
    .read
    .format("csv")
    .option("sep", ";")
    .option("inferSchema","true")
    .option("header", "true")
    .load("data/user.csv")

5. MySQL

        Spark SQL 可以通过 JDBC 从关系型数据库中读取数据的方式创建 DataFrame，通过对DataFrame 一系列的计算后，还可以将数据再写回关系型数据库中。如果使用 spark-shell 操作，可在启动 shell 时指定相关的数据库驱动路径或者将相关的数据库驱动放到 spark 的类路径下。 

bin/spark-shell
    --jars mysql-connector-java-5.1.27-bin.jar

5.1. 导入依赖

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.27</version>
</dependency>

5.2. 读取数据


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

import java.util.Properties

/**
 * @ date: 2023/7/4
 * @ author: yangshibiao
 * @ desc: 项目描述
 */
object BasicModel {
    def main(args: Array[String]): Unit = {

        // 运行环境
        val conf: SparkConf = new SparkConf()
            .setAppName("BasicModel")
            .setMaster("local[*]")
        val spark: SparkSession = SparkSession
            .builder()
            .config(conf)
            .getOrCreate()
        val sc: SparkContext = spark.sparkContext
        import spark.implicits._

        //方式 1：通用的 load 方法读取
        spark
            .read
            .format("jdbc")
            .option("url", "jdbc:mysql://linux1:3306/spark-sql")
            .option("driver", "com.mysql.jdbc.Driver")
            .option("user", "root")
            .option("password", "123123")
            .option("dbtable", "user")
            .load().show

        //方式 2:通用的 load 方法读取 参数另一种形式
        spark
            .read
            .format("jdbc")
            .options(
                Map(
                    "url" -> "jdbc:mysql://linux1:3306/spark-sql?user=root&password=123123",
                    "dbtable" -> "user",
                    "driver" -> "com.mysql.jdbc.Driver"
                )
            )
            .load().show

        //方式 3:使用 jdbc 方法读取
        val props: Properties = new Properties()
        props.setProperty("user", "root")
        props.setProperty("password", "123123")
        spark
            .read
            .jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", props)

        // 关闭资源
        spark.stop()

    }
}

5.3. 写入数据

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

import java.util.Properties

/**
 * @ date: 2023/7/4
 * @ author: yangshibiao
 * @ desc: 项目描述
 */
object BasicModel {
    def main(args: Array[String]): Unit = {

        // 运行环境
        val conf: SparkConf = new SparkConf()
            .setAppName("BasicModel")
            .setMaster("local[*]")
        val spark: SparkSession = SparkSession
            .builder()
            .config(conf)
            .getOrCreate()
        val sc: SparkContext = spark.sparkContext
        import spark.implicits._

        val rdd: RDD[User] = sc.makeRDD(List(User("lisi", 20), User("zs", 30)))
        val ds: Dataset[User] = rdd.toDS

        //方式 1：通用的方式 format 指定写出类型
        ds.write
            .format("jdbc")
            .option("url", "jdbc:mysql://linux1:3306/spark-sql")
            .option("user", "root")
            .option("password", "123123")
            .option("dbtable", "user")
            .mode(SaveMode.Append)
            .save()
        
        //方式 2：通过 jdbc 方法
        val props: Properties = new Properties()
        props.setProperty("user", "root")
        props.setProperty("password", "123123")
        ds
            .write
            .mode(SaveMode.Append)
            .jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", props)

        // 关闭资源
        spark.stop()

    }
}

case class User(name: String, age: Long)

6. Hive

6.1. SparkSQL连接Hive

        Apache Hive 是 Hadoop 上的 SQL 引擎，Spark SQL 编译时可以包含 Hive 支持，也可以不包含。包含 Hive 支持的 Spark SQL 可以支持 Hive 表访问、UDF (用户自定义函数) 以及 Hive 查询语言(HiveQL/HQL)等。需要强调的一点是，如果要在 Spark SQL 中包含 Hive 的库，并不需要事先安装 Hive。一般来说，最好还是在编译 Spark SQL 时引入 Hive 支持，这样就可以使用这些特性了。如果你下载的是二进制版本的 Spark，它应该已经在编译时添加了 Hive 支持。 

        若要把 Spark SQL 连接到一个部署好的 Hive 上，你必须把 hive-site.xml 复制到 Spark 的配置文件目录中($SPARK_HOME/conf)。即使没有部署好 Hive，Spark SQL 也可以运行。 需要注意的是，如果你没有部署好 Hive，Spark SQL 会在当前的工作目录中创建出自己的 Hive 元数据仓库，叫作 metastore_db。此外，如果你尝试使用 HiveQL 中的 CREATE TABLE (并非 CREATE EXTERNAL TABLE)语句来创建表，这些表会被放在你默认的文件系统中的 /user/hive/warehouse 目录中(如果你的 classpath 中有配好的 hdfs-site.xml，默认的文件系统就是 HDFS，否则就是本地文件系统)。 spark-shell 默认是 Hive 支持的；代码中是默认不支持的，需要手动指定（加一个参数即可）。 

6.2. 内嵌的 HIVE

如果使用 Spark 内嵌的 Hive, 则什么都不用做, 直接使用即可；但是注意：在实际使用中, 几乎没有任何人会使用内置的 Hive。

Hive 的元数据存储在 derby 中, 默认仓库地址:$SPARK_HOME/spark-warehouse：

scala> spark.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
+--------+---------+-----------+

scala> spark.sql("create table aa(id int)")

scala> spark.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
| default| aa      | false     |
+--------+---------+-----------+

向表加载本地数据：

scala> spark.sql("load data local inpath 'input/ids.txt' into table aa")

scala> spark.sql("select * from aa").show
+---+
| id|
+---+
| 1 |
| 2 |
| 3 |
| 4 |
+---+

6.3. 外部的 HIVE

如果想连接外部已经部署好的 Hive，需要通过以下几个步骤：

  • Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下
  • 把 Mysql 的驱动 copy 到 jars/目录下
  • 如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下
  • 重启 spark-shell
scala> spark.sql("show tables").show
20/04/25 22:05:14 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
+--------+--------------------+-----------+
|database| tableName          |isTemporary|
+--------+--------------------+-----------+
| default| emp                | false     |
| default|hive_hbase_emp_table| false     |
| default| relevance_hbase_emp| false     |
| default| staff_hive         | false     |
| default| ttt                | false     |
| default| user_visit_action  | false     |
+--------+--------------------+-----------+

6.4. 运行 Spark SQL CLI

Spark SQL CLI 可以很方便的在本地运行 Hive 元数据服务以及从命令行执行查询任务。在Spark 目录下执行如下命令启动 Spark SQL CLI，直接执行 SQL 语句，类似一 Hive 窗口：

bin/spark-sql

6.5. 运行 Spark beeline

        Spark Thrift Server 是 Spark 社区基于 HiveServer2 实现的一个 Thrift 服务。旨在无缝兼容HiveServer2。因为 Spark Thrift Server 的接口和协议都和 HiveServer2 完全一致，因此我们部署好 Spark Thrift Server 后，可以直接使用 hive 的 beeline 访问 Spark Thrift Server 执行相关语句。Spark Thrift Server 的目的也只是取代 HiveServer2，因此它依旧可以和 Hive Metastore进行交互，获取到 hive 的元数据。 

如果想连接 Thrift Server，需要通过以下几个步骤： 

  • Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下
  • 把 Mysql 的驱动 copy 到 jars/目录下
  • 如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下
  • 启动 Thrift Server
sbin/start-thriftserver.sh
  • 使用 beeline 连接 Thrift Server
sbin/start-thriftserver.sh

6.6. 代码操作Hive

  • 导入依赖
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.12</artifactId>
    <version>3.0.0</version>
</dependency>

<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-exec</artifactId>
    <version>1.2.1</version>
</dependency>

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.27</version>
</dependency>
  • 将 hive-site.xml 文件拷贝到项目的 resources 目录中，代码实现
//创建 SparkSession
val spark: SparkSession = SparkSession
    .builder()
    .enableHiveSupport()
    .master("local[*]")
    .appName("sql")
    .getOrCreate()
  • 注意：在开发工具中创建数据库默认是在本地仓库，通过参数修改数据库仓库的地址
config("spark.sql.warehouse.dir", "hdfs://linux1:8020/user/hive/warehouse")
  • 如果在执行操作时，出现如下错误，可以代码最前面增加设置用户解决（此处的 root 改为你们自己的 hadoop 用户名称即可，如果使用的是CDH安装，一般设置为hdfs为root用户，针对hive的目录，一般是为hive用户）：

System.setProperty("HADOOP_USER_NAME", "root")

注：其他Spark相关系列文章链接由此进 ->  Spark文章汇总 

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

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

相关文章

FastDFS文件系统

FastDFS文件系统

FastDFS文件系统 环境代码实现 一.FastDFS 1.什么是FastDFS FastDFS:Fast Distributed File System,快速的分布式文件系统,是一款用c语言开发的开源的分布式文件系统. FastDFS 是用 c 语言编写的一款开源的分布式文件系统。FastDFS为互联网量身定制&#xff0c;充分考虑了…
阅读更多...
前端漏洞xss

前端漏洞xss

网络钓鱼、获取Cookie、强制弹窗获取流量、网站挂马(将恶意代码嵌入程序&#xff0c;用户浏览页面时计算机将被嵌入木马)、发送垃圾信息或广告、传播蠕虫病毒 漏洞原理 XSS(Cross Site Scripting),是一种跨站的脚本攻击&#xff0c;曾简称为CSS&#xff0c; 后改为XSS。 攻击…
阅读更多...
springboot在线考试

springboot在线考试

在线考试系统的开发运用java技术&#xff0c;MIS的总体思想&#xff0c;以及MYSQL等技术的支持下共同完成了该系统的开发&#xff0c;实现了在线考试管理的信息化&#xff0c;使用户体验到更优秀的在线考试管理&#xff0c;管理员管理操作将更加方便&#xff0c;实现目标.
阅读更多...
Dell服务器的iDRAC管理卡连接

Dell服务器的iDRAC管理卡连接

Dell服务器的iDRAC管理卡图文教程 1、网线连接idrac口2、查看idrac地址3、匹配IP地址4、web登录idrac页面5、登录成功页面 带有集成戴尔远程访问控制器 &#xff08;idrac&#xff09;的系统具有默认用户名和密码&#xff0c;但您也可以使用安全密码对其进行配置。默认使用web浏…
阅读更多...
在idea中添加mapper.xml文件模板

在idea中添加mapper.xml文件模板

文章目录 1、创建模板2. 创建mapper.xml文件 1、创建模板 打开创建模板页面 位置&#xff1a;File-》Settings-》Editor-》File and Code Templates 点击号&#xff0c;新建模板&#xff0c;填入Name&#xff08;随意&#xff09;、Extension&#xff08;必须为xml&#xff…
阅读更多...
hadoop安装集群搭建

hadoop安装集群搭建

准备工作&#xff1a; 1&#xff0c;创建一台虚拟机&#xff0c;安装java jdk 这里选择1.8 2&#xff0c;给虚拟机并配置好静态IP地址 一&#xff1a;Java jdk安装 1.1检查安装环境&#xff0c;是否已安装其他版本的jdk&#xff0c;有的话卸载。 # 查看系统JDK进程 [rooth…
阅读更多...
PSINS工具箱学习（二）姿态阵、四元数、欧拉角、等效旋转矢量的概念和转换

PSINS工具箱学习（二）姿态阵、四元数、欧拉角、等效旋转矢量的概念和转换

上一篇博客&#xff1a;PSINS工具箱学习&#xff08;一&#xff09;下载安装初始化、SINS-GPS组合导航仿真、习惯约定与常用变量符号、数据导入转换、绘图显示 文章目录 一、基础概念1、坐标系定义1. 惯性坐标系&#xff08; i 系 &#xff09;2. 地心地固坐标系&#xff08; e…
阅读更多...
solr快速上手：搭建solr集群并创建核心，设置数据同步（十一）

solr快速上手：搭建solr集群并创建核心，设置数据同步（十一）

0. 引言 前几章我们已经讲解了solr单机版的基本使用&#xff0c;但实际生产中&#xff0c;为了保证高可用、高性能&#xff0c;我们一般会采用集群模式&#xff0c;所以接下来&#xff0c;我们继续讲解solr集群的搭建和基本操作 1. 集群模式 1.1 分片 在讲解solr集群模式前…
阅读更多...
【七天入门数据库】第二天 数据库理论基础

【七天入门数据库】第二天 数据库理论基础

系列文章传送门&#xff1a; 【七天入门数据库】第一天 MySQL的安装部署 【七天入门数据库】第二天 数据库理论基础 【七天入门数据库】第三天 MySQL的库表操作 文章目录 一、什么是数据库 二、数据库管理系统DBMS 三、数据库与文件系统的区分 四、数据库的发展史 五、…
阅读更多...
选择屏幕文本元素

选择屏幕文本元素

标题 列标题 当你使用write语句时才能展现使用. 效果: 选择屏幕文本符号
阅读更多...
前端Vue自定义tabs标题栏选项卡组件 可设置文字color

前端Vue自定义tabs标题栏选项卡组件 可设置文字color

随着技术的发展&#xff0c;开发的复杂度也越来越高&#xff0c;传统开发方式将一个系统做成了整块应用&#xff0c;经常出现的情况就是一个小小的改动或者一个小功能的增加可能会引起整体逻辑的修改&#xff0c;造成牵一发而动全身。 通过组件化开发&#xff0c;可以有效实现…
阅读更多...
7.10 qt作业

7.10 qt作业

闹钟 #ifndef WIDGET_H #define WIDGET_H#include <QWidget> #include <QDebug> #include <QIcon> #include <QLabel> #include <QTextEdit> #include <QPushButton> #include <QLineEdit> #include <QPainter> #include <…
阅读更多...
实现临界区互斥访问的基本方法

实现临界区互斥访问的基本方法

1. 问题引入 在我们之前的生产者与消费者问题中, 在文章的最后, 我们曾尝试过把我们的代码封装成P()和V()操作, 结果却以失败告终. 归根结底是因为我们无法在不使用mutex的情况下来完成对临界区的互斥访问, 本篇文章我们就来探讨一下, 如何不使用mutex实现临界区的互斥访问. …
阅读更多...
形态学操作

形态学操作

目录 1、腐蚀 1.1 腐蚀目的 1.2 原理与代码实现 2、膨胀 3、应用 3.1 开闭运算、形态学梯度 3.1.1 开运算 3.1.2 闭运算 ​编辑 3.1.3 形态学梯度 ​编辑 3.1.4 顶帽与黑帽运算 3.2 相关函数 形态学操作常用于对二值化图像的操作 1、腐蚀 1.1 腐蚀目的 去除图像中…
阅读更多...
[工业互联-22]：常见EtherCAT主站方案：Acontis公司的商用Windows 解决方案

[工业互联-22]：常见EtherCAT主站方案：Acontis公司的商用Windows 解决方案

目录 前言&#xff1a;非实时、纯软件解决方案 1.1 概述 1.2 缺点 1.3 实时性思路 方案1&#xff1a;非实时性能的解决方案&#xff1a;etherCAT优化网卡驱动程序 方案2&#xff1a;EtherCAT内核调度模块EcatDrv 方案3&#xff1a;具有硬实时性能的解决方案&#xff1a;…
阅读更多...
【C++】C++创建动态链接库并调用

【C++】C++创建动态链接库并调用

&#x1f60f;★,:.☆(&#xffe3;▽&#xffe3;)/$:.★ &#x1f60f; 这篇文章主要介绍。 学其所用&#xff0c;用其所学。——梁启超 欢迎来到我的博客&#xff0c;一起学习&#xff0c;共同进步。 喜欢的朋友可以关注一下&#xff0c;下次更新不迷路&#x1f95e; 文章目…
阅读更多...
车载蓝牙通信开发之各种协议原理解析

车载蓝牙通信开发之各种协议原理解析

车载蓝牙开发需要考虑到蓝牙协议栈集成、连接管理、电话功能集成、媒体播放控制、数据交换和服务发现、安全性和隐私保护等方面。这对于实现车辆与蓝牙设备之间的无线通信和交互功能非常关键。 使车辆能够与蓝牙设备进行通信和交互的开发过程。 蓝牙协议栈集成&#xff1a; …
阅读更多...
SpringCloud与SpringBoot版本对应关系

SpringCloud与SpringBoot版本对应关系

浏览器访问start.spring.io/actuator/info 出现JSON字符串&#xff0c;再访问json.cn&#xff0c;在里面解析该字符串 如果对应不上&#xff0c;可能会出现很多环境上的坑
阅读更多...
KKRT16 PSI算法

KKRT16 PSI算法

概念介绍 KKRT16 算法是一种基于OT的轻量级隐私求交协议&#xff0c;用于在半诚实敌手存在的情况下对伪随机函数&#xff08;OPRF&#xff09;进行不经意的评估。 在 OPRF 协议中&#xff0c;接收器有一个输入 r r r&#xff1b; 发送方获得输出 s s s&#xff0c;接收方获得…
阅读更多...
【Hello mysql】 mysql的基本查询

【Hello mysql】 mysql的基本查询

Mysql专栏&#xff1a;Mysql 本篇博客简介&#xff1a;介绍mysql的基本查询 mysql的基本查询 create单行插入全列插入多行查询指定列查询插入否则更新 &#xff08;不常用&#xff09;替换 Retrieveselect列全列查询指定列查询查询字段为表达式结果去重 where条件找到英语小于6…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023