一、概述
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 版本的语法
|
1
2
3
4
5
6
7
|
//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 的方法如下：
|
1
2
3
4
5
6
7
|
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 来设置运行参数
|
1
2
3
4
5
6
|
//配置设置
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来获取。
|
1
2
|
spark.catalog.listtables.show(false)
spark.catalog.listdatabases.show(false)
|
这里返回的都是dataset，所以可以根据需要再使用dataset api来读取
catalog 和 schema 是两个不同的概念
catalog是目录的意思，从数据库方向说，相当于就是所有数据库的集合;
schema是模式的意思, 从数据库方向说, 类似catelog下的某一个数据库;
（3）创建dataset和dataframe
通过sparksession来创建dataset和dataframe有多种方法。
通过range()方法来创建dataset
通过createdataframe()来创建dataframe。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
// 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)
|
（4）读取数据
可以用sparksession读取json、csv、txt 和 parquet表。
|
1
2
3
4
|
import spark.implicits //使rdd转化为dataframe以及后续sql操作
//读取json文件，生成dataframe
val jsonfile = args(0)
val zipsdf = spark.read.json(jsonfile)
|
（5）使用sparksql
借助sparksession用户可以像sqlcontext一样使用spark sql的全部功能。
|
1
2
3
4
5
6
7
|
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表并进行查询的方法。
|
1
2
3
4
5
6
7
8
9
10
11
|
//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 的类实例
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
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 两个文件夹
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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 创建
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
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）文件数据源
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
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 数据源
|
1
2
3
4
5
|
## 也需要把 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]
|
sparksession vs sparkcontext
sparksession 是 spark2.x 引入的新概念，sparksession 为用户提供统一的切入点，字面理解是创建会话，或者连接 spark
在 spark1.x 中，sparkcontext 是 spark 的主要切入点，由于 rdd 作为主要的 api，我们通过 sparkcontext 来创建和操作 rdd,
sparkcontext 的问题在于：
1. 不同的应用中，需要使用不同的 context，在 streaming 中需要使用 streamingcontext，在 sql 中需要使用 sqlcontext，在 hive 中需要使用 hivecontext，比较麻烦
2. 随着 dataset 和 dataframe api 逐渐成为标准 api，需要为他们创建接入点，即 sparksession
sparksession 实际上封装了 sparkcontext，另外也封装了 sparkconf、sqlcontext，随着版本增加，可能更多，
所以我们尽量使用 sparksession ，如果发现有些 api 不在 sparksession 中，也可以通过 sparksession 拿到 sparkcontext 和其他 context 等
在 shell 操作中，原生创建了 sparksession，故无需再创建，创建了也不会起作用
在 shell 中，sparkcontext 叫 sc，sparksession 叫 spark。
到此这篇关于sparksession和sparksql的文章就介绍到这了,更多相关sparksession和sparksql内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！
原文链接：https://blog.csdn.net/ytp552200ytp/article/details/114068166