大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源：JSON、CSV、JDBC、Hive

devtools/2024/11/14 6:52:31/

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

  • Hadoop（已更完）
  • HDFS（已更完）
  • MapReduce（已更完）
  • Hive（已更完）
  • Flume（已更完）
  • Sqoop（已更完）
  • Zookeeper（已更完）
  • HBase（已更完）
  • Redis （已更完）
  • Kafka（已更完）
  • Spark（正在更新！）

章节内容

上节我们完成了如下的内容：

  • SparkSQL 核心操作
  • Action操作 详细解释+测试案例
  • Transformation操作 详细解释+测试案例

在这里插入图片描述

SQL 语句

总体而言：SparkSQL语HQL兼容；与HQL相比，SparkSQL更简洁。
SparkSQL是Apache Spark框架中的一个模块，专门用于处理结构化和半结构化数据。它提供了对数据进行查询、处理和分析的高级接口。

SparkSQL的核心特点包括：

  • DataFrame API：SparkSQL提供了DataFrame API，它是一种以行和列为结构的数据集，与关系数据库中的表非常相似。DataFrame支持多种数据源，如Hive、Parquet、JSON、JDBC等，可以轻松地将数据导入并进行操作。
  • SQL查询：SparkSQL允许用户通过标准的SQL语法查询DataFrame，这使得数据分析师和工程师可以使用他们熟悉的SQL语言来处理大数据。SparkSQL会自动将SQL查询转换为底层的RDD操作，从而在分布式环境中执行。
  • 与Hive集成：SparkSQL可以与Hive无缝集成，使用Hive的元数据和查询引擎。它支持HiveQL（Hive Query Language）语法，并且能够直接访问Hive中的数据。
  • 性能优化：SparkSQL采用了多种优化技术，如Catalyst查询优化器和Tungsten物理执行引擎。这些优化技术能够自动生成高效的执行计划，提高查询的执行速度。

数据样例

// 数据
1 1,2,3
2 2,3
3 1,2// 需要实现如下的效果
1 1
1 2
1 3
2 2
2 3
3 1
3 2

编写代码

