Spark session 深入理解
Spark session
- Spark session 深入理解
- 创建SparkSession
- Builder 的方法如下:
- 设置参数
- 读取元数据
- 读取数据
- 使用SparkSQL
- 存储/读取Hive表
- 下图是 SparkSession 的类和方法
参考自:https://www.cnblogs.com/zzhangyuhang/p/9039695.html
Spark session 深入理解
在Spark1.6中我们使用的叫Hive on spark,主要是依赖hive生成spark程序,有两个核心组件SQLcontext和HiveContext。
这是Spark 1.x 版本的语法
//set up the spark configuration and create contextsval 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)
而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 或者 Datafram 编写 Spark SQL 应用的时候,第一个要创建的对象就是 SparkSession。
Builder 是 SparkSession 的构造器。 通过 Builder, 可以添加各种配置。
Builder 的方法如下:
MethodDescriptiongetOrCreate获取或者新建一个 sparkSessionenableHiveSupport增加支持 hive SupportappName设置 application 的名字config设置各种配置
你可以通过 SparkSession.builder 来创建一个 SparkSession 的实例,并通过 stop 函数来停止 SparkSession。
import
org.apache.spark.sql.SparkSession
val spark: SparkSession=SparkSession.builder.appName("My Spark Application").master("local[*]") // 如果打包到集群中 master("local[2]") 注释掉.enableHiveSupport() // 使sparkSesssion支持hive.getOrCreate()
这样我就就可以使用我们创建的SparkSession类型的spark对象了。
设置参数
创建SparkSession之后可以通过 spark.conf.set 来设置运行参数
//set new runtime optionsspark.conf.set("spark.sql.shuffle.partitions", 6)spark.conf.set("spark.executor.memory", "2g")//get all settingsval configMap:Map[String, String] = spark.conf.getAll()
//可以使用Scala的迭代器来读取configMap中的数据。
读取元数据
如果需要读取元数据(catalog),可以通过SparkSession来获取。
//fetch metadata data from the catalogspark.catalog.listDatabases.show(false)spark.catalog.listTables.show(false)
这里返回的都是Dataset,所以可以根据需要再使用Dataset API来读取。
注意:catalog 和 schema 是两个不同的概念
Catalog是目录的意思,从数据库方向说,相当于就是所有数据库的集合;
Schema是模式的意思, 从数据库方向说, 类似Catelog下的某一个数据库;
创建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读取JSON、CSV、TXT和parquet表。
import spark.implicits
//使RDD转化为DataFrame以及后续SQL操作
//读取JSON文件,生成DataFrame
val jsonFile = args(0)
val zipsDF = spark.read.json(jsonFile)
使用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)
//展示结果
存储/读取Hive表
下面的代码演示了通过SparkSession来创建Hive表并进行查询的方法。
//drop the table if exists to get around existing table errorspark.sql("DROP TABLE IF EXISTS zips_hive_table")
下图是 SparkSession 的类和方法
这些方法包含了创建 DataSet
方法 | 说明 |
---|---|
builder | 创建一个sparkSession实例 |
version | 返回当前spark的版本 |
implicits | 引入隐式转化 |
emptyDataset[T] | 创建一个空DataSet |
range | 创建一个DataSet[Long] |
sql | 执行sql查询(返回一个dataFrame) |
udf | 自定义udf(自定义函数) |
table | 从表中创建DataFrame |
catalog | 访问结构化查询实体的目录 |
read | 外部文件和存储系统读取DataFrame。 |
conf | 当前运行的configuration |
readStream | 访问DataStreamReader以读取流数据集。 |
streams | 访问StreamingQueryManager以管理结构化流式传输查询。 |
newSession | 创建新的SparkSession |
stop | 停止SparkSession |
write | 访问DataStreamReader以写入流数据集。 |
当我们使用Spark-Shell的时候,Spark会自动帮助我们建立好了一个名字为spark的SparkSesson和一个名字为sc的SparkContext。
Spark session 深入理解相关推荐
- 通过案例对 spark streaming 透彻理解三板斧之一: spark streaming 另类实验
本期内容 : spark streaming另类在线实验 瞬间理解spark streaming本质 一. 我们最开始将从Spark Streaming入手 为何从Spark Streaming切入 ...
- 谈谈spark.sql.shuffle.partitions和 spark.default.parallelism 的区别及spark并行度的理解
谈谈spark.sql.shuffle.partitions和 spark.default.parallelism 的区别及spark并行度的理解 spark.sql.shuffle.partitio ...
- [html] 说说你对cookie和session的理解
[html] 说说你对cookie和session的理解 cookie: 可以通过客户端, 服务端设置, 容量小, 可以通过设置domain来实现同步登录, 除了name, value, 它还有多个选 ...
- spark RDD的理解
2019独角兽企业重金招聘Python工程师标准>>> spark RDD的理解 博客分类: 分布式计算 RDD是什么东西?在Spark中有什么作用?如何使用? 1.RDD是什么 ( ...
- Spark Session 与 Spark Context的区别
Spark Session是Spark 2.0中Spark应用程序的统一入口点. 它提供了一种以较少数量的构造与各种spark功能交互的方法. 此时就不需要spark context, hive co ...
- 【Azkaban报错解决】FAILED SemanticException Failed to get a spark session
问题描述: ods_to_dwd_log报以下错误 ods_to_dim_db报一下错误 可以看出都是相同报错,无法创建spark事务 Logging initialized using config ...
- Failed to create Spark client for Spark session/30041Code
记录排错历程 问题简介:根据尚硅谷数仓4.0学习集群运行了一段时间,可以正常使用spark运行,出现阶段运行情况的红色框,但是不知道为什么突然有一次,返回30041code,无法运行创建spark s ...
- 对cookie与session的理解
cookie: 在网站中,http请求是无状态的.也就是说即使第一次和服务器连接后并且登录成功后,第二次请求服务器依然不能知道当前请求是哪个用户. cookie的出现就是为了解决这个问题, 第一次登录 ...
- A-Deeper-Understanding-of-Spark-Internals(Spark内核深入理解)
这篇文章是对Spark Submit 2014会议上Aaron Davidson做的报告的PPT内容的整理,报告主要讲了Spark中对一个统计各个字母开头的名字的个数的代码做的优化. 对此PPT做了下 ...
最新文章
- . NET5一出,. NET岗面试普遍喊难,真相是…
- jboss eap 7_使用JBoss EAP 7的HTTP / 2
- spring----06 更多DI知识
- 计算机网络之网络层:2、IP数据报、IP数据报分片
- C#3.0 为我们带来什么(3) —— 初始化器
- python watchdog_Python watchdog
- android 月牙动画,一加6月牙白版:可能是目前最漂亮的白色安卓旗舰
- python电脑下载教程-电脑64位怎么下载python
- 【已解决】python远程连接数据库问题
- 转iOS 多线程 RunLoop 机制 (三)
- nano命令,vi ed pico sed joe emacs jed ex
- undefined reference to错误的解决方法
- Oracle--Oracle 11.2.0.1客户端安装
- linux下用c语言写吃金豆,吃金豆pacmanTC版
- 微信小程序地图实现展示路线路
- 视频转码(Java)
- 神经派考古学 - 推荐一个blog
- access行列转换。
- TestCenter测试管理工具功能详解六(K)
- 卡尔曼滤波的细致讲解从一维到多维
热门文章
- Opencv之生成Halcon标定板
- python怎么把图片变成字符_用Python把图片变成字符画
- Python基础知识回顾及scrapy框架爬虫基础
- skywalking调研相关资料整理
- 【Computer Organization笔记01】计算机组成原理课程要求、计算机的层次结构、计算机的发展历史
- 在Ubuntu和CentOS上搭建NodeJs的执行环境步骤
- Low Power概念介绍<Level Shifter>
- (二)PowerLink理论知识
- java线程异常终止_java线程莫名异常退出时,如何捕获异常信息
- ff7重制版青魔法_狂父重制版发布+妖精的尾巴首次打折¥244+最终幻想4解锁国区新增中文...