第五章 Spark SQL——结构化数据文件处理思维导图

弓长-张

2024-05-08

浏览量: 5

数据

通过外部数据库创建

查看数据

结构化数据文件处理简述

树图思维导图提供《第五章 Spark SQL——结构化数据文件处理》在线思维导图免费制作，点击“编辑”按钮，可对《第五章 Spark SQL——结构化数据文件处理》进行在线思维导图编辑，本思维导图属于思维导图模板主题，文件编号是：7e8bd38d7e3737cf8e20e549e6b66c5a

思维导图大纲

认识 Spark SQL

Spark SOL 是一个用于处理结构化数据的框架,可被视为一个分布式的 SOL 查询引擎提供了一个抽象的可编程数据模型 DataFrame。Spark SQL 框架的前身是 Shark 框架，由于Shark 需要依赖于 Hive,这制约了 Spark各个组件的相互集成,因此 Spark 团队提出了 SparkSOL项目。Spark SOL 借鉴了 Shark 的优点，同时摆脱了对 Hive 的依赖。相对于 Shark,Spark SQL 在数据兼容、性能优化、组件扩展等方面更有优势。 Spark SQL 在数据兼容方面的发展，使得开发人员不仅可以直接处理 RDD，还可以处理 Parquet 文件或 JSON 文件，甚至可以处理外部数据库中的数据、Hive 中存在的表数据。Spark SOL 的一个重要特点是能够统一处理关系表数据和 RDD 数据，开发人员可以轻松地使用 SOL 或 HiveOL,语句进行外部查询，也可以进行更复杂的数据分析。

Spark SOL 的运行过程。 Spark SOL 提供的核心的编程数据模型是DataFrame。DataFrame是个分布式的Row对象的数据集合，实现了 RDD 的绝大多数功能。Spark SOL 通过 SparkSession 人口对象提供的方法可从外部数据源如 Parquent 文件、JSON 文件、RDDs、Hive 表等加载数据为 DataFrame,再通过 DataFrame 提供的 AP接口、DSL(领域特定语言)、spark-shel1、spark-sgl或Thrift Server等方式对 Data Fram数据进行查询、转换等操作，并将结果展现出来或使用 save0、saveAsTable0方法将结果存储为不同格式的文件。

配置 Spark SQl

Spark SOL 可以兼容 Hive 以便在 Spark SQL 中访问 Hive 表、使用 UDF(用户自定义函数)和使用 Hive 查询语言。从Spark 1.1开始，Spark 增加了 Spark SQL 命令行界面(Command-Line Interface，CLI)和 Thrift Server 功能，使得 Hive 的用户和更熟悉 SOL 语句的数据库管理员更容易上手。即使没有部署好 Hive，SparkSQL也可以运行。若使用SparkSOL的方式访问并操作Hive 表数据,则需要对 Spark SQL进行如下的环境配置( Spark集群已搭建好),将 Spark SOL连接至部署成功的 Hive 上。

子主题 2

Spark SQL与 Shell 交互

Spark SQL框架其实已经集成在 spark-shell 中，因此，启动spark-shell 即可使用 SparkSOL的 Shell 交互接口。从 Spark 2.x版本开始，Spark对 SOLContext和 HiveContext 进行整合，提供一种全新的入口方式:SparkSession。如果在 spark-shell 中执行 SQL语句，那么需要使用 SparkSession 对象调用 sql0方法。spark-shell 在启动的过程中会初始化 SparkSession 对象为 spark，此时初始化的 spark对象既支持 SOL 语法解析器，也支持 HiveQL语法解析器。也就是说，使用 spark 可以执行 SOL 语句和 HiveOL 语句。如果是使用IntelliJ IDEA软件开发SparkSQL程序,则需要在程序开头创建SparkSession对象。如果需要支持 Hive,还需要启用 enableHiveSupport0)方法。

子主题 2

创建 DataFrame 对象

DataFrame 可以通过结构化数据文件、外部数据库、Spark 计算过程中生成的 RDDHive 中的表进行创建。不同数据源的数据转换成 DataFrame 的方式也不同。

通过结构化数据文件创建 DataFrame

一般情况下，结构化数据文件存储在 HDFS 中，较为常见的结构化数据文件是 Parquet文件或 JSON 文件。Spark SQL 可以通过 load()方法将 HDFS 上的结构化文件数据转换为DataFrame，load(方法默认导人的文件格式是 Parquet。

通过外部数据库创建 DataFrame

Spark SOL还可以通过外部数据库(如 MySQL、Oracle 数据库)创建 DataFrame，使用该方式创建 DataFrame 需要通过 Java数据库互连(Java Database Connectivity，JDBC)连接或开放式数据库互连(Open Database Connectivity,ODBC)连接的方式访问数据库。以MySOL数据库的表数据为例，将 MySOL数据库test 中的 people 表的数据转换为 DataFrame,'password”对应的值修改为实际进入 MySOL 数据，需要将“user’库时的账户名称和密码。

通过 RDD 创建 DataFrame

通过 RDD 数据创建 DataFrame 有两种方式。第一种方式是利用反射机制推断 RDD 模首先需要定义一个样例类，因为只有样例类才能被 Spark 隐式地转换为 DataFrame。

第二种方式是采用编程指定 Schema的方式将 RDD 转换成 DataFrame,实现步骤如下 (1)加载数据创建 RDD。(2)使用 StructType 创建一个和步骤(1)的 RDD 中的数据结构相匹配的 Schema。(3)通过createDataFrame()方法将Schema应用到RDD上，将 RDD 数据转换成DataFrame。

通过 Hive 中的表创建 DataFrame，可以使用 SparkSession 对象。

通过 Hive 中的表创建 DataFrame，可以使用 SparkSession 对象。使用 SparkSession对象并调用 sql()方法査询 Hive 中的表数据并将其转换成 DataFrame如査询 test 数据库中的 people 表数据并将其转换成 DataFrame。

查看 DataFrame 数据

Spark DataFrame 派生于 RDD，因此类似于 RDD。DataFrame 只有在提交行动操作时才进行计算。

printSchema:输出数据模式

创建 DataFrame 对象后，一般会査看 DataFrame 的数据模式。使用 printSchema 函数可以查看 DataFrame 数据模式，输出列的名称和类型。

show():查看数据

使用 show()方法可以査看 DataFrame 数据。

使用 show()方法査看 DataFrame 对象 movies 中的数据，show()方法与 show(true)方法查询到的结果一样，只显示前 20条记录，并且最多只显示 20个字符。如果需要显示所有字符，那么需要使用 show(false)方法。

show()方法默认只显示前 20 条记录。若需要査看前 numRows条记录则可以使用 show(numRows:Int)方法，如通过“movies.show(5)”命令査看 movies 前5条记录，

first()/head()/take()/takeAsList():获取若干条记录

获取 DataFrame若干条记录除了使用show()方法之外,还可以使用 first()、head()、take()、takeAsListO方法.

分别使用 first0)、head0)、take0)、takeAsList()方法査看 movies 中前几条记录。first()和 head()方法的功能类似，以 Row 或 Array[Row]的形式返回一条或多条数据。takeO)和 takeAsList0)方法则会将获得的数据返回 Driver 端，为避免 Driver 提示 OutofMemoryError，数据量比较大时不建议使用这两个方法。

