SparkSQL学习03-数据读取与存储

文章目录

    • 1 数据的加载
      • 1.1 方式一：spark.read.format
        • 1.1.1读取json数据
        • 1.1.2 读取jdbc数据
      • 1.2 方式二：spark.read.xxx
        • 1.2.1 读取json数据
        • 1.2.2 读取csv数据
        • 1.2.3 读取txt数据
        • 1.2.4 读取parquet数据
        • 1.2.5 读取orc数据
        • 1.2.6 读取jdbc数据
    • 2 数据的保存
      • 2.1 方式一：spark.write.format
        • 2.1.1 读取orc数据
      • 2.2 方式二：spark.write.xxx
        • 2.2.1 写入到jdbc数据库中

SparkSQL提供了通用的保存数据和数据加载的方式。这里的通用指的是使用相同的API，根据不同的参数读取和保存不用格式的数据。SparkSQL默认读取和保存的文件格式为parquet，parquet是一种能够有效存储嵌套数据的列式存储格式。

1 数据的加载

SparkSQL提供了两种方式可以加载数据

1.1 方式一：spark.read.format

  • spark.read.format读取数据文件格式.load加载数据路径”
  • 数据文件格式包括csv、jdbc、json、orc、parquet和textFile。
  • 需要注意：在读取jdbc时需要在format和load之间添加多个option进行相应的JDBC参数设置【url、user、password.tablename】load中不用传递路经空参数即可
  • 数据源为Parquet文件时，SparkSQL可以方便的执行所有的操作，不需要使用format
1.1.1读取json数据

json数据：
在这里插入图片描述
读取代码：

package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()

    //使用第一种范式加载数据
    var frame: DataFrame = session.read.format("json")
      .load("data/people.json")
    frame.printSchema()
    /**
     * 运行结果：
     root
      |-- age: long (nullable = true)
      |-- height: double (nullable = true)
      |-- name: string (nullable = true)
      |-- province: string (nullable = true)
     */
    frame.show()
    /**
     * 运行结果：
      +---+------+-------+--------+
      |age|height|   name|province|
      +---+------+-------+--------+
      | 10| 168.8|Michael|    广东|
      | 30| 168.8|   Andy|    福建|
      | 19| 169.8| Justin|    浙江|
      | 32| 188.8| 王启峰|    广东|
      | 10| 168.8|   John|    河南|
      | 19| 179.8|   Domu|    浙江|
      +---+------+-------+--------+
     * */
  }
}
1.1.2 读取jdbc数据

读取代码：

package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()
    // 如果读取的JDBC操作（即读取mysql中的数据）
    val frame = session.read.format("jdbc")
            .option("url","jdbc:mysql://localhost:3306/mydb1")
            .option("dbtable","location_info")
            .option("user","root")
            .option("password","123456")
            .load()
    frame.printSchema()
  }
}

1.2 方式二：spark.read.xxx

  • 上述的书写方式太过项，所以SparksQL推出了更加便捷的方式spark.read.xxx加载数据路径”）
  • XXX包括csv、jdbc、json、orc、parquet和text
  • 需要注意：在读取jdbc时方法参数为三个分别为【url、tablename、properties对象】，其中properties对象中存储的是【user，password】
1.2.1 读取json数据

json数据：
在这里插入图片描述
读取代码：

package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()
    
    //【推荐使用】第二种方式进行读取操作
    val frame = session.read.json("data/people.json")
    frame.printSchema()
    /**
    root
     |-- age: long (nullable = true)
     |-- height: double (nullable = true)
     |-- name: string (nullable = true)
     |-- province: string (nullable = true)
     */
    frame.show()
    /**
    +---+------+-------+--------+
    |age|height|   name|province|
    +---+------+-------+--------+
    | 10| 168.8|Michael|    广东|
    | 30| 168.8|   Andy|    福建|
    | 19| 169.8| Justin|    浙江|
    | 32| 188.8| 王启峰|    广东|
    | 10| 168.8|   John|    河南|
    | 19| 179.8|   Domu|    浙江|
    +---+------+-------+--------+ 
     */
   }
 }
1.2.2 读取csv数据

csv数据：
在这里插入图片描述
读取代码：

