Spark SQL实战(07)-Data Sources

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
RDS MySQL Serverless 高可用系列,价值2615元额度,1个月
简介: Spark SQL通过DataFrame接口支持对多种数据源进行操作。DataFrame可使用关系型变换进行操作,也可用于创建临时视图。将DataFrame注册为临时视图可以让你对其数据运行SQL查询。

1 概述


Spark SQL通过DataFrame接口支持对多种数据源进行操作。


DataFrame可使用关系型变换进行操作,也可用于创建临时视图。将DataFrame注册为临时视图可以让你对其数据运行SQL查询。


本节介绍使用Spark数据源加载和保存数据的一般方法,并进一步介绍可用于内置数据源的特定选项。


数据源关键操作:


load

save


2 大数据作业基本流程


input 业务逻辑 output

不管是使用MR/Hive/Spark/Flink/Storm。


Spark能处理多种数据源的数据,而且这些数据源可以是在不同地方:


file/HDFS/S3/OSS/COS/RDBMS

json/ORC/Parquet/JDBC

object DataSourceApp {

 def main(args: Array[String]): Unit = {

   val spark: SparkSession = SparkSession.builder()

     .master("local").getOrCreate()

 

   text(spark)

   // json(spark)

   // common(spark)

   // parquet(spark)

   // convert(spark)

   // jdbc(spark)

   jdbc2(spark)

   spark.stop()

 }

}




3 text数据源读写


读取文本文件的 API,SparkSession.read.text()


参数:


path:读取文本文件的路径。可以是单个文件、文件夹或者包含通配符的文件路径。

wholetext:如果为 True,则将整个文件读取为一条记录;否则将每行读取为一条记录。

lineSep:如果指定,则使用指定的字符串作为行分隔符。

pathGlobFilter:用于筛选文件的通配符模式。

recursiveFileLookup:是否递归查找子目录中的文件。

allowNonExistingFiles:是否允许读取不存在的文件。

allowEmptyFiles:是否允许读取空文件。

返回一个 DataFrame 对象,其中每行是文本文件中的一条记录。