collect()/collectAsList():获取所有数据

collect()方法可以查询 DataFrame 中所有的数据，并返回一个数组，collectAsList()方法和 collect()方法类似，可以査询 DataFrame 中所有的数据，但是返回的是列表。分别使用collect()和 collectAsList()方法査看 movies 所有数据。

掌握 DataFrame 查询操作

DataFrame 查询数据有两种方法，第一种是将 DataFrame 注册成临时表，再通过 SOL语句查询数据。创建了DataFrame 对象 peopleDataFrame，先将peopleDataFrame 注册成临时表，使用 spark.sql()方法查询。

基于 DataFrame 对象 rating 和 use 进行的查询操作

where()/filter()方法使用 where()或 flter()方法可以查询数据中符合条件的所有字段的信息。 (1)where()方法 DataFrame 可以使用 where(conditionExpr:String)方法査询符合指定条件的数据,参数中可以使用 and 或 or。where()方法的返回结果仍然为 DataFrame。

(2)flter()方法 DataFrame 还可以使用 flter()方法筛选出符合条件的数据，

select()/selectExpr()/col()/apply()方法

where()和 filter()方法查询的数据包含的是所有字段的信息，但是有时用户只需要查询部分字段的值即可，DataFrame 提供了查询指定字段的值的方法，如select()、selectExpr()、col()和 apply()方法等。

select()方法:获取指定字段值 select()方法根据传入的 String类型字段名获取对应的值，并返回一个 DataFrame 对象。

selectExpr()方法:对指定字段进行特殊处理在实际业务中，可能需要对某些字段进行特殊处理，如为某个字段取别名、对某个字段的数据进行四舍五入等。DataFrame 提供了 selectExpr()方法，可以对指定字段取别名或调用UDF函数对其进行其他处理。selectExpr()方法传人 String 类型的参数，返回一个 DataFrame 对象。

col()/apply()方法 col()和 apply()方法也可以获取 DataFrame 指定字段，但只能获取一个字段，并且返回的是一个 Column 对象。

limit()方法 limit()方法可以获取指定 DataFrame 数据的前 n条记录。不同于 take()与 head()方法limit()方法不是行动操作,因此并不会直接返回查询结果,需要结合 show()方法或其他行动操作才可以显示结果。

orderBy()/sort()方法 orderBy()方法用于根据指定字段对数据进行排序，默认为升序排序。若要求降序排序orderBy()方法的参数可以使用“desc("字段名称")”或“$"字段名称".desc”，也可以在指定字段前面加“.”。

sort()方法也可以根据指定字段对数据进行排序,用法与 orderBy()方法一样。

groupBy()方法可以根据指定字段对数据进行分组操作。groupBy()方法的输入参数既可以是 String 类型的字段名，也可以是 Column 对象。根据 gender 字段对 user 对象进行分组。

join()方法数据并不一定都存放在同一个表中，也有可能存放在两个或两个以上的表中。根据业务需求，有时候需要连接两个表才可以查询出业务所需的数据。DataFrame 提供了join()方法用于连接两个表。

掌握 DataFrame 输出操作

DataFrame 提供了很多输出操作的方法，其中 save()方法可以将 DataFrame 数据保存成文件;saveAsTable0方法可以将 DataFrame 数据保存成持久化的表，并在 Hive的元数据库中创建一个指针指向该表的位置，持久化的表会一直保留，即使Spark 程序重启也没有影响，只要连接至同一个元数据服务即可读取表数据。读取持久化表时，只需要用表名作为参数，调用 spark.table0方法即可加载表数据并创建 DataFrame。默认情况下，saveAsTable0方法会创建一个内部表，表数据的位置是由元数据服务控制的。如果删除表，那么表数据也会同步删除。

小结

介绍了 Spark SQL 框架，首先简述了 Spark SQL的基本概念，并且讲解了如何配置Spark SQL 以及 Spark SQL与 Shell 交互，接着详细介绍了 Spark SQL的核心抽象编程模型DataFrame 及其基础操作,包括创建 DataFrame 对象、DataFrame 的查询操作和输出操作,最后使用 Spark SQL, 对房屋售价数据进行探索分析。

第五章思维导图