package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()
    val frame = session.read.csv("data/country.csv")
    frame.printSchema()
    /**
    root
     |-- _c0: string (nullable = true)
     |-- _c1: string (nullable = true)
     |-- _c2: string (nullable = true)
     */
    frame.show()
    /**
    +---+----------------+---+
    |_c0|             _c1|_c2|
    +---+----------------+---+
    |  1|            中国|  1|
    |  2|      阿尔巴尼亚|ALB|
    |  3|      阿尔及利亚|DZA|
    |  4|          阿富汗|AFG|
    |  5|          阿根廷|ARG|
    |  6|阿拉伯联合酋长国|ARE|
    |  7|          阿鲁巴|ABW|
    |  8|            阿曼|OMN|
    |  9|        阿塞拜疆|AZE|
    | 10|        阿森松岛|ASC|
    | 11|            埃及|EGY|
    | 12|      埃塞俄比亚|ETH|
    | 13|          爱尔兰|IRL|
    | 14|        爱沙尼亚|EST|
    | 15|          安道尔|AND|
    | 16|          安哥拉|AGO|
    | 17|          安圭拉|AIA|
    | 18|安提瓜岛和巴布达|ATG|
    | 19|        澳大利亚|AUS|
    | 20|          奥地利|AUT|
    +---+----------------+---+
     */
   }
 }
1.2.3 读取txt数据

txt数据：
在这里插入图片描述
读取代码：

package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()
    val frame = session.read.text("data/dailykey.txt")
    frame.printSchema()
    /**
    root
     |-- value: string (nullable = true)
     * */
    frame.show()
    /**
    +--------------------+
    |               value|
    +--------------------+
    |2018-11-13\ttom\t...|
    |2018-11-13\ttom\t...|
    |2018-11-13\tjohn\...|
    |2018-11-13\tlucy\...|
    |2018-11-13\tlucy\...|
    |2018-11-13\tjohn\...|
    |2018-11-13\tricha...|
    |2018-11-13\tricha...|
    |2018-11-13\tricha...|
    |2018-11-14\ttom\t...|
    |2018-11-14\ttom\t...|
    |2018-11-14\ttom\t...|
    +--------------------+
     * */
    }
  }
1.2.4 读取parquet数据
package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()
    val frame = session.read.parquet("data/users.parquet")
    frame.printSchema()
    /**
    root
     |-- name: string (nullable = true)
     |-- favorite_color: string (nullable = true)
     |-- favorite_numbers: array (nullable = true)
     |    |-- element: integer (containsNull = true)
     */
    frame.show()

    /*
    +------+--------------+----------------+
    |  name|favorite_color|favorite_numbers|
    +------+--------------+----------------+
    |Alyssa|          null|  [3, 9, 15, 20]|
    |   Ben|           red|              []|
    +------+--------------+----------------+
     */
  }
}
1.2.5 读取orc数据
package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()
    val frame = session.read.orc("data/student.orc")
    frame.printSchema()
    /**
    root
     |-- id: string (nullable = true)
     |-- name: string (nullable = true)
     |-- age: string (nullable = true)
     |-- gender: string (nullable = true)
     |-- course: string (nullable = true)
     |-- score: string (nullable = true)
     */
    frame.show()

    /**
    +---+------+---+------+-------+-----+
    | id|  name|age|gender| course|score|
    +---+------+---+------+-------+-----+
    | 12|  张三| 25|    男|chinese|   50|
    | 12|  张三| 25|    男|   math|   60|
    | 12|  张三| 25|    男|english|   70|
    | 12|  李四| 20|    男|chinese|   50|
    | 12|  李四| 20|    男|   math|   50|
    | 12|  李四| 20|    男|english|   50|
    | 12|  王芳| 19|    女|chinese|   70|
    | 12|  王芳| 19|    女|   math|   70|
    | 12|  王芳| 19|    女|english|   70|
    | 13|张大三| 25|    男|chinese|   60|
    | 13|张大三| 25|    男|   math|   60|
    | 13|张大三| 25|    男|english|   70|
    | 13|李大四| 20|    男|chinese|   50|
    | 13|李大四| 20|    男|   math|   60|
    | 13|李大四| 20|    男|english|   50|
    | 13|王小芳| 19|    女|chinese|   70|
    | 13|王小芳| 19|    女|   math|   80|
    | 13|王小芳| 19|    女|english|   70|
    +---+------+---+------+-------+-----+
     */
  }
}
1.2.6 读取jdbc数据
package _02SparkSQL
import java.util.Properties
import org.apache.spark.sql.{DataFrame, SparkSession}

