SparkSession和sparkSQL

2023-11-16

一、概述

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来读取

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) 

(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]

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。

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

SparkSession和sparkSQL 的相关文章

随机推荐

  • 依据ASCII排序:有关map.entrySet()的一些心得

    List
  • java的main是主线程吗_main函数是主线程吗

    1 线程的概念 线程是程序最基本的运行单位 而进程不能运行 所以能运行的 是进程中的线程 2 线程是如何创建起来的 进程仅仅是一个容器 包含了线程运行中所需要的数据结构等信息 一个进程创建时 操作系统会创建一个线程 这就是主线程 而其他的从
  • AI 与智能化 API 治理的探索实践

    7月 Eolink 受邀参加 QECon 2023 全球软件质量 效能大会 北京站 Eolink CEO 刘昊臻 发表了主题为 AI 与智能化 API 治理的探索实践 的演讲 分享 Eolink 在 API 全生命周期中治理实践与 AI 结
  • Qt工具相关--无法Debug调试的原因

    第一个原因 缺少对应的调试器 第二个原因 调试器的位数没有对应 这些基本原因都可以在项目 构建套件里面看到 逐项检查 耐心分析问题原因是工程师的基本功 第二次遇到类似问题 解决效率务必大幅度提升 这个一道及格线
  • javascript atob()函数和 btoa()函数-Base64的编码与解码

    在 JavaScript 中 有两个函数被分别用来处理解码和编码 base64 字符串 atob ASCII to Base64 btoa Base64 to ASCII atob 函数能够解码通过base 64编码的字符串数据 相反地 b
  • Vue3.0 PC端滑块拼图验证,配合后端验证

    Vue3 0 PC端滑块拼图验证 配合后端验证 简介 最近因为产品需要实现一个滑块拼图验证 而且需要配合后端进行验证 不想接入第三方SDK 所以自己手写了一个 主要是配合element plus 实现UI大致框架 背景图片和拼图都是通过后端
  • matlab求解整数规划、0-1规划

    matlab求解整数规划 0 1规划 R2014以前无法求解整数规划 2014之后用bintprog求解0 1规划 线性规划在2016版本中暂时还可用linprog求解 注 代码中标注的pXXXtaskX指的是西安交大采用的第二版数学实验教
  • Chrome浏览器高级参数

    重新启动一个新的Chrome浏览器 Applications Google Chrome app Contents MacOS Google Chrome disable web security user data dir Users s
  • 服务网格实施周期缩短 50%,丽迅物流基于阿里云 ACK 和 ASM 的云原生应用管理实践

    公司介绍 丽迅物流是百丽旗下专注于时尚产业 为企业提供专业物流及供应链解决方案的服务商 其产品服务主要包括城市落地配 仓配一体 干线运输及定制化解决方案 通过自研智能化物流管理平台 全面助力企业合作集约化发展 目前 丽迅物流已在全国拥有 7
  • Cassandra部分参数调优

    优化一 commitlog文件 由于客户端写数据时需要阻塞写commitlog 默认64GB 所以可以把commitlog放在SSD或者较为空闲的盘中 执行一次刷操作成功后 commitlog文件将被删除 commitlog directo
  • servlet传json数据给前端

    重点 1 json数据的发送 后 2 json数据的提取 前 例如 servlet返回从session提取的当前用户信息给前端使用 1 前端不带数据发送axios给后端请求数据 用get axios method get url http
  • 使用Docker compose部署SpringBoot项目

    我们使用Docker的时候 定义Dockerfile文件 然后使用docker build docker run等命令操作容器 对Docker不熟悉的可以前往查看中文文档 Docker文档 然而微服务架构的应用系统一般包含若干个微服务 每个
  • 6.webpack 代码编译工具 的使用 (实现模块化)

    6 1 关于webpack webpack 是一个用于现代 JavaScript 应用程序的 静态模块打包工具 当 webpack 处理应用程序时 它会在内部从一个或多个入口点构建一个 依赖图 dependency graph 然后将你项目
  • 【剑指offer】面试题8:旋转数组的最小数字

    题目描述 把一个数组最开始的若干个元素搬到数组的末尾 我们称之为数组的旋转 输入一个非递减排序的数组的一个旋转 输出旋转数组的最小元素 例如数组 3 4 5 1 2 为 1 2 3 4 5 的一个旋转 该数组的最小值为1 NOTE 给出的所
  • 海康威视RSTP摄像头视频数据从内网读取在web用HLS播放

    环境 海康威视录像机DS 7804NB K1 4P Ubuntu16 04 录像机内网地址192 168 100 99 一个摄像头接在4号口 32口以下通道号从33开始 摄像头内网地址 192 168 1 64 服务器内网地址192 168
  • SpringBoot中@Transactional注解不起作用的解决办法

    今天写完Service 想测试一下事务性 看是不是设置的自定义异常都能成功完成数据回滚 于是我写了一个方法 Override public boolean addNewUser UserInfo userInfo try int count
  • 云计算的未来:云原生架构和自动化运维的崭露头角

    文章目录 云原生架构 重新定义应用开发和部署 什么是云原生架构 为什么云原生架构重要 1 弹性和伸缩性 2 故障隔离 3 更快的交付 4 资源利用率 5 多云支持 云原生架构的实践 步骤 1 容器化 步骤 2 微服务 步骤 3 自动化运维
  • java设计模式——抽象工厂模式(Abstract Factory Pattern)

    抽象工厂模式产生的动机 为了更清晰地理解工厂方法模式 需要先引入两个概念 产品等级结构 产品等级结构即产品的继承结构 如一个抽象类是电视机 其子类有海尔电视机 海信电视机 TCL电视机 则抽象电视机与具体品牌的电视机之间构成了一个产品等级结
  • 电脑老是弹出vrvedp_m_VrvEdp_m.exe-应用程序错误。。。这弹窗每天弹个三十多次,烦死了,怎么消除?...

    满意答案 qingk0318 2014 05 06 采纳率 49 等级 8 已帮助 1862人 VrvEdp m exe北信源桌面管理软件想再用此软件又无法删除干净建议下载款unlocker软件小安装会鼠标右键菜单里先找北信源桌面管理软件V
  • SparkSession和sparkSQL

    一 概述 spark 有三大引擎 spark core sparkSQL sparkStreaming spark core 的关键抽象是 SparkContext RDD SparkSQL 的关键抽象是 SparkSession Data