创建SparkSession和sparkSQL的详细过程_Python

一、概述

spark 有三大引擎，spark core、sparksql、sparkstreaming，

spark core 的关键抽象是 sparkcontext、rdd；

sparksql 的关键抽象是 sparksession、dataframe；

sparkstreaming 的关键抽象是 streamingcontext、dstream

sparksession 是 spark2.0 引入的概念，主要用在 sparksql 中，当然也可以用在其他场合，他可以代替 sparkcontext；

sparksession 其实是封装了 sqlcontext 和 hivecontext

(1) 在spark1.6 中我们使用的叫hive on spark，主要是依赖hive生成spark程序，有两个核心组件sqlcontext和hivecontext。

这是spark 1.x 版本的语法

				?

									//set up the spark configuration and create contexts

									 val sparkconf = new sparkconf().setappname("sparksessionzipsexample").setmaster("local")

									// your handle to sparkcontext to access other context like sqlcontext

									 val sc = new sparkcontext(sparkconf).set("spark.some.config.option", "some-value")

									 val sqlcontext = new org.apache.spark.sql.sqlcontext(sc)

（2）spark2.0中我们使用的就是sparksql，是后继的全新产品，解除了对hive的依赖。

从spark2.0以上的版本开始，spark是使用全新的sparksession接口代替spark1.6 中的sqlcontext和hivecontext 来实现对数据的加载、转换、处理等工作，并且实现了sqlcontext和hivecontext的所有功能。

在新版本中并不需要之前那么繁琐的创建很多对象，只需要创建一个sparksession对象即可。sparksession支持从不同的数据源加载数据，并把数据转换成dataframe，并支持把dataframe转换成sqlcontext自身中的表。然后使用sql语句来操作数据，也提供了hiveql以及其他依赖于hive的功能支持。

二、创建sparksession

sparksession 是 spark sql 的入口。使用 dataset 或者 dataframe 编写 spark sql 应用的时候，第一个要创建的对象就是 sparksession。builder 是 sparksession 的构造器。通过 builder, 可以添加各种配置，并通过 stop 函数来停止 sparksession。

builder 的方法如下：

				?

									import org.apache.spark.sql.sparksession

									val spark: sparksession = sparksession.builder

									 .appname("my spark application")  //设置 application 的名字

									 .master("local[*]")             

									 .enablehivesupport() //增加支持 hive support      

									 .config("spark.sql.warehouse.dir", "target/spark-warehouse") //设置各种配置

									 .getorcreate     //获取或者新建一个 sparksession

（1）设置参数

创建sparksession之后可以通过 spark.conf.set 来设置运行参数

				?

									//配置设置

									 spark.conf.set("spark.sql.shuffle.partitions", 6)

									 spark.conf.set("spark.executor.memory", "2g")

									 //获取配置，可以使用scala的迭代器来读取configmap中的数据。 

									 val configmap:map[string, string] = spark.conf.getall()

（2）读取元数据

如果需要读取元数据(catalog)，可以通过sparksession来获取。

				?

									spark.catalog.listtables.show(false)

									 spark.catalog.listdatabases.show(false)

这里返回的都是dataset，所以可以根据需要再使用dataset api来读取

创建SparkSession和sparkSQL的详细过程

catalog 和 schema 是两个不同的概念
catalog是目录的意思，从数据库方向说，相当于就是所有数据库的集合;
schema是模式的意思, 从数据库方向说, 类似catelog下的某一个数据库;

（3）创建dataset和dataframe

通过sparksession来创建dataset和dataframe有多种方法。

通过range()方法来创建dataset
通过createdataframe()来创建dataframe。

				?

									// create a dataset using spark.range starting from 5 to 100,

									// with increments of 5

									val numds = spark.range(5, 100, 5)//创建dataset

									// reverse the order and display first 5 items

									numds.orderby(desc("id")).show(5)

									//compute descriptive stats and display them

									numds.describe().show()

									// create a dataframe using spark.createdataframe from a list or seq

									val langpercentdf = spark.createdataframe( list( ("scala", 35), 

									    ("python", 30), ("r", 15), ("java", 20)) )//创建dataframe

									//rename the columns

									val lpdf = langpercentdf.withcolumnrenamed("_1", "language").

									    withcolumnrenamed("_2", "percent")

									//order the dataframe in descending order of percentage

									lpdf.orderby(desc("percent")).show(false)

创建SparkSession和sparkSQL的详细过程

（4）读取数据

可以用sparksession读取json、csv、txt 和 parquet表。

				?

									import spark.implicits //使rdd转化为dataframe以及后续sql操作

									//读取json文件，生成dataframe

									val jsonfile = args(0)

									val zipsdf = spark.read.json(jsonfile)

（5）使用sparksql

借助sparksession用户可以像sqlcontext一样使用spark sql的全部功能。

				?

									zipsdf.createorreplacetempview("zips_table")//对上面的dataframe创建一个表

									zipsdf.cache()//缓存表

									val resultsdf = spark.sql("select city, pop, state, zip from zips_table")

									//对表调用sql语句

									resultsdf.show(10)//展示结果

（6）存储/读取hive表

