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

CSDN 2024-08-26 10:35:01 阅读 94

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

目前已经更新到了:

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物理执行引擎。这些优化技术能够自动生成高效的执行计划,提高查询的执行速度。

数据样例

<code>// 数据

1 1,2,3

2 2,3

3 1,2

// 需要实现如下的效果

1 1

1 2

1 3

2 2

2 3

3 1

3 2

编写代码

package icu.wzk

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.{ Dataset, SparkSession}

import org.apache.spark.sql.Encoders

case 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.sparkContext

sc.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

).show

sparkSession.sql(

"""

| select id, explode(split(tags, ","))

| from t1

|""".stripMargin

).show

sparkSession.close()

}

}

运行测试

控制台输出结果为:

+---+---+

| 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 (默认数据源)JSONCSVAvroImagesBinaryFiles(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格式。

测试案例

<code>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 的方式:

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

在这里插入图片描述

导入依赖

<code><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>

编写代码

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.sparkContext

sc.setLogLevel("warn")

spark.sql("show databases").show

spark.sql("select * from ods.ods_trade_product_info").show

val 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").show

spark.close()

}

}



声明

本文内容仅代表作者观点,或转载于其他网站,本站不以此文作为商业用途
如有涉及侵权,请联系本站进行删除
转载本站原创文章,请注明来源及作者。