scala">package icu.wzkimport org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.sql.Encoderscase class Info(id: String, tags: String)object SparkSql01 {def main(args: Array[String]): Unit = {val sparkSession = SparkSession.builder().appName("SparkSQLDemo").master("local[*]").getOrCreate()val sc = sparkSession.sparkContextsc.setLogLevel("WARN")val arr = Array("1 1,2,3", "2 2,3", "3 1,2")val rdd: RDD[Info] = sc.makeRDD(arr).map{line => val fields: Array[String] = line.split("\\s+")Info(fields(0), fields(1))}import sparkSession.implicits._implicit val infoEncoder = Encoders.product[Info]val ds: Dataset[Info] = sparkSession.createDataset(rdd)ds.createOrReplaceTempView("t1")sparkSession.sql("""| select id, tag| from t1| lateral view explode(split(tags, ",")) t2 as tag|""".stripMargin).showsparkSession.sql("""| select id, explode(split(tags, ","))| from t1|""".stripMargin).showsparkSession.close()}}

运行测试

控制台输出结果为：

scala">+---+---+
| id|tag|
+---+---+
|  1|  1|
|  1|  2|
|  1|  3|
|  2|  2|
|  2|  3|
|  3|  1|
|  3|  2|
+---+---++---+---+
| id|col|
+---+---+
|  1|  1|
|  1|  2|
|  1|  3|
|  2|  2|
|  2|  3|
|  3|  1|
|  3|  2|
+---+---+

运行结果

运行结果如下图所示：
在这里插入图片描述

输入与输出

SparkSQL 内建支持的数据源包括：

  • Parquet （默认数据源）
  • JSON
  • CSV
  • Avro
  • Images
  • BinaryFiles（Spark 3.0）

简单介绍一下，Parquet 是一种列式存储格式，专门为大数据处理和分析而设计。

  • 列式存储：Parquet 采用列式存储格式，这意味着同一列的数据存储在一起。这样可以极大地提高查询性能，尤其是当查询只涉及少量列时。
  • 高效压缩：由于同一列的数据具有相似性，Parquet 能够更高效地进行压缩，节省存储空间。
  • 支持复杂数据类型：Parquet 支持嵌套的数据结构，包括嵌套列表、映射和结构体，这使得它非常适合处理复杂的、半结构化的数据。
  • 跨平台：Parquet 是一种开放标准，支持多种编程语言和数据处理引擎，包括 Apache Spark、Hadoop、Impala 等。

在这里插入图片描述

Parquet

特点：Parquet是一种列式存储格式，特别适合大规模数据的存储和处理。它支持压缩和嵌套数据结构，因此在存储效率和读取性能方面表现优异。

使用方式：spark.read.parquet(“path/to/data”) 读取Parquet文件；df.write.parquet(“path/to/output”) 将DataFrame保存为Parquet格式。

JSON

特点：JSON是一种轻量级的数据交换格式，广泛用于Web应用程序和NoSQL数据库中。SparkSQL能够解析和生成JSON格式的数据，并支持嵌套结构。

使用方式：spark.read.json(“path/to/data”) 读取JSON文件；df.write.json(“path/to/output”) 将DataFrame保存为JSON格式。

CSV

特点：CSV（逗号分隔值）是最常见的平面文本格式之一，简单易用，但不支持嵌套结构。SparkSQL支持读取和写入CSV文件，并提供了处理缺失值、指定分隔符等功能。

使用方式：spark.read.csv(“path/to/data”) 读取CSV文件；df.write.csv(“path/to/output”) 将DataFrame保存为CSV格式。

Avro

特点：Avro是一种行式存储格式，适合大规模数据的序列化。它支持丰富的数据结构和模式演化，通常用于Hadoop生态系统中的数据存储和传输。

使用方式：spark.read.format(“avro”).load(“path/to/data”) 读取Avro文件；df.write.format(“avro”).save(“path/to/output”) 将DataFrame保存为Avro格式。

ORC

特点：ORC（Optimized Row Columnar）是一种高效的列式存储格式，专为大数据处理而设计，支持高压缩率和快速读取性能。它在存储空间和I/O性能方面表现优越。

使用方式：spark.read.orc(“path/to/data”) 读取ORC文件；df.write.orc(“path/to/output”) 将DataFrame保存为ORC格式。

Hive Tables

特点：SparkSQL能够无缝集成Hive，直接访问Hive元数据，并对Hive表进行查询。它支持HiveQL语法，并能够利用Hive的存储格式和结构。

使用方式：通过spark.sql(“SELECT * FROM hive_table”)查询Hive表；也可以使用saveAsTable将DataFrame写入Hive表。

JDBC/ODBC

特点：SparkSQL支持通过JDBC/ODBC接口连接关系型数据库，如MySQL、PostgreSQL、Oracle等。它允许从数据库读取数据并将结果写回数据库。

使用方式：spark.read.format(“jdbc”).option(“url”, “jdbc:mysql://host/db”).option(“dbtable”, “table”).option(“user”, “username”).option(“password”, “password”).load() 读取数据库表；df.write.format(“jdbc”).option(“url”, “jdbc:mysql://host/db”).option(“dbtable”, “table”).option(“user”, “username”).option(“password”, “password”).save() 将DataFrame写入数据库。

Text Files

特点：SparkSQL可以处理简单的文本文件，每一行被读取为一个字符串。适合用于处理纯文本数据。

使用方式：spark.read.text(“path/to/data”) 读取文本文件；df.write.text(“path/to/output”) 将DataFrame保存为文本格式。

Delta Lake (外部插件)

特点：Delta Lake是一种开源存储层，构建在Parquet格式之上，支持ACID事务、可扩展元数据处理和流批一体的实时数据处理。尽管不是内建的数据源，但它在Spark生态系统中得到了广泛支持。

使用方式：spark.read.format(“delta”).load(“path/to/delta-table”) 读取Delta表；df.write.format(“delta”).save(“path/to/delta-table”) 将DataFrame保存为Delta格式。

测试案例

scala">val df1 =
spark.read.format("parquet").load("data/users.parquet")
// Use Parquet; you can omit format("parquet") if you wish as
it's the default
val df2 = spark.read.load("data/users.parquet")// Use CSV
val df3 = spark.read.format("csv")
.option("inferSchema", "true")
.option("header", "true")
.load("data/people1.csv")// Use JSON
val df4 = spark.read.format("json")
.load("data/emp.json")

此外还支持 JDBC 的方式：

scala">val jdbcDF = sparkSession.read.format("jdbc").option("url", "jdbc:mysql://h122.wzk.icu/spark_test?useSSL=false").option("driver", "com.mysql.jdbc.Driver").option("user", "hive").option("password", "hive@wzk.icu").load()
jdbcDF.show()

访问Hive

在这里插入图片描述

导入依赖

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.12</artifactId><version>${spark.version}</version>
</dependency>

hive-site

需要在项目的 Resource 目录下，新增一个 hive-site.xml
备注：最好使用 metastore service连接Hive，使用直接metastore的方式时，SparkSQL程序会修改Hive的版本信息

<configuration><property><name>hive.metastore.uris</name><value>thrift://h122.wzk.icu:9083</value></property>
</configuration>

编写代码

scala">object AccessHive {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("Demo1").master("local[*]").enableHiveSupport()// 设为true时，Spark使用与Hive相同的约定来编写Parquet数据.config("spark.sql.parquet.writeLegacyFormat", true).getOrCreate()val sc = spark.sparkContextsc.setLogLevel("warn")spark.sql("show databases").showspark.sql("select * from ods.ods_trade_product_info").showval df: DataFrame = spark.table("ods.ods_trade_product_info")df.show()df.write.mode(SaveMode.Append).saveAsTable("ods.ods_trade_product_info_back")spark.table("ods.ods_trade_product_info_back").showspark.close()}
}

http://www.ppmy.cn/devtools/98258.html

相关文章

JVM性能监控工具

JVM性能监控工具

文章目录 JDK的命令行工具jps: 虚拟机进程状况工具jstat&#xff1a;虚拟机统计信息监视工具jinfo&#xff1a;Java配置信息工具jmap&#xff1a;Java内存映像工具jhat&#xff1a;虚拟机堆转储快照分析工具jstack&#xff1a;Java堆栈跟踪工具 可视化工具JConsole启动JConsole…
阅读更多...
2024年最新盘点！这18款CRM最值得推荐！

2024年最新盘点！这18款CRM最值得推荐！

CRM 系统对企业至关重要。它不仅能整合客户信息&#xff0c;助力企业提供个性化服务&#xff0c;提升客户满意度&#xff0c;还能通过销售流程自动化&#xff0c;提高销售效率&#xff0c;实现多渠道沟通整合&#xff01;可以说&#xff0c;CRM系统是企业走上数字化转型道路上不…
阅读更多...
【axios get请求 中文乱码】

【axios get请求 中文乱码】

问题复现 前端请求&#xff1a; company/queryFenByOrgNo?orgNo5&qcNam%D2%BB%C6%DA qcNam 一期 后端接收&#xff1a; CompanyManagementController - 入参 orgNo“5”,qcNamһ&#xfffd;&#xfffd; 问题解决方案 let httpUrl this.httpcompany/queryFenByOrgNo…
阅读更多...
Qt五大核心特性之信号与槽

Qt五大核心特性之信号与槽

前言 Qt 引入信号与槽&#xff08;Signal & Slot&#xff09;机制的主要原因是为了提供一种灵活、松耦合的方式&#xff0c;使对象之间能够进行安全、方便的通信&#xff0c;尤其在事件驱动的编程环境中&#xff08;如 GUI 应用&#xff09;特别有效。 松耦合&#xff1a…
阅读更多...
C语言 | Leetcode C语言题解之第344题反转字符串

C语言 | Leetcode C语言题解之第344题反转字符串

题目&#xff1a; 题解&#xff1a; void swap(char *a, char *b) {char t *a;*a *b, *b t; }void reverseString(char *s, int sSize) {for (int left 0, right sSize - 1; left < right; left, --right) {swap(s left, s right);} }
阅读更多...
C语言 | Leetcode C语言题解之第357题统计各位数字都不同的数字个数

C语言 | Leetcode C语言题解之第357题统计各位数字都不同的数字个数

题目&#xff1a; 题解&#xff1a; int countNumbersWithUniqueDigits(int n) {if (n 0) {return 1;}if (n 1) {return 10;}int ans 10, cur 9;for (int i 0; i < n - 1; i) {cur * 9 - i;ans cur;}return ans; }
阅读更多...
TCP协议中的建立连接机制

TCP协议中的建立连接机制

目录 客户端与服务器间的三次握手 1、关于SYN_RCVD状态 2、关于系统调用listen的第二个参数 3、为什么服务端操作系统内核中的全连接队列不能太长&#xff1f; 4、服务端操作系统内核中可以没有全连接队列&#xff1f; 客户端与服务器间的三次握手 要想成功创建连接需要客…
阅读更多...
ubuntu20.4 安装postgresql并远程连接

ubuntu20.4 安装postgresql并远程连接

1、安装对应的库 如果提示安装失败&#xff0c;E: Unable to locate package postgresql这个错误的话&#xff0c;那就更新下&#xff0c;执行sudo apt update 命令&#xff0c;再次下载&#xff0c;应该就可以成功 sudo apt install postgresql 2、端口设置 检查5432端口是否监…
阅读更多...
最新文章

Copyright @ 2022~2023