下面的代码演示了通过sparksession来创建hive表并进行查询的方法。

				?

									//drop the table if exists to get around existing table error

									 spark.sql("drop table if exists zips_hive_table")

									 //save as a hive table

									 spark.table("zips_table").write.saveastable("zips_hive_table")

									 //make a similar query against the hive table

									 val resultshivedf = spark.sql("select city, pop, state, 

									        zip from zips_hive_table where pop > 40000")

									 resultshivedf.show(10)

三、 sqlcontext

它是 sparksql 的入口点，sparksql 的应用必须创建一个 sqlcontext 或者 hivecontext 的类实例

				?

									from pyspark import sparkcontext, sparkconf

									from pyspark.sql import sparksession, sqlcontext, hivecontext

									conf = sparkconf().setappname('test').setmaster('yarn')

									sc = sparkcontext(conf=conf)

									sqlc = sqlcontext(sc)

									print(dir(sqlc))

									# 'cachetable', 'clearcache', 'createdataframe', 'createexternaltable', 'droptemptable', 'getconf', 'getorcreate', 'newsession', 'range', 'read', 'readstream',

									# 'registerdataframeastable', 'registerfunction', 'registerjavafunction', 'setconf', 'sparksession', 'sql', 'streams', 'table', 'tablenames', 'tables', 'udf', 'uncachetable'

									### sqlcontext 读取数据也自动生成 df

									data = sqlc.read.text('/usr/yanshw/test.txt')

									print(type(data))

四、 hivecontext

它是 sparksql 的另一个入口点，它继承自 sqlcontext，用于处理 hive 中的数据

hivecontext 对 sqlcontext 进行了扩展，功能要强大的多

1. 它可以执行 hivesql 和 sql 查询

2. 它可以操作 hive 数据，并且可以访问 hiveudf

3. 它不一定需要 hive，在没有 hive 环境时也可以使用 hivecontext

注意，如果要处理 hive 数据，需要把 hive 的 hive-site.xml 文件放到 spark/conf 下，hivecontext 将从 hive-site.xml 中获取 hive 配置信息；

如果 hivecontext 没有找到 hive-site.xml，他会在当前目录下创建 spark-warehouse 和 metastore_db 两个文件夹

				?

									from pyspark import sparkcontext, sparkconf

									from pyspark.sql import sparksession, sqlcontext, hivecontext

									conf = sparkconf().setappname('test').setmaster('yarn')

									sc = sparkcontext(conf=conf)

									## 需要把 hive/conf/hive-site.xml 复制到 spark/conf 下

									hivec = hivecontext(sc)

									print(dir(hivec))

									# 'cachetable', 'clearcache', 'createdataframe', 'createexternaltable', 'droptemptable', 'getconf', 'getorcreate', 'newsession', 'range', 'read', 'readstream','refreshtable',

									# 'registerdataframeastable', 'registerfunction', 'registerjavafunction', 'setconf', 'sparksession', 'sql', 'streams', 'table', 'tablenames', 'tables', 'udf', 'uncachetable'

									data = hivec.sql('''select * from hive1101.person limit 2''')

									print(type(data))

sparksession 创建

				?

									from pyspark.sql import sparksession

									### method 1

									sess = sparksession.builder \

									    .appname("aaa") \

									    .config("spark.driver.extraclasspath", sparkclasspath) \

									    .master("local") \

									    .enablehivesupport() \  # sparksql 连接 hive 时需要这句

									    .getorcreate()      # builder 方式必须有这句

									### method 2

									conf = sparkconf().setappname('myapp1').setmaster('local[4]')   # 设定 appname 和 master

									sess = sparksession.builder.config(conf=conf).getorcreate() # builder 方式必须有这句

									### method 3

									from pyspark import sparkcontext, sparkconf

									conf = sparkconf().setappname('myapp1').setmaster('local[4]')   # 设定 appname 和 master

									sc = sparkcontext(conf=conf)

									sess = sparksession(sc)

1）文件数据源

				?

									from pyspark import sparkcontext, sparkconf

									from pyspark.sql import sparksession, sqlcontext, hivecontext

									conf = sparkconf().setappname('test').setmaster('yarn')

									sc = sparkcontext(conf=conf)

									#### 替代了 sqlcontext 和 hivecontext，其实只是简单的封装，提供了统一的接口

									spark = sparksession(sc)

									print(dir(spark))

									# 很多属性，我把私有属性删了

									# 'builder','builder', 'catalog', 'conf', 'createdataframe', 'newsession', 'range', 'read', 'readstream','sparkcontext', 'sql', 'stop', 'streams', 'table', 'udf', 'version'

									### sess 读取数据自动生成 df

									data = spark.read.text('/usr/yanshw/test.txt')      #read 可读类型 [ 'csv', 'format', 'jdbc', 'json', 'load', 'option', 'options', 'orc', 'parquet', 'schema', 'table', 'text']

									print(type(data))       # <class 'pyspark.sql.dataframe.dataframe'>

2） hive 数据源

				?

									## 也需要把 hive/conf/hive-site.xml 复制到 spark/conf 下

									spark = sparksession.builder.appname('test').master('yarn').enablehivesupport().getorcreate()

									hive_data = spark.sql('select * from hive1101.person limit 2')

									print(hive_data)        # dataframe[name: string, idcard: string]