def text(spark: SparkSession): Unit = {

 import spark.implicits._

 val textDF: DataFrame = spark.read.text(

   "/Users/javaedge/Downloads/sparksql-train/data/people.txt")

 val result: Dataset[(String, String)] = textDF.map(x => {

   val splits: Array[String] = x.getString(0).split(",")

   (splits(0).trim, splits(1).trim)

 })


编译无问题,运行时报错:


Exception in thread "main" org.apache.spark.sql.AnalysisException: Text data source supports only a single column, and you have 2 columns.;


思考下,如何使用text方式,输出多列的值?


修正后

val result: Dataset[String] = textDF.map(x => {

 val splits: Array[String] = x.getString(0).split(",")

 splits(0).trim

})

result.write.text("out")


继续报错:


Exception in thread "main" org.apache.spark.sql.AnalysisException: path file:/Users/javaedge/Downloads/sparksql-train/out already exists.;


回想Hadoop中MapReduce的输出:


第一次0K

第二次也会报错输出目录已存在

这关系到 Spark 中的 mode


SaveMode

Spark SQL中,使用DataFrame或Dataset的write方法将数据写入外部存储系统时,使用“SaveMode”参数指定如何处理已存在的数据。


SaveMode有四种取值:


SaveMode.ErrorIfExists:如果目标路径已经存在,则会引发异常

SaveMode.Append:将数据追加到现有数据

SaveMode.Overwrite:覆盖现有数据

SaveMode.Ignore:若目标路径已经存在,则不执行任何操作

所以,修正如下:


result.write.mode(SaveMode.overwrite).text("out")


4 JSON 数据源


// JSON

def json(spark: SparkSession): Unit = {

 import spark.implicits._

 val jsonDF: DataFrame = spark.read.json(

   "/Users/javaedge/Downloads/sparksql-train/data/people.json")

 jsonDF.show()


 // 只要age>20的数据

 jsonDF.filter("age > 20")

   .select("name")

   .write.mode(SaveMode.Overwrite).json("out")

 

output:

+----+-------+

| age|   name|

+----+-------+

|null|Michael|

|  30|   Andy|

|  19| Justin|

+----+-------+


嵌套 JSON


// 嵌套 JSON

val jsonDF2: DataFrame = spark.read.json(

 "/Users/javaedge/Downloads/sparksql-train/data/people2.json")

jsonDF2.show()

jsonDF2.select($"name",

 $"age",

 $"info.work".as("work"),

 $"info.home".as("home"))

 .write.mode("overwrite")

 .json("out")


output:

+---+-------------------+----+

|age|               info|name|

+---+-------------------+----+

| 30|[shenzhen, beijing]|  PK|

+---+-------------------+----+



5 标准写法


// 标准API写法

private def common(spark: SparkSession): Unit = {

 import spark.implicits._

 val textDF: DataFrame = spark.read.format("text").load(

   "/Users/javaedge/Downloads/sparksql-train/data/people.txt")

 val jsonDF: DataFrame = spark.read.format("json").load(

   "/Users/javaedge/Downloads/sparksql-train/data/people.json")

 textDF.show()

 println("~~~~~~~~")

 jsonDF.show()

 jsonDF.write.format("json").mode("overwrite").save("out")

}

output:

+-----------+

|      value|

+-----------+

|Michael, 29|

|   Andy, 30|

| Justin, 19|

+-----------+

~~~~~~~~

+----+-------+

| age|   name|

+----+-------+

|null|Michael|

|  30|   Andy|

|  19| Justin|

+----+-------+



6 Parquet数据源


6.1 简介

一种列式存储格式,在大数据环境中高效地存储和处理数据。由Hadoop生态系统中的Apache Parquet项目开发的。


6.2 设计目标

支持高效的列式存储和压缩,并提供高性能的读/写能力,以便处理大规模结构化数据。


Parquet可以与许多不同的计算框架一起使用,如Apache Hadoop、Apache Spark、Apache Hive等,因此广泛用于各种大数据应用程序中。


6.3 优点

高性能、节省存储空间、支持多种编程语言和数据类型、易于集成和扩展等。


private def parquet(spark: SparkSession): Unit = {

 import spark.implicits._

 val parquetDF: DataFrame = spark.read.parquet(

   "/Users/javaedge/Downloads/sparksql-train/data/users.parquet")

 parquetDF.printSchema()

 parquetDF.show()

 parquetDF.select("name", "favorite_numbers")

   .write.mode("overwrite")

   .option("compression", "none")

   .parquet("out")

 

output:

root

|-- name: string (nullable = true)

|-- favorite_color: string (nullable = true)

|-- favorite_numbers: array (nullable = true)

|    |-- element: integer (containsNull = true)

+------+--------------+----------------+

|  name|favorite_color|favorite_numbers|

+------+--------------+----------------+

|Alyssa|          null|  [3, 9, 15, 20]|

|   Ben|           red|              []|

+------+--------------+----------------+



7convert

方便从一种数据源写到另一种数据源。


存储类型转换:JSON==>Parquet


def convert(spark: SparkSession): Unit = {

 import spark.implicits._

 val jsonDF: DataFrame = spark.read.format("json")

   .load("/Users/javaedge/Downloads/sparksql-train/data/people.json")

 jsonDF.show()

 jsonDF.filter("age>20")

   .write.format("parquet").mode(SaveMode.Overwrite).save("out")

9.png



8 JDBC


有些数据是在MySQL,使用Spark处理,肯定要通过Spark读出MySQL的数据。

数据源是text/json,通过Spark处理完后,要将统计结果写入MySQL。


查 DB

写法一

def jdbc(spark: SparkSession): Unit = {

 import spark.implicits._

 val jdbcDF = spark.read

   .format("jdbc")

   .option("url", "jdbc:mysql://localhost:3306")

   .option("dbtable", "smartrm_monolith.order")

   .option("user", "root")

   .option("password", "root")

   .load()

 jdbcDF.filter($"order_id" > 150).show(100)

}


8.png


写法二

val connectionProperties = new Properties()

connectionProperties.put("user", "root")

connectionProperties.put("password", "root")

val jdbcDF2: DataFrame = spark.read

 .jdbc(url, srcTable, connectionProperties)

jdbcDF2.filter($"order_id" > 100)


写 DB

val connProps = new Properties()

connProps.put("user", "root")

connProps.put("password", "root")

val jdbcDF: DataFrame = spark.read.jdbc(url, srcTable, connProps)

jdbcDF.filter($"order_id" > 100)

 .write.jdbc(url, "smartrm_monolith.order_bak", connProps)


若 目标表不存在,会自动帮你创建:

7.png



统一配置管理

如何将那么多数据源配置参数统一管理呢?


先引入依赖:


<dependency>

   <groupId>com.typesafe</groupId>

   <artifactId>config</artifactId>

   <version>1.3.3</version>

</dependency>


配置文件:


6.png


读配置的程序:


package com.javaedge.bigdata.chapter05

import com.typesafe.config.{Config, ConfigFactory}

object ParamsApp {

 def main(args: Array[String]): Unit = {

   val config: Config = ConfigFactory.load()

   val url: String = config.getString("db.default.url")

   println(url)

 }

}

private def jdbcConfig(spark: SparkSession): Unit = {

 import spark.implicits._

 val config = ConfigFactory.load()

 val url = config.getString("db.default.url")

 val user = config.getString("db.default.user")

 val password = config.getString("db.default.password")

 val driver = config.getString("db.default.driver")

 val database = config.getString("db.default.database")

 val table = config.getString("db.default.table")

 val sinkTable = config.getString("db.default.sink.table")

 val connectionProperties = new Properties()

 connectionProperties.put("user", user)

 connectionProperties.put("password", password)

 val jdbcDF: DataFrame = spark.read.jdbc(url, s"$database.$table", connectionProperties)

 jdbcDF.filter($"order_id" > 100).show()



写到新表:


jdbcDF.filter($"order_id" > 158)

.write.jdbc(url, s"$database.$sinkTable", connectionProperties)


5.png

5.png

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
18天前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
52 2
ClickHouse与大数据生态集成:Spark & Flink 实战
|
9天前
|
SQL JSON 分布式计算
【赵渝强老师】Spark SQL的数据模型:DataFrame
本文介绍了在Spark SQL中创建DataFrame的三种方法。首先,通过定义case class来创建表结构,然后将CSV文件读入RDD并关联Schema生成DataFrame。其次,使用StructType定义表结构,同样将CSV文件读入RDD并转换为Row对象后创建DataFrame。最后,直接加载带有格式的数据文件(如JSON),通过读取文件内容直接创建DataFrame。每种方法都包含详细的代码示例和解释。
|
1月前
|
SQL 分布式计算 大数据
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(一)
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(一)
42 0
|
1月前
|
SQL 分布式计算 算法
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(二)
大数据-97 Spark 集群 SparkSQL 原理详细解析 Broadcast Shuffle SQL解析过程(二)
78 0
|
1月前
|
SQL 分布式计算 Java
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
35 0
|
1月前
|
SQL 分布式计算 大数据
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
53 0
|
1月前
|
SQL 存储 分布式计算
大数据-93 Spark 集群 Spark SQL 概述 基本概念 SparkSQL对比 架构 抽象
大数据-93 Spark 集群 Spark SQL 概述 基本概念 SparkSQL对比 架构 抽象
40 0
|
1月前
|
SQL 数据处理 数据库
SQL语句优化与查询结果优化:提升数据库性能的实战技巧
在数据库管理和应用中,SQL语句的编写和查询结果的优化是提升数据库性能的关键环节
|
2月前
|
关系型数据库 MySQL 网络安全
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
|
4月前
|
SQL 存储 监控
SQL Server的并行实施如何优化?
【7月更文挑战第23天】SQL Server的并行实施如何优化?
111 13