Spark SQL简介_深夜的猫213_spark sql

irpas 02-07 8309

Spark SQL简介一、从Shark说起

1、在这之前我们要先理解Hive的工作原理：

Hive是一个基于Hadoop的数据仓库工具，提供了类似于关系数据库SQL的查询语言——HiveSQL，用户可以通过HiveSQL语句快速实现简单的MapReduce统计，Hive自身可以自动将HiveSQL语句快速转换成MapReduce任务进行运行。

2、Shark提供了类似于Hive的功能，与Hive不同的是，Shark把SQL语句转换成Spark作业，而不是MapReduce作业。

可以近似地认为：Shark仅将物理执行计划从MapReduce作业替换成了Spark作业，也就是通过Hive的HiveSQL解析功能，把HiveSQL翻译成Spark上的RDD操作。

Shark的设计导致了两个问题：一、是执行计划优化完全依赖于Hive，不方便添加新的优化策略。

二、是因为Spark是线程级并行，而MapReduce是进程级并行，因此，Spark在兼容Hive的实现上存在线程安全问题，导致Shark不得不使用另外一套独立维护的打了补丁的Hive源码分支。

3、Spark SQL架构如下：

Spark SQL在Hive兼容层面仅依赖HiveQL解析、Hive元数据，也就是说，从HQL被解析成抽象语法树（AST）起，就全部由Spark SQL接管了。Spark SQL执行计划生成和优化都由Catalyst（函数式关系查询优化框架）负责。

Spark SQL增加了DataFrame（即带有Schema信息的RDD），使用户可以在Spark SQL中执行SQL语句，数据既可以来自RDD，也可以是Hive、HDFS、Cassandra等外部数据源，还可以是JSON格式的数据。 Spark SQL目前支持Scala、Java、Python三种语言，支持SQL-92规范。

二、DataFrame概述

1、DataFrame的推出，让Spark具备了处理大规模结构化数据的能力，不仅比原有的RDD转化方式更加简单易用，而且获得了更高的计算性能。 Spark能够轻松实现从MySQL到DataFrame的转化，并且支持SQL查询。

RDD是分布式的 Java对象的集合，但是，对象内部结构对于RDD而言却是不可知的。 DataFrame是一种以RDD为基础的分布式数据集，提供了详细的结构信息。从Spark2.0以上版本开始，Spark使用全新的SparkSession接口替代Spark1.6中的SQLContext及HiveContext接口来实现其对数据加载、转换、处理等功能。SparkSession实现了SQLContext及HiveContext所有功能。

SparkSession支持从不同的数据源加载数据，并把数据转换DataFrame，并且支持把DataFrame转换成SQLContext自身中的表，然后使用SQL语句来操作数据。SparkSession亦提供了HiveQL以及其他依赖于Hive的功能的支持。

在编写独立应用程序时，可以通过如下语句创建一个SparkSession对象

from pyspark import SparkContext,SparkConf from pyspark.sql import SparkSession spark = SparkSession.builder.config(conf = SparkConf()).getOrCreate()

实际上，在启动进入pyspark以后，pyspark就默认提供了一个SparkContext对象（名称为sc）和一个SparkSession对象（名称为spark）

2、从不同类型的文件中加载数据创建DataFrame

#从不同类型的文件中加载数据创建DataFrame df1 = spark.read.text("file:///home/hadoop/program1/people.txt") df1.show() df2 = spark.read.json("file:///home/hadoop/program1/people.json") df2.show() df1_1 = spark.read.format("text").load("file:///home/hadoop/program1/people.txt") df1_1.show() df2_1 = spark.read.format("json").load("file:///home/hadoop/program1/people.json") df2_1.show()

结果：

3、DataFrame的保存

#DataFrame的保存 #例：把上面名称为df1的文件保存到不同格式文件中 df1.write.text("df1.txt") df1.write.json("df1.json") df1.write.format("text").save("df1.txt") df1.write.format("json").save("df1.json") df2.select("name","age").write.format("json").save("file:///home/hadoop/program1/df2.json") #选取指定的列保存

另一种存储方式Parquet。详细见下面链接。

很详细的Parquet存储讲解

当把该数据保存到一个文本文件中会新生成一个名称为df1.json的目录（不是文件）和一个名称df1.txt的目录（不是文件）

如果再次读取json或text文件生成DataFrame，可以直接用这个目录名称，不需要使用part-00000-093d3250-a36a-4ca4-affc-5144b2a2759a-c000.txt文件（当然，使用这个文件也可以）。

三、DataFrame的常用操作 printSchema()

打印出DataFrame的模式（Schema）信息。

select()

从DataFrame中选取部分列的数据。

filter()

实现条件查询，找到满足条件要求的记录。

groupBy()

用于对记录进行分组。

sort()

用于对记录进行排序。

四、从RDD转换得到DataFrame

Spark提供了如下两种方法实现从RDD转换得到DataFrame

1.利用反射机制推断RDD模式

利用反射机制来推断包含特定类型对象的RDD的模式（Schema），适用于数据结构已知时的RDD转换。

例：现在要把people.txt加载到内存中生成一个DataFrame，并查询其中的数据：

from pyspark.sql import Row people = sc.textFile("file:///home/hadoop/program1/people.txt") #生成RDD文件 people1 = people.map(lambda x:x.split(" ")).map(lambda x:Row(name=x[0],age=x[1])) #得到新的RDD，每个元素都是Row对象 schemaPeople = spark.createDataFrame(people1) #转换成DataFrame schemaPeople.createOrReplaceTempView("people") #注册为临时表,临时表名字为people personsDF = spark.sql("select name,age from people where age>20") #SQL语句查询 personsRDD = personsDF.rdd.map(lambda x:"Name: "+x.name+","+"Age "+x.age) #格式化输出 personsRDD.collect()

结果：

['Name: Michael,Age 40', 'Name: Andy,Age 30']

2.使用编程方式定义RDD模式

使用编程接口构造一个模式（Schema），并将其应用在已知的RDD上，适用于数据结构未知时的RDD转换。

from pyspark.sql.types import * from pyspark.sql import Row #下面生成“表头” schemaString = "name age" fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split(" ")] schema = StructType(fields) #下面生成“表中的记录” lines = sc.textFile("file:///home/hadoop/program1/people.txt") parts = lines.map(lambda x: x.split(" ")) people = parts.map(lambda x: Row(x[0], x[1].strip())) #下面把“表头”和“表中的记录”拼装在一起 schemaPeople = spark.createDataFrame(people, schema) schemaPeople.createOrReplaceTempView("people") results = spark.sql("select name,age from people") results.show()

结果

+-------+---+ | name|age| +-------+---+ |Michael| 40| | Andy| 30| | Justin| 19| +-------+---+