理解Spark SQL(二)—— SQLContext和HiveContext

使用Spark SQL,除了使用之前介绍的方法,实际上还可以使用SQLContext或者HiveContext通过编程的方式实现。前者支持SQL语法解析器(SQL-92语法),后者支持SQL语法解析器和HiveSQL语法解析器,默认为HiveSQL语法解析器,用户可以通过配置切换成SQL语法解析器来运行HiveQL不支持的语法,如:select 1。实际上HiveContext是SQLContext的子类,因此在HiveContext运行过程中除了override的函数和变量,可以使用和SQLContext一样的函数和变量。

因为spark-shell工具实际就是运行的scala程序片段,为了方便,下面采用spark-shell进行演示。

首先来看SQLContext,因为是标准SQL,可以不依赖于Hive的metastore,比如下面的例子(没有启动hive metastore):

[[email protected] ~]# $SPARK_HOME/bin/spark-shell --master yarn --conf spark.sql.catalogImplementation=in-memory

scala> case class offices(office:Int,city:String,region:String,mgr:Int,target:Double,sales:Double)
defined class offices

scala> val rddOffices=sc.textFile("/user/hive/warehouse/orderdb.db/offices/offices.txt").map(_.split("\t")).map(p=>offices(p(0).trim.toInt,p(1),p(2),p(3).trim.toInt,p(4).trim.toDouble,p(5).trim.toDouble))
rddOffices: org.apache.spark.rdd.RDD[offices] = MapPartitionsRDD[3] at map at <console>:26

scala> val officesDataFrame = spark.createDataFrame(rddOffices)
officesDataFrame: org.apache.spark.sql.DataFrame = [office: int, city: string ... 4 more fields]

scala> officesDataFrame.createOrReplaceTempView("offices")

scala> spark.sql("select city from offices where region=‘Eastern‘").map(t=>"City: " + t(0)).collect.foreach(println)
City: NewYork                                                                   
City: Chicago
City: Atlanta

scala>

执行上面的命令后,实际上在yarn集群中启动了一个yarn client模式的Spark Application,然后在scala>提示符后输入的语句会生成RDD的transformation,最后一条命令中的collect会生成RDD的action,即会触发Job的提交和程序的执行。

命令行中之所以加上--conf spark.sql.catalogImplementation=in-memory选项,是因为spark-shell中的默认启动的SparkSession对象spark是默认支持Hive的,不带这个选项启动的话,程序就会去连接hive metastore,因为这里并没有启动hive metastore,因此程序在执行createDataFrame函数时会报错。

程序中的第一行是1个case class语句,这里是定义后面的数据文件的模式的(定义模式除了这个方法,其实还有另外一种方法,后面再介绍)。第二行从hdfs中读取一个文本文件,并工通过map映射到了模式上面。第三行基于第二行的RDD生成DataFrame,第四行基于第三行的DataFrame注册了一个逻辑上的临时表,最后一行就可以通过SparkSession的sql函数来执行sql语句了。

实际上,SQLContext是Spark 1.x中的SQL入口,在Spark 2.x中,使用SparkSession作为SQL的入口,但是为了向后兼容,Spark 2.x仍然支持SQLContext来操作SQL,不过会提示deprecated,所以上面的例子是采用Spark 2.x中的写法。

实际上还有另外一种方法来操作SQL,针对同样的数据,例如:

scala> import org.apache.spark.sql._
import org.apache.spark.sql._

scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._

scala> val schema = new StructType(Array(StructField("office", IntegerType, false), StructField("city", StringType, false), StructField("region", StringType, false), StructField("mgr", IntegerType, true), StructField("target", DoubleType, true), StructField("sales", DoubleType, false)))
schema: org.apache.spark.sql.types.StructType = StructType(StructField(office,IntegerType,false), StructField(city,StringType,false), StructField(region,StringType,false), StructField(mgr,IntegerType,true), StructField(target,DoubleType,true), StructField(sales,DoubleType,false))

scala> val rowRDD = sc.textFile("/user/hive/warehouse/orderdb.db/offices/offices.txt").map(_.split("\t")).map(p => Row(p(0).trim.toInt,p(1),p(2),p(3).trim.toInt,p(4).trim.toDouble,p(5).trim.toDouble))
rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[3] at map at <console>:30

scala> val dataFrame = spark.createDataFrame(rowRDD, schema)
dataFrame: org.apache.spark.sql.DataFrame = [office: int, city: string ... 4 more fields]

scala> dataFrame.createOrReplaceTempView("offices")

scala> spark.sql("select city from offices where region=‘Eastern‘").map(t=>"City: " + t(0)).collect.foreach(println)
City: NewYork                                                                   
City: Chicago
City: Atlanta

这个例子与之前的例子有一些不同,主要的地方有3个:

1. 之前的例子是采用case class定义模式,Spark采用反射来推断Schema;而这个例子采用StructType类型的对象来定义模式,它接收一个数组,数组成员是StructField对象,代表一个字段的定义,每个字段的定义由字段名称、字段类型和是否允许为空组成;

2. 对于代表数据的RDD,之前的例子是直接用case class定义的类型来分割字段,而这个例子是用的Row类型;

3. 在使用createDataFrame函数生成DataFrame时,该函数的参数不一样,之前的例子只要传入RDD对象即可(对象中隐含了模式),而这个例子需要同时传入RDD和定义的schema;

实际编程中建议采用第二种方法,因为其更加灵活,schema信息可以不必是写死的,而是可以在程序运行的过程中生成。

下面接着来看HiveContext的用法,使用HiveContext之前需要确保:

  • 使用的Spark是支持Hive的;
  • Hive的配置文件hive-site.xml已经在Spark的conf目录下;
  • hive metastore已经启动;

举例说明:

首先启动hive metastore:

[[email protected] ~]# nohup hive --service metastore &

然后仍然通过spark-shell来举例说明,启动spark-shell,如下所示:

[[email protected] ~]# $SPARK_HOME/bin/spark-shell --master yarn

scala> spark.sql("show databases").collect.foreach(println)
[default]
[orderdb]

scala> spark.sql("use orderdb")
res2: org.apache.spark.sql.DataFrame = []

scala> spark.sql("show tables").collect.foreach(println)
[orderdb,customers,false]
[orderdb,offices,false]
[orderdb,orders,false]
[orderdb,products,false]
[orderdb,salesreps,false]

scala> spark.sql("select city from offices where region=‘Eastern‘").map(t=>"City: " + t(0)).collect.foreach(println)
City: NewYork                                                                   
City: Chicago
City: Atlanta

scala>

可以看到这次启动spark-shell没有带上最后那个选项,这是因为这里我们打算用HiveContext来操作Hive中的数据,需要支持Hive。前面说过spark-shell是默认开启了Hive支持的。同SQLContext类似,Spark 2.x中也不需要再用HiveContext对象来操作SQL了,直接用SparkSession对象来操作就好了。可以看到这里可以直接操作表,不用再定义schema,这是因为schema是由外部的hive metastore定义的,spark通过连接到hive metastore来读取表的schema信息,因此这里能直接操作SQL。

另外,除了上面的使用SQLContext操作普通文件(需要额外定义模式)和使用HiveContext操作Hive表数据(需要开启hive metastore)之外,SQLContext还能操作JSON、PARQUET等文件,由于这两种数据文件自己带了模式信息,因此可以直接基于文件创建DataFrame,例如:

scala> val df = spark.read.json("file:///opt/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> df.createOrReplaceTempView("people")

scala> spark.sql("select name,age from people where age>19").map(t=>"Name :" + t(0) + ", Age: " + t(1)).collect.foreach(println)
Name :Andy, Age: 30

最后来看下DataFrame的另一种叫做DSL(Domain Specific Language)的用法。

scala> val df = spark.read.json("file:///opt/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> df.show()
+----+-------+                                                                  
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

scala> df.select("name").show()
+-------+                                                                       
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+

scala> df.select(df("name"), df("age") + 1).show()
+-------+---------+                                                             
|   name|(age + 1)|
+-------+---------+
|Michael|     null|
|   Andy|       31|
| Justin|       20|
+-------+---------+

scala> df.filter(df("age") > 21).show()
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+

scala> df.groupBy("age").count().show()
+----+-----+                                                                    
| age|count|
+----+-----+
|  19|    1|
|null|    1|
|  30|    1|
+----+-----+

scala>

以上是对Spark SQL的SQLContext和HiveContext基本用法的一些总结,都是采用spark-shell工具举的例子。实际上由于spark-shell是运行scala程序片段的工具,上述例子完全可以改成独立的应用程序。我将在下一篇博文当中尝试使用Scala、Java和Python来编写独立的程序来操作上面的示例hive数据库orderdb,可以适当使用一些较为复杂的SQL来统计分析数据。

原文地址:https://www.cnblogs.com/roushi17/p/sqlcontext_hivecontext.html

时间: 2024-08-29 12:38:08

理解Spark SQL(二)—— SQLContext和HiveContext的相关文章

Spark SQL操作详细讲解

一. Spark SQL和SchemaRDD 关于Spark SQL的前生就不再多说了,我们只关注它的操作.但是,首先要搞明白一个问题,那就是究竟什么是SchemaRDD呢?从Spark的Scala API可以知道org.apache.spark.sql.SchemaRDD和class SchemaRDD extends RDD[Row] with SchemaRDDLike,我们可以看到类SchemaRDD继承自抽象类RDD.官方文档的定义是"An RDD of Row objects tha

Spark SQL 编程

Spark SQL的依赖 Spark SQL的入口:SQLContext 官方网站参考 https://spark.apache.org/docs/1.6.2/sql-programming-guide.html#starting-point-sqlcontext 针对几种不同的语言来写. Spark SQL的入口:HiveContext SQLContext vs HiveContext Spark SQL的作用与使用方式 Spark SQL支持的API 从程序中使用SparkSQL的基本套路

spark sql 创建dataframes

Table of Contents 1. spark sql 2. SQLContext 2.1. sql context是spark sql的所有功能入口点 2.2. 通过spark context创建sql context 2.3. hive context功能上多于sql context,未来sql context也会增加功能 3. DataFrames 3.1. 功能 3.2. 创建DataFrames 3.3. DSL 1 spark sql 是spark的一个模块 处理结构化数据 提

spark sql 小样

package dev.spark.sql import java.util.Properties import org.apache.spark.sql.{Row, SQLContext, SaveMode}import org.apache.spark.sql.hive.HiveContextimport org.apache.spark.sql.types.{IntegerType, StructField, StructType}import org.apache.spark.{Spar

【原创 Hadoop&amp;Spark 动手实践 10】Spark SQL 程序设计基础与动手实践(下)

[原创 Hadoop&Spark 动手实践 10]Spark SQL 程序设计基础与动手实践(下) 目标: 1. 深入理解Spark SQL 程序设计的原理 2. 通过简单的命令来验证Spark SQL的运行原理 3. 通过一个完整的案例来验证Spark SQL的运行原理,自己实际动手来进行掌握 4. 顺利完成“篮球运动员评估系统”

【原创 Hadoop&amp;Spark 动手实践 9】Spark SQL 程序设计基础与动手实践(上)

[原创 Hadoop&Spark 动手实践 9]SparkSQL程序设计基础与动手实践(上) 目标: 1. 理解Spark SQL最基础的原理 2. 可以使用Spark SQL完成一些简单的数据分析任务 3. 可以利用Spark SQL完成一个完整的案例

spark sql 操作

DSL风格语法 1.查看DataFrame中的内容 scala> df1.show +---+--------+---+ | id| name|age| +---+--------+---+ | 1|zhansgan| 16| | 2| lisi| 18| | 3| wangwu| 21| | 4|xiaofang| 22| +---+--------+---+ 2.查看DataFrame部分列的数据 scala> df1.select(df1.col("name")).s

通过Spark SQL关联查询两个HDFS上的文件操作

order_created.txt   订单编号  订单创建时间 10703007267488 2014-05-01 06:01:12.334+01 10101043505096 2014-05-01 07:28:12.342+01 10103043509747 2014-05-01 07:50:12.33+01 10103043501575 2014-05-01 09:27:12.33+01 10104043514061 2014-05-01 09:03:12.324+01 order_pic

Spark SQL自定义外部数据源

1 涉及到的API BaseRelation: In a simple way, we can say it represents the collection of tuples with known schema TableScan: provides a way to scan the data and generates the RDD[Row] from the data RelationProvider: takes a list of parameters and returns