object _06SparkReadData {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder()
      .appName("SparkReadData")
      .master("local[*]").getOrCreate()
    // 读取jdbc文件
    val properties = new Properties()
    properties.put("user","root")
    properties.put("password","123456")
    val frame = session.read.jdbc("jdbc:mysql://localhost:3306/mydb1"
      ,"location-info",properties)
    frame.printSchema()
    frame.show()
  }
}

2 数据的保存

SparkSQL提供了两种方式可以保存数据

2.1 方式一：spark.write.format

  • spark.write.format(“保存数据格式”).mode(“存储格式”).save(“存储数据路径”)
  • 数据文件格式包括csv、jdbc、json、orc、parquet和textFile。
  • 保存数据可以使用SaveMode，用来指明如何处理数据，使用mode()方法来设置
  • SaveMode是一个枚举类，其中的常量包括：
scala/javaAny LanguageMeaning
SaveMode.ErrorifExists(default）“error”(default）如果文件已经存在,则抛出异常
SaveMode.Append“append”如果文件已经存在,则追加
SaveMode.Overwrite“overwrite”如果文件已经存在,则覆盖
SaveMode.Ignore“ignore”如果文件已经存在,则忽略

需要注意：在读取jdbc时需要在format和save之间添加多个option进行相应的JDBC参数设置【url、user、password、tablename】save中不用传递路经空参数即可，可以不用设置mode

数据源为Parquet文件时，SparkSQL可以方便的执行所有的操作，不需要使用format

2.1.1 读取orc数据
package _02SparkSQL
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object _07SparkWriteData {
  def main(args: Array[String]): Unit = {
    //提供SparkSession对象
    val session = SparkSession.builder()
    .appName("SparkWriteData")
    .master("local").getOrCreate()
    //先读取数据
    var frame: DataFrame = session.read.orc("data/student.orc")
    //保存到某个路径下，OWstudent为文件夹，不需要文件名
    frame.write.format("json").mode(SaveMode.Overwrite).save("data/OWstudent")
    session.stop()
  }
}

最后结果为：
在这里插入图片描述

2.2 方式二：spark.write.xxx

上述的书写方式太过繁项，所以SparksQL推出了更加便捷的方式:

  • spark.write.xxx(“保存数据路径”)
  • XXX包括csv、jdbc、json、orc、parquet和text
  • 需要注意：在保存jdbc时方法参数为三个分别为【url、tablename、properties对象】，其中properties对象中存储的是【user,password】
  • mode可以选择性设置
2.2.1 写入到jdbc数据库中
package _02SparkSQL
import java.util.Properties

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object _07SparkWriteData {
  def main(args: Array[String]): Unit = {
    //提供SparkSession对象
    val session = SparkSession.builder()
    .appName("SparkWriteData")
    .master("local").getOrCreate()

    //先读取数据
    var frame: DataFrame = session.read.orc("data/student.orc")
    
    val properties = new Properties()
    properties.put("user","root")
    properties.put("password","123456")
    frame.write.mode(SaveMode.Append)
      .jdbc("jdbc:mysql://localhost:3306/mydb1","student",properties)

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

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

相关文章

较通用web脚手架模板搭建

较通用web脚手架模板搭建

较通用web脚手架模板搭建 从这里开始就接触到以后写项目的思维了。 做一个web开发，那就要层次分明，要有个实现的规划，这通常也是有一个较为通用的模板的。 总的来说：不同的层次有不同的模块，每个模块有必须实现的功…
阅读更多...
【软件测试】定位前后端bug总结+Web/APP测试分析

【软件测试】定位前后端bug总结+Web/APP测试分析

目录：导读 前言一、Python编程入门到精通二、接口自动化项目实战三、Web自动化项目实战四、App自动化项目实战五、一线大厂简历六、测试开发DevOps体系七、常用自动化测试工具八、JMeter性能测试九、总结（尾部小惊喜） 前言 1、Web测试中简单…
阅读更多...
《Python 语音转换简易速速上手小册》第10章 未来趋势和发展方向（2024 最新版）

《Python 语音转换简易速速上手小册》第10章 未来趋势和发展方向（2024 最新版）

文章目录 10.1 语音技术的未来展望10.1.1 基础知识10.1.2 主要案例：语音驱动的虚拟助理案例介绍案例 Demo案例分析10.1.3 扩展案例 1：情感敏感的客服机器人案例介绍案例 Demo案例分析10.1.4 扩展案例 2：多模态智能会议系统案例介绍案例 Demo
阅读更多...
个人博客系统测试

个人博客系统测试

文章目录 一、项目介绍二、测试1. 功能测试2. 自动化测试（1）添加相关依赖（2）新建包并在报下创建测试类（3）亮点及难点 一、项目介绍 个人博客系统采用前后端分离的方法来实现，同时使用了数据库来…
阅读更多...
cmake 项目。qt5升级 qt6 报错 error: “Qt requires a C++17 compiler 已解决

cmake 项目。qt5升级 qt6 报错 error: “Qt requires a C++17 compiler 已解决

日常项目开发中。需要对qt5升级到qt6 做cmake兼容配置，在编译中发现，有c 编译环境 报错 2>C:\Qt\6.5.3\msvc2019_64\include\QtCore/qcompilerdetection.h(1226,1): fatal error C1189: #error: "Qt requires a C17 compiler, and a suitable …
阅读更多...
目标检测卷王YOLO卷出新高度：YOLOv9问世

目标检测卷王YOLO卷出新高度：YOLOv9问世

论文摘要：如今的深度学习方法重点关注如何设计最合适的目标函数，使得模型的预测结果能够最接近真实情况。 同时，必须设计一个适当的架构，可以帮助获取足够的信息进行预测。 现有方法忽略了一个事实，即当输入数据经过逐层特征提取和空间变换时，大量信息将会丢失。 本文将深…
阅读更多...
【JAVA】Tomcat的安装

【JAVA】Tomcat的安装

目录 官网链接 下载安装程序 解压 启动Tomcat 问题 其他文件介绍 官网链接 Apache Tomcat - Welcome!https://tomcat.apache.org/ 下载安装程序 进入官网后，点击如下tomcat 8: 进入tomcat 8后选择zip文件下载 解压 Tomcat是一个基于java实现的“绿色软件…
阅读更多...
golang tun设备创建并监听

golang tun设备创建并监听

golang tun设备创建并监听 linux tun设备文件地址为/dev/net/tun.直接打开即可(关闭文件描述符创建的tun虚拟接口自动注销) fd,err:syscall.Open("/dev/net/tun",syscall.O_RDWR,0640)//关闭 syscall.Close(fd)初始化 配置ip地址启动虚拟网卡 ip addr add xxx.xx…
阅读更多...
2024-02-21 作业

2024-02-21 作业

作业要求： 复习课上内容 //已完成结构体字节对齐，64位没做完的做完，32位重新都做一遍，课上指定2字节对齐的做一遍，自己验证 //已完成两种验证大小端对齐的代码写一遍复习指针内容 //已完成完善顺序表已写出的…
阅读更多...
使用HiveMQ实现Android MQTT

使用HiveMQ实现Android MQTT

MQTT官网：https://mqtt.org/ 百度Android MQTT，或者B站上搜索，发现大多使用https://github.com/eclipse/paho.mqtt.android，这是Eclipse的一个Android MQTT客户端实现库，但是我发现这个库在运行到高版本的手机上时报错…
阅读更多...
Oracle迁移到mysql-表结构的坑

Oracle迁移到mysql-表结构的坑

1.mysql中id自增字段必须是整数类型 id BIGINT AUTO_INCREMENT not null, 2.VARCHAR2改为VARCHAR 3.NUMBER(16)改为decimal(16,0) 4.date改为datetime 5.mysql范围分区必须int格式，不能list类型 ERROR 1697 (HY000): VALUES value for partition …
阅读更多...
【leetcode热题】填充每个节点的下一个右侧节点指针

【leetcode热题】填充每个节点的下一个右侧节点指针

给定一个 完美二叉树 ，其所有叶子节点都在同一层，每个父节点都有两个子节点。二叉树定义如下： struct Node {int val;Node *left;Node *right;Node *next; } 填充它的每个 next 指针，让这个指针指向其下一个右侧节点。如果找不到…
阅读更多...
Linux线程同步（2）死锁与互斥锁

Linux线程同步（2）死锁与互斥锁

死锁（Deadlock）是指两个或两个以上的进程（或线程）在执行过程中，由于竞争资源或者由于彼此通信而造成的一种阻塞的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了…
阅读更多...
【Linux进阶之路】Socket —— “UDP“ “TCP“

【Linux进阶之路】Socket —— “UDP“ “TCP“

文章目录 一、再识网络1. 端口号2. 网络字节序列3.TCP 与 UDP 二、套接字1.sockaddr结构2.UDP1.server端1.1 构造函数1.2 Init1.3 Run 2.客户端1.Linux2.Windows 3.TCP1. 基本接口2. 客户端3. 服务端1.版本12.版本23.版本34.版本4 三、守护进程尾序 一、再识网络 1. 端口号 在…
阅读更多...
RT-Thread 时钟 timer delay 相关

RT-Thread 时钟 timer delay 相关

前言 此处,介绍对delay 时钟 timer 这几部分之间的关联和相关的知识点;本来只是想介绍一下 delay的,但是发现说到delay 不先 提到 先验知识 晶振\时钟\时钟节拍\定时器 好像没法解释透彻,所以就变成了 晶振\时钟\时钟节拍\定时器\delay 的很简单的概括一遍;并附带上能直接运行的…
阅读更多...
【数据结构】链式队列

【数据结构】链式队列

链式队列实现&#xff1a; 1.创建一个空队列 2.尾插法入队 3.头删法出队 4.遍历队列 一、main函数 #include <stdio.h> #include "./3.linkqueue.h" int main(int…
阅读更多...
备考2025年AMC8数学竞赛：2000-2024年AMC8真题练一练

备考2025年AMC8数学竞赛：2000-2024年AMC8真题练一练

我们今天来随机看五道AMC8的真题和解析&#xff0c;对于想了解或者加AMC8美国数学竞赛的孩子来说&#xff0c;吃透AMC8历年真题是备考最科学、最有效的方法之一。 为帮助孩子们更高效地备考&#xff0c;我整理了2000-2004年的全部AMC8真题&#xff0c;并且独家制作了多种在线练…
阅读更多...
Rust通用代码生成器莲花发布红莲尝鲜版二十一发布介绍视频，前端代码生成物大翻新

Rust通用代码生成器莲花发布红莲尝鲜版二十一发布介绍视频，前端代码生成物大翻新

Rust通用代码生成器莲花发布红莲尝鲜版二十一发布介绍视频&#xff0c;前端代码生成物大翻新 Rust通用代码生成器发布了红莲尝鲜版二十一的最新介绍视频&#xff0c;前端代码生成物大翻新。视频请见&#xff1a; Rust通用代码生成器&#xff1a;莲花&#xff0c;红莲尝鲜版二…
阅读更多...
构建生物医学知识图谱from zero to hero （3）：生物医学命名实体识别和链接

构建生物医学知识图谱from zero to hero （3）：生物医学命名实体识别和链接

生物医学实体链接 🤓现在是激动人心的部分。对于NLP和命名实体识别和链接的新手，让我们从一些基础知识开始。命名实体识别技术用于检测文本中的相关实体或概念。例如，在生物医学领域，我们希望在文本中识别各种基因、药物、疾病和其他概念。 生物医学概念提取 在这个例子中…
阅读更多...
爬虫知识--03

爬虫知识--03

数据存mysql import requests from bs4 import BeautifulSoup import pymysql# 链接数据库pymysql conn pymysql.connect(userroot,password"JIAJIA",host127.0.0.1,databasecnblogs,port3306, ) cursor conn.cursor() cursor conn.cursor()# 爬数据 res request…
阅读更多...
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站