怎么创立网站wordpress 多字段

每日一句正能量

哪儿有勤奋，哪儿就有成功。

第4章 Spark SQL结构化数据文件处理

章节概要

在很多情况下，开发工程师并不了解Scala语言，也不了解Spark常用API，但又非常想要使用Spark框架提供的强大的数据分析能力。Spark的开发工程师们考虑到了这个问题，利用SQL语言的语法简洁、学习门槛低以及在编程语言普及程度和流行程度高等诸多优势，从而开发了Spark SQL模块，通过Spark SQL，开发人员能够通过使用SQL语句，实现对结构化数据的处理。本章将针对Spark SQL的基本原理、使用方式进行详细讲解。

4.2 DataFrame概述

• Spark SQL使用的数据抽象并非是RDD，而是DataFrame。
• 在Spark 1.3.0版本之前，DataFrame被称为SchemaRDD。
• DataFrame使Spark具备处理大规模结构化数据的能力。
• 在Spark中，DataFrame是一种以RDD为基础的分布式数据集。
• DataFrame的结构类似传统数据库的二维表格，可以从很多数据源中创建，如结构化文件、外部数据库、Hive表等数据源。

4.2.1 DataFrame简介

Spark SQL使用的数据抽象并非是RDD，而是DataFrame。在Spark 1.3.0版本之前，DataFrame被称为SchemaRDD。DataFrame使Spark具备了处理大规模结构化数据的能力。在Spark中，DataFrame是一种以RDD为基础的分布式数据集，因此DataFrame可以完成RDD的绝大多数功能，在开发使用时，也可以调用方法将RDD和DataFrame进行相互转换。DataFrame的结构类似于传统数据库的二维表格，并且可以从很多数据源中创建，例如结构化文件、外部数据库、Hive表等数据源。

DataFrame可以看作是分布式的Row对象的集合，在二维表数据集的每一列都带有名称和类型，这就是Schema元信息，这使得Spark框架可获取更多数据结构信息，从而对在DataFrame背后的数据源以及作用于DataFrame之上数据变换进行针对性的优化，最终达到提升计算效率。

RDD是分布式的Java对象的集合，例如图中的RDD[Person]数据集，虽然它以Person为类型参数，但是对象内部之间的结构相对于Spark框架本身是无法得知的，这样在转换数据形式时效率相对较低。

总的来说，DataFrame除了提供比RDD更丰富的算子以外，更重要的特点是提升Spark框架执行效率、减少数据读取时间以及优化执行计划。通过DataFrame API或SQL处理数据，Spark优化器(Catalyst）会自动优化，即使我们写的程序或SQL不高效，程序也可以高效的执行。

4.2.2 DataFrame的创建

在Spark2.0版本之前;Spark SQL中的SQLContext是创建DataFrame和执行SQL的入口,而在Spark2.0之后，Spark使用全新的SparkSession接口替代SQLContext接口完成数据的加载、转换、处理等功能。
创建SparkSession对象可以通过“SparkSession.builder().getOrCreate()“方法获取，但当我们使用Spark-Shell编写程序时，SparkShell客户端会默认提供了一个名为“sc”“的SparkContext对象和一个名为”“spark"的SparkSession对象，因此我们可以直接使用这两个对象，不需要自行创建。启动Spark-Shell命令如下所示。
spark-shell --master local[2]

在图中可以看出，SparkContext、SparkSession对象已创建完成。

创建DataFrame的两种基本方式：

• 已存在的RDD调用toDF()方法转换得到DataFrame。
• 通过Spark读取数据源直接创建DataFrame。

在创建DataFrame之前，为了支持RDD转换成DataFrame及后续的SQL操作，需要导入spark.implicits._包启用隐式转换。若使用SparkSession方式创建DataFrame，可以使用spark.read操作，从不同类型的文件中加载数据创建DataFrame，具体操作API如表所示。

代码示例	描述
spark.read.text(“people.txt”)	读取txt格式的文本文件，创建DataFrame
spark.read.csv (“people.csv”)	读取csv格式的文本文件，创建DataFrame
spark.read.json(“people.json”)	读取json格式的文本文件，创建DataFrame
spark.read.parquet(“people.parquet”")	读取parquet格式的文本文件，创建DataFrame

• 通过数据源的方式创建DataFrame
  1.在hadoop01上进入/export/data目录，编辑文件，命令如下

cd /export/data/
vi person.txt

2.文件内容如下 ：

1 zhangsan 20
2 lisi 29
3 wangwu 25
4 zhaoliu 30
5 tianqi 35
6 jerry 40

3.查看一下HDFS文件系统中是否存在/spark目录。目录存在，不用创建，如果没有这个目录，就需要使用命令hadoop fs mkdir进行创建。

4.将person.txt文件上传到HDFS文件系统中的/spark目录下，命令如下
hadoop fs -put person.txt /spark

5. 在spark shell中通过文件直接创建DataFrame。在创建DataFrame之前，为了支持RDD转换成DataFrame及后续的SQL操作，需要导入spark.implicits._

scala >import spark.implicits._
import spark.implicits._scala > val personDF = spark.read.text("/spark/person.txt")
personDF: org.apache.spark.sql.DataFrame = [value: String]

6.打印DataFrame的源数据

scala > personDF.printSchema()
root|-- value: String (Nullable = true)

从上述返回结果personDF的属性可以看出，创建DataFrame对象完成，之后调用DataFrame的printSchema()方法可以打印当前对象的Schema元数据信息。从返回结果可以看出，当前value字段是String数据类型，并且还可以为Null。

使用DataFrame的show()方法可以查看当前DataFrame的结果数据，具体代码和返回结果如下所示。

从上述返回结果看出，当前personDF对象中的6条记录就对应了person.txt文本文件中的数据。

• 通过RDD转换DataFrame
  调用RDD的toDF()方法，可以将RDD转换为DataFrame对象，具体如下所示。
  1.通过数据源的方式获得RDD，如下所示
  
  2.定义一个样例类Person。
  
  3.将RDD转换为Person类的RDD。
  
  4.RDD转换为DataFrame，并调用show显示打印。
  
  5.还可以打印DataFrame的元素。
  

完整代码如下

scala > val lineRDD = sc.textFile("/spark/person.txt").map(_.split(" "))
lineRDD: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[6] at map at <console>:24scala > case class Person(id:Int,name:String,age:Int) defined class Personscala > val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))
personRDD: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[7] at map at <console>:27scala > val personDF = personRDD.toDF()
personDF: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]

4.2.3 DataFrame的常用操作

DataFrame提供了两种语法风格，即DSL风格语法和SQL风格语法。二者在功能上并无区别，仅仅是根据用户习惯，自定义选择操作方式。

• DSL风格
  DataFrame提供了一个领域特定语言（DSL）以方便操作结构化数据。
• SQL风格
  在程序中直接使用spark.sql()方式执行SQL查询，结果将作为一个DataFrame返回，使用SQL风格操作的前提是将DataFrame注册成一个临时表。
  1.DSL风格操作DataFrame
  DSL风格操作DataFrame的常用方法，具体如下表如示。

方法名称	相关说明
show()	查看DataFrame中的具体内容信息
printSchema()	查看DataFrame的Schema信息
select()	查看DataFrame中选取部分列的数据及进行重命名
filter()	实现条件查询，过滤出想要的结果
groupBy()	对记录进行分组
sort()	对特定字段进行排序操作

DataFrame提供了一个领域特定语言（DSL）以方便操作结构化数据，下面将针对DSL操作风格，讲解DataFrame常用操作示例

• show()
  查看DataFrame中的具体内容信息
• printSchema()
  查看DataFrame的Schema信息
• select()
  查看DataFrame中选取部分列的数据及进行重命名

下面演示查看personDF对象的name字段数据，具体代码如下所示
personDF.select(personDF.col("name")).show()

结果如下图所示

上述代码中，查询name字段的数据还可以直接使用"personDF.select(“name”).show"代码直接查询

select()操作还可以实现对列名进行重命名，具体代码如下所示
personDF.select(personDF("name").as("username"),personDF("age")).show()

结果如下图所示

从返回结果看出，原name字段重命名为username字段

• filter()
  实现条件查询，过滤出想要的结果

下面演示过滤age大于等于25的数据，具体代码如下所示。
personDF.filter(personDF("age")>=25).show()

结果如下图所示

• groupBy()
  对记录进行分组

下面演示按年龄进行分组并统计相同年龄的人数，具体代码如下所示。
personDF.groupBy("age").count().show()

结果如下图所示

• sort()
  对特定字段进行排序操作

下面演示按年龄降序排列，具体代码如下所示。
personDF.sort(personDF("age").desc).show()

结果如下图所示

2.SQL风格操作DataFrame
DataFrame的强大之处就是我们可以将它看作是一个关系型数据表，然后可以在程序中直接使用spark.sql()的方式执行SQL查询，结果将作为一个DataFrame返回。使用SQL风格操作的前提是需要将DataFrame注册成一个临时表，代码如下所示。

1. 将DataFrame注册成一个临时表
   scala > personDF.registerTempTable("t_person")

结果如下图所示

下面通过多个示例，演示使用SQL风格方式操作DataFrame。

2. 查询年龄最大的前两名人的信息，具体执行代码如下所示。

scala > spark.sql("select * from t_person order by age desc limit 2").show()
+---+------+---+
| id|  name|age|
+---+------+---+
|  6| jerry| 40|
|  5|tianqi| 35|
+---+------+---+

结果如下图所示

3. 查询年龄大于25的人的信息，具体执行代码如下所示。

scala > spark.sql("select * from t_person where age > 25").show()
+---+-------+---+
| id   |  name |age|
+---+-------+---+
|  2   |   lisi     | 29  |
|  4   | zhaoliu| 30 |
|  5   | tianqi  | 35  |
|  6   |  jerry   | 40  |
+---+-------+---+

结果如下图所示

DataFrame操作方式简单，并且功能强大，熟悉SQL语法的开发者都能够快速地掌握DataFrame的操作，本小节只讲解了部分常用的操作方式，读者可通过Spark官方文档http://spark.apache.org/docs/latest/sql-programming-guide.html详细学习DataFrame的操作方式

转载自：https://blog.csdn.net/u014727709/article/details/136033246
欢迎 👍点赞✍评论⭐收藏，欢迎指正