spark-初阶①(介绍+RDD)
Spark是什么?
Apache Spark 是一个快速的, 多用途的集群计算系统, 相对于 Hadoop MapReduce 将中间结果保存在磁盘中, Spark 使用了内存保存中间结果, 能在数据尚未写入硬盘时在内存中进行运算.
Spark的特点(优点)
速度快
- Spark 的在内存时的运行速度是 Hadoop MapReduce 的100倍
- 基于硬盘的运算速度大概是 Hadoop MapReduce 的10倍
- Spark 实现了一种叫做 RDDs 的 DAG 执行引擎, 其数据缓存在内存中可以进行迭代处理
易用
df = spark.read.json("logs.json")df.where("age > 21").select("name.first").show()
- Spark 支持 Java, Scala, Python, R, SQL 等多种语言的API.
- Spark 支持超过80个高级运算符使得用户非常轻易的构建并行计算程序
- Spark 可以使用基于 Scala, Python, R, SQL的 Shell 交互式查询.
通用
- Spark 提供一个完整的技术栈, 包括 SQL执行, Dataset命令式API, 机器学习库MLlib, 图计算框架GraphX, 流计算SparkStreaming
- 用户可以在同一个应用中同时使用这些工具, 这一点是划时代的
兼容
- Spark 可以运行在 Hadoop Yarn, Apache Mesos, Kubernets, Spark Standalone等集群中
- Spark 可以访问 HBase, HDFS, Hive, Cassandra 在内的多种数据库
Spark组件
Spark 最核心的功能是 RDDs, RDDs 存在于 spark-core 这个包内, 这个包也是 Spark 最核心的包.
同时 Spark 在 spark-core 的上层提供了很多工具, 以便于适应不用类型的计算.
Spark-Core 和 弹性分布式数据集(RDDs)
Spark-Core 是整个 Spark 的基础, 提供了分布式任务调度和基本的 I/O 功能Spark 的基础的程序抽象是弹性分布式数据集(RDDs), 是一个可以并行操作, 有容错的数据集合RDDs 可以通过引用外部存储系统的数据集创建(如HDFS, HBase), 或者通过现有的 RDDs 转换得到RDDs 抽象提供了 Java, Scala, Python 等语言的APIRDDs 简化了编程复杂性, 操作 RDDs 类似通过 Scala 或者 Java8 的 Streaming 操作本地数据集合
Spark SQL
Spark SQL 在
spark-core基础之上带出了一个名为 DataSet 和 DataFrame 的数据抽象化的概念Spark SQL 提供了在 Dataset 和 DataFrame 之上执行 SQL 的能力Spark SQL 提供了 DSL, 可以通过 Scala, Java, Python 等语言操作 DataSet 和 DataFrame它还支持使用 JDBC/ODBC 服务器操作 SQL 语言Spark Streaming
Spark Streaming 充分利用
spark-core的快速调度能力来运行流分析它截取小批量的数据并可以对之运行 RDD Transformation它提供了在同一个程序中同时使用流分析和批量分析的能力MLlib
MLlib 是 Spark 上分布式机器学习的框架. Spark分布式内存的架构 比 Hadoop磁盘式 的 Apache Mahout 快上 10 倍, 扩展性也非常优良MLlib 可以使用许多常见的机器学习和统计算法, 简化大规模机器学习汇总统计, 相关性, 分层抽样, 假设检定, 随即数据生成支持向量机, 回归, 线性回归, 逻辑回归, 决策树, 朴素贝叶斯协同过滤, ALSK-meansSVD奇异值分解, PCA主成分分析TF-IDF, Word2Vec, StandardScalerSGD随机梯度下降, L-BFGS
GraphX
GraphX 是分布式图计算框架, 提供了一组可以表达图计算的 API, GraphX 还对这种抽象化提供了优化运行
Spark 集群结构
Spark 自身是没有集群管理工具的, 但是如果想要管理数以千计台机器的集群, 没有一个集群管理工具还不太现实, 所以 Spark 可以借助外部的集群工具来进行管理
整个流程就是使用 Spark 的 Client 提交任务, 找到集群管理工具申请资源, 后将计算任务分发到集群中运行
Driver该进程调用 Spark 程序的 main 方法, 并且启动 SparkContext
Cluster Manager该进程负责和外部集群工具打交道, 申请或释放集群资源
Worker该进程是一个守护进程, 负责启动和管理 Executor
Executor该进程是一个JVM虚拟机, 负责运行 Spark Task
运行一个 Spark 程序大致经历如下几个步骤
- 启动 Drive, 创建 SparkContext
- Client 提交程序给 Drive, Drive 向 Cluster Manager 申请集群资源
- 资源申请完毕, 在 Worker 中启动 Executor
- Driver 将程序转化为 Tasks, 分发给 Executor 执行
Driver 和 Worker 什么时候被启动?
- Standalone 集群中, 分为两个角色: Master 和 Slave, 而 Slave 就是 Worker, 所以在 Standalone 集群中, 启动之初就会创建固定数量的 Worker
- Driver 的启动分为两种模式: Client 和 Cluster. 在 Client 模式下, Driver 运行在 Client 端, 在 Client 启动的时候被启动. 在 Cluster 模式下, Driver 运行在某个 Worker 中, 随着应用的提交而启动
- 在 Yarn 集群模式下, 也依然分为 Client 模式和 Cluster 模式, 较新的版本中已经逐渐在废弃 Client 模式了, 所以上图所示为 Cluster 模式
- 如果要在 Yarn 中运行 Spark 程序, 首先会和 RM 交互, 开启 ApplicationMaster, 其中运行了 Driver, Driver创建基础环境后, 会由 RM 提供对应的容器, 运行 Executor, Executor会反向向 Driver 反向注册自己, 并申请 Tasks 执行
- 在后续的 Spark 任务调度部分, 会更详细介绍
总结
Master负责总控, 调度, 管理和协调 Worker, 保留资源状况等Slave对应 Worker 节点, 用于启动 Executor 执行 Tasks, 定期向 Master汇报Driver运行在 Client 或者 Slave(Worker) 中, 默认运行在 Slave(Worker) 中
| 地址 | 解释 |
|---|---|
local[N] | 使用 N 条 Worker 线程在本地运行 |
spark://host:port | 在 Spark standalone 中运行, 指定 Spark 集群的 Master 地址, 端口默认为 7077 |
mesos://host:port | 在 Apache Mesos 中运行, 指定 Mesos 的地址 |
yarn | 在 Yarn 中运行, Yarn 的地址由环境变量 HADOOP_CONF_DIR 来指定 |
spark-submit 命令
spark-submit [options] <app jar> <app options>
app jar程序 Jar 包app options程序 Main 方法传入的参数options提交应用的参数, 可以有如下选项
同 Spark shell 的 Master, 可以是spark, yarn, mesos, kubernetes等 URL --deploy-mode Driver 运行位置, 可选 Client 和 Cluster, 分别对应运行在本地和集群(Worker)中 --class Jar 中的 Class, 程序入口 --jars 依赖 Jar 包的位置 --driver-memory Driver 程序运行所需要的内存, 默认 512M --executor-memory Executor 的内存大小, 默认 1G " style="margin: 20px 0px 10px; padding: 0px; cursor: text; position: relative; font-family: Helvetica, "Hiragino Sans GB", 微软雅黑, "Microsoft YaHei UI", SimSun, SimHei, arial, sans-serif;">参数 解释 --master <url>同 Spark shell 的 Master, 可以是spark, yarn, mesos, kubernetes等 URL --deploy-mode <client or cluster>Driver 运行位置, 可选 Client 和 Cluster, 分别对应运行在本地和集群(Worker)中 --class <class full name>Jar 中的 Class, 程序入口 --jars <dependencies path>依赖 Jar 包的位置 --driver-memory <memory size>Driver 程序运行所需要的内存, 默认 512M --executor-memory <memory size>Executor 的内存大小, 默认 1G
| 参数 | 解释 |
|---|---|
--master <url> | 同 Spark shell 的 Master, 可以是spark, yarn, mesos, kubernetes等 URL |
--deploy-mode <client or cluster> | Driver 运行位置, 可选 Client 和 Cluster, 分别对应运行在本地和集群(Worker)中 |
--class <class full name> | Jar 中的 Class, 程序入口 |
--jars <dependencies path> | 依赖 Jar 包的位置 |
--driver-memory <memory size> | Driver 程序运行所需要的内存, 默认 512M |
--executor-memory <memory size> | Executor 的内存大小, 默认 1G |
弹性分布式数据集(RDDs)
分布式
RDD 支持分区, 可以运行在集群中
弹性
- RDD 支持高效的容错
- RDD 中的数据即可以缓存在内存中, 也可以缓存在磁盘中, 也可以缓存在外部存储中
- Task如果失败会自动进行特定次数的重试
RDD的计算任务如果运行失败，会自动进行任务的重新计算，默认次数是4次 - Stage如果失败会自动进行特定次数的重试
如果Job的某个Stage阶段计算失败，框架也会自动进行任务的重新计算，默认次数也是4次。 - Checkpoint(斩断依赖链)和Persist(缓存)可主动或被动触发
RDD可以通过Persist持久化将RDD缓存到内存或者磁盘，当再次用到该RDD时直接读取就行。也可以将RDD进行检查点，检查点会将数据存储在HDFS中，该RDD的所有父RDD依赖都会被移除。 - 数据分片[partition]的高度弹性
可以根据业务的特征，动态调整数据分片的个数，提升整体的应用执行效率。
数据集
- RDD 可以不保存具体数据, 只保留创建自己的必备信息, 例如依赖和计算函数
- RDD 也可以缓存起来, 相当于存储具体数据
特点
RDD 是混合型的编程模型, 可以支持迭代计算, 关系查询, MapReduce, 流计算
数据集
RDD 是只读的
RDD 是只读的, 不允许任何形式的修改. 虽说不能因为 RDD 和 HDFS 是只读的, 就认为分布式存储系统必须设计为只读的. 但是设计为只读的, 会显著降低问题的复杂度, 因为 RDD 需要可以容错, 可以惰性求值, 可以移动计算, 所以很难支持修改.
- RDD2 中可能没有数据, 只是保留了依赖关系和计算函数, 那修改啥?
- 如果因为支持修改, 而必须保存数据的话, 怎么容错?
- 如果允许修改, 如何定位要修改的那一行? RDD 的转换是粗粒度的, 也就是说, RDD 并不感知具体每一行在哪.
RDD 可以包含多个分区.RDD 作为数据结构, 本质上是一个只读的分区记录集合. 一个 RDD 可以包含多个分区, 每个分区就是一个 DataSet 片段.
RDD 之间可以相互依赖, 如果 RDD 的每个分区最多只能被一个子 RDD 的一个分区使用，则称之为窄依赖, 若被多个子 RDD 的分区依赖，则称之为宽依赖. 不同的操作依据其特性, 可能会产生不同的依赖. 例如 map 操作会产生窄依赖, 而 join 操作则产生宽依赖.
RDD 是可以容错的
- RDD 的容错有两种方式
保存 RDD 之间的依赖关系, 以及计算函数, 出现错误重新计算直接将 RDD 的数据存放在外部存储系统, 出现错误直接读取, Checkpoint
问题1:在集群中运行的前提?
- 必须可以分解为多个并发计算的部分
- 每部分可以在不同处理器上执行
- 需要一个共享内存的机制
- 需要一个总体上的协作机制来调度
问题2:如果放在集群中的话, 可能要对整个计算任务进行分解, 如何分解?
概述
对于 HDFS 中的文件, 是分为不同的 Block 的在进行计算的时候, 就可以按照 Block 来划分, 每一个 Block 对应一个不同的计算单元
扩展
RDD并没有真实的存放数据, 数据是从 HDFS 中读取的, 在计算的过程中读取即可RDD至少是需要可以 分片 的, 因为HDFS中的文件就是分片的,RDD分片的意义在于表示对源数据集每个分片的计算,RDD可以分片也意味着 可以并行计算
问题3:移动数据不如移动计算是一个基础的优化, 如何做到?
每一个计算单元需要记录其存储单元的位置, 尽量调度过去
问题4:在集群中运行, 需要很多节点之间配合, 出错的概率也更高, 出错了怎么办?
- 备份机制
- 重新计算(前提,要记录依赖关系)
问题5:假如任务特别复杂, 流程特别长, 有很多 RDD 之间有依赖关系, 如何优化?
记录数据集的状态
- 缓存
- Checkpoint
总结: RDD 的五大属性
Partition List分片列表, 记录 RDD 的分片, 可以在创建 RDD 的时候指定分区数目, 也可以通过算子来生成新的 RDD 从而改变分区数目Compute Function为了实现容错, 需要记录 RDD 之间转换所执行的计算函数RDD DependenciesRDD 之间的依赖关系, 要在 RDD 中记录其上级 RDD 是谁, 从而实现容错和计算Partitioner分区函数为了执行 Shuffled 分发(Hash)操作, 必须要有一个函数用来计算数据应该发往哪个分区Preferred Location优先位置, 为了实现数据本地性操作, 从而移动计算而不是移动存储, 需要记录每个 RDD 分区最好应该放置在什么位置
Transformations算子
代码:
需求
- 给定一个网站的访问记录, 俗称 Access log
- 计算其中出现的独立 IP, 以及其访问的次数
@Testdef tets1(): Unit = {//添加本地文件val tuples: Array[(String, Int)] = sc.textFile("dataset/access_log_sample.txt")//切割取第一个计数1.map(x => (x.split(" ")(0), 1))//判断是否为空(以后常用).filter(li => StringUtils.isNotBlank(li._1))//统计.reduceByKey((c, a) => c + a)//排序,第二个参数决定是否降序.sortBy(i => i._2, false)//取TOP值.take(10)//遍历打印tuples.foreach(println(_))
mapPartitions(List[T] ⇒ List[U])
RDD[T] ⇒ RDD[U] 和 map 类似, 但是针对整个分区的数据转换
/*** mapPartitions 和 map 算子是一样的, 只不过 map 是针对每一条数据进行转换, mapPartitions 针对一整个分区的数据进行转换* 所以:* 1. map 的 func 参数是单条数据, mapPartitions 的 func 参数是一个集合(一个分区整个所有的数据)* 2. map 的 func 返回值也是单条数据, mapPartitions 的 func 返回值是一个集合*/@Testdef mapPartitions(): Unit = {// 1. 数据生成// 2. 算子使用// 3. 获取结果sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 2).mapPartitions(iter => {iter.foreach(item => println(item))iter}).collect()}//5//3//1//4//6//2
mapPartitionsWithIndex
和 mapPartitions 类似, 只是在函数中增加了分区的 Index
/*** mapPartitionsWithIndex 和 mapPartitions 的区别是 func 中多了一个参数, 是分区号*/@Testdef mapPartitionsWithIndex(): Unit = {sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 2).mapPartitionsWithIndex( (index, iter) => {println("index: " + index)iter.foreach(item => println(item))iter} ).collect()}//index:2//index:0//index:1//1//5//2//3//4//6
reduceByKey((V, V) ⇒ V, numPartition)
@Testdef reduceByKey(): Unit ={sc.parallelize(Seq(("a",2),("b",3),("a",1))).reduceByKey(_ + _).collect().foreach(println(_))}//(a,3)//(b,3)
注意点
- ReduceByKey 只能作用于 Key-Value 型数据, Key-Value 型数据在当前语境中特指 Tuple2
- ReduceByKey 是一个需要 Shuffled 的操作
- 和其它的 Shuffled 相比, ReduceByKey是高效的, 因为类似 MapReduce 的, 在 Map 端有一个 Cominer, 这样 I/O 的数据便会减少
groupByKey()
@Testdef groupByKey(): Unit ={sc.parallelize(Seq(("aa",2),("bb",3),("aa",2))).groupByKey().collect().foreach(println(_))}//(bb,CompactBuffer(3))//(aa,CompactBuffer(2, 2))
作用
GroupByKey 算子的主要作用是按照 Key 分组, 和 ReduceByKey 有点类似, 但是 GroupByKey 并不求聚合, 只是列举 Key 对应的所有 Value
注意点
GroupByKey 是一个 ShuffledGroupByKey 和 ReduceByKey 不同, 因为需要列举 Key 对应的所有数据, 所以无法在 Map 端做 Combine, 所以 GroupByKey 的性能并没有 ReduceByKey 好
combineByKey()
@Testdef conbineByKey(): Unit ={val rdd = sc.parallelize(Seq(("zhangsan", 99.0),("zhangsan", 96.0),("lisi", 97.0),("lisi", 98.0),("zhangsan", 97.0),("wangwu",76.0),("lisi",56.0),("zhangsan",60.0)))val unit: RDD[(String, (Double, Int))] = rdd.combineByKey(//参数一:针对每一个分区中每一个key的第一个值createCombiner = (curr: Double) => (curr, 1),//(value,1),//参数二:针对每个分区中合并相同的key的数据mergeValue = (curr: (Double, Int), nextValue: Double) => (curr._1 + nextValue, curr._2 + 1),//(x:(Int,Int),y:(Int,Int))=>(x._1+y,x._2+1),//参数三:针对所有分区中第二个函数的结果进行合并mergeCombiners = (curr: (Double, Int), agg: (Double, Int)) => (curr._1 + agg._1, curr._2 + agg._2)//(x:(Int,Int),y:(Int,Int))=>(x._1+y._1,x._2+y._2).map(x=>(x._1,x._2._1/x._2._2)).foreach(println(_)))val unit1: RDD[(String, Double)] = unit.map(x=>(x._1,x._2._1/x._2._2))unit1.collect().foreach(println(_))}
作用
对数据集按照 Key 进行聚合
调用
combineByKey(createCombiner, mergeValue, mergeCombiners, [partitioner], [mapSideCombiner], [serializer])参数
createCombiner将 Value 进行初步转换mergeValue在每个分区把上一步转换的结果聚合mergeCombiners在所有分区上把每个分区的聚合结果聚合partitioner可选, 分区函数mapSideCombiner可选, 是否在 Map 端 Combineserializer序列化器注意点
combineByKey的要点就是三个函数的意义要理解groupByKey,reduceByKey的底层都是combineByKey
aggregateByKey()
@Testdef aggregateByKey(): Unit ={val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))//rdd.aggregateByKey(zeroValue)(seqOp, combOp)//zeroValue:指定初始值//seqOp:作用与每一个元素,根据初始值,进行计算//combOp:将seqOp处理过的结果进行聚合rdd.aggregateByKey(0.8)((zeroValue,item)=> item *zeroValue,(curr,agg)=>curr + agg).collect().foreach(println(_))}//(电脑,16.0)//(手机,20.0)
作用
聚合所有 Key 相同的 Value, 换句话说, 按照 Key 聚合 Value
调用
rdd.aggregateByKey(zeroValue)(seqOp, combOp)参数
zeroValue初始值seqOp转换每一个值的函数comboOp将转换过的值聚合的函数
注意点 * 为什么需要两个函数? aggregateByKey 运行将一个RDD[(K, V)]聚合为RDD[(K, U)], 如果要做到这件事的话, 就需要先对数据做一次转换, 将每条数据从V转为U, seqOp就是干这件事的 ** 当seqOp的事情结束以后, comboOp把其结果聚合
- 和 reduceByKey 的区别::
- aggregateByKey 最终聚合结果的类型和传入的初始值类型保持一致
- reduceByKey 在集合中选取第一个值作为初始值, 并且聚合过的数据类型不能改变
foldByKey(zeroValue)((V, V) ⇒ V)
/*** foldByKey 和 Spark 中的 reduceByKey 的区别是可以指定初始值* foldByKey 和 Scala 中的 foldLeft 或者 foldRight 区别是, 这个初始值作用于每一个数据*/@Testdef foldByKey(): Unit ={sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1))).foldByKey(zeroValue = 10)((curr,agg)=>curr+agg).collect().foreach(println(_))}//(a,22)//(b,11)
作用
和 ReduceByKey 是一样的, 都是按照 Key 做分组去求聚合, 但是 FoldByKey 的不同点在于可以指定初始值
调用
foldByKey(zeroValue)(func)参数
zeroValue初始值funcseqOp 和 combOp 相同, 都是这个参数注意点
FoldByKey 是 AggregateByKey 的简化版本, seqOp 和 combOp 是同一个函数FoldByKey 指定的初始值作用于每一个 Value
join(other, numPartitions)
@Testdef join(): Unit ={val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 1)))val rdd2 = sc.parallelize(Seq(("a", 10), ("a", 11), ("a", 12)))rdd1.join(rdd2).collect().foreach(println(_))}//(a,(1,10))//(a,(1,11))//(a,(1,12))//(a,(2,10))//(a,(2,11))//(a,(2,12))
作用
将两个 RDD 按照相同的 Key 进行连接
调用
join(other, [partitioner or numPartitions])参数
other其它 RDDpartitioner or numPartitions可选, 可以通过传递分区函数或者分区数量来改变分区注意点
Join 有点类似于 SQL 中的内连接, 只会再结果中包含能够连接到的 KeyJoin 的结果是一个笛卡尔积形式, 例如
"a", 1), ("a", 2和"a", 10), ("a", 11的 Join 结果集是"a", 1, 10), ("a", 1, 11), ("a", 2, 10), ("a", 2, 11
sortBy(ascending, numPartitions)
@Testdef sortBy(): Unit ={val rdd1 = sc.parallelize(Seq(("a", 3), ("b", 2), ("c", 1)))rdd1.sortBy(_._1).collect().foreach(println(_))println("=============================")rdd1.sortBy(_._2).collect().foreach(println(_))println("=============================")rdd1.sortBy(_._1,false).collect().foreach(println(_))}//(a,3)//(b,2)//(c,1)//=============================//(c,1)//(b,2)//(a,3)//=============================//(c,1)//(b,2)//(a,3)
作用
- 排序相关相关的算子有两个, 一个是
sortBy, 另外一个是sortByKey
调用
sortBy(func, ascending, numPartitions)
参数
func通过这个函数返回要排序的字段ascending是否升序numPartitions分区数
注意点
- 普通的 RDD 没有
sortByKey, 只有 Key-Value 的 RDD 才有 sortBy可以指定按照哪个字段来排序,sortByKey直接按照 Key 来排序
partitionBy(partitioner)
使用用传入的 partitioner 重新分区, 如果和当前分区函数相同, 则忽略操作
重写hashpartitioner
//创建类继承patitioner//引用后传参(1:new defaultp)//conf.set
repartition(numPartitions)
重新分区
/*** repartition 进行重分区的时候, 默认是 Shuffle 的* coalesce 进行重分区的时候, 默认是不 Shuffle 的, coalesce 默认不能增大分区数*/@Testdef partitioning(): Unit = {val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5), 2)// repartition// println(rdd.repartition(5).partitions.size)// println(rdd.repartition(1).partitions.size)// coalesceprintln(rdd.coalesce(5, shuffle = true).partitions.size)}
repartition 算子无论是增加还是减少分区都是有效的, 因为本质上 repartition 会通过 shuffle 操作把数据分发给新的 RDD 的不同的分区, 只有 shuffle 操作才可能做到增大分区数, 默认情况下, 分区函数是 RoundRobin, 如果希望改变分区函数, 也就是数据分布的方式, 可以通过自定义分区函数来实现
Action算子
reduce( (T, T) ⇒ U )
作用
- 对整个结果集规约, 最终生成一条数据, 是整个数据集的汇总
调用
reduce( (currValue[T], agg[T]) ⇒ T )
注意点
- reduce 和 reduceByKey 是完全不同的, reduce 是一个 action, 并不是 Shuffled 操作
- 本质上 reduce 就是现在每个 partition 上求值, 最终把每个 partition 的结果再汇总
collect()
以数组的形式返回数据集中所有元素
count()
返回元素个数
countByKey()
作用
- 求得整个数据集中 Key 以及对应 Key 出现的次数
注意点
- 返回结果为
Map(key → count) - 常在解决数据倾斜问题时使用, 查看倾斜的 Key
RDD 的 Shuffle 和分区
分区的作用
RDD 使用分区来分布式并行处理数据, 并且要做到尽量少的在不同的 Executor 之间使用网络交换数据, 所以当使用 RDD 读取数据的时候, 会尽量的在物理上靠近数据源.
分区和 Shuffle 的关系
分区的主要作用是用来实现并行计算, 本质上和 Shuffle 没什么关系, 但是往往在进行数据处理shuffle的时候,这个时候因为这些 Key 相同的 Value 可能会坐落于不同的分区, 于是理解分区才能理解 Shuffle 的根本原理
shuffle
reduceByKey 这个算子本质上就是先按照 Key 分组, 后对每一组数据进行 reduce, 所面临的挑战就是 Key 相同的所有数据可能分布在不同的 Partition 分区中, 甚至可能在不同的节点中, 但是它们必须被共同计算.
为了让来自相同 Key 的所有数据都在 reduceByKey 的同一个 reduce 中处理, 需要执行一个 all-to-all 的操作, 需要在不同的节点(不同的分区)之间拷贝数据, 必须跨分区聚集相同 Key 的所有数据, 这个过程叫做 Shuffle.
Shuffle 操作的特点
- 只有
Key-Value型的 RDD 才会有 Shuffle 操作, 例如RDD[(K, V)], 但是有一个特例, 就是repartition算子可以对任何数据类型 Shuffle - 早期版本 Spark 的 Shuffle 算法是
Hash base shuffle, 后来改为Sort base shuffle, 更适合大吞吐量的场景
Spark 的 Shuffle 发展大致有两个阶段: Hash base shuffle 和 Sort base shuffle
Hash base shuffle
大致的原理是分桶, 假设 Reducer 的个数为 R, 那么每个 Mapper 有 R 个桶, 按照 Key 的 Hash 将数据映射到不同的桶中, Reduce 找到每一个 Mapper 中对应自己的桶拉取数据.假设 Mapper 的个数为 M, 整个集群的文件数量是
M * R, 如果有 1,000 个 Mapper 和 Reducer, 则会生成 1,000,000 个文件, 这个量非常大了.过多的文件会导致文件系统打开过多的文件描述符, 占用系统资源. 所以这种方式并不适合大规模数据的处理, 只适合中等规模和小规模的数据处理, 在 Spark 1.2 版本中废弃了这种方式.
Sort base shuffle
对于 Sort base shuffle 来说, 每个 Map 侧的分区只有一个输出文件, Reduce 侧的 Task 来拉取, 大致流程如下Map 侧将数据全部放入一个叫做 AppendOnlyMap 的组件中, 同时可以在这个特殊的数据结构中做聚合操作
然后通过一个类似于 MergeSort 的排序算法 TimSort 对 AppendOnlyMap 底层的 Array 排序先按照 Partition ID 排序, 后按照 Key 的 HashCode 排序
最终每个 Map Task 生成一个 输出文件, Reduce Task 来拉取自己对应的数据
从上面可以得到结论, Sort base shuffle 确实可以大幅度减少所产生的中间文件, 从而能够更好的应对大吞吐量的场景, 在 Spark 1.2 以后, 已经默认采用这种方式.
但是需要大家知道的是, Spark 的 Shuffle 算法并不只是这一种, 即使是在最新版本, 也有三种 Shuffle 算法, 这三种算法对每个 Map 都只产生一个临时文件, 但是产生文件的方式不同, 一种是类似 Hash 的方式, 一种是刚才所说的 Sort, 一种是对 Sort 的一种优化(使用 Unsafe API 直接申请堆外内存)
缓存的意义
多次使用 RDD,提高效率
缓存相关的 API
可以使用 cache 方法进行缓存
val interimRDD = sc.textFile("dataset/access_log_sample.txt").map(item => (item.split(" ")(0), 1)).filter(item => StringUtils.isNotBlank(item._1)).reduceByKey((curr, agg) => curr + agg).cache()
也可以使用 persist 方法进行缓存
val interimRDD = sc.textFile("dataset/access_log_sample.txt").map(item => (item.split(" ")(0), 1)).filter(item => StringUtils.isNotBlank(item._1)).reduceByKey((curr, agg) => curr + agg).persist(StorageLevel.MEMORY_ONLY)
缓存其实是一种空间换时间的做法, 会占用额外的存储资源, 如何清理?
val interimRDD = sc.textFile("dataset/access_log_sample.txt").map(item => (item.split(" ")(0), 1)).filter(item => StringUtils.isNotBlank(item._1)).reduceByKey((curr, agg) => curr + agg).persist()interimRDD.unpersist()
根据缓存级别的不同, 缓存存储的位置也不同, 但是使用 unpersist 可以指定删除 RDD 对应的缓存信息, 并指定缓存级别为 NONE
缓存级别
其实如何缓存是一个技术活, 有很多细节需要思考, 如下
- 是否使用磁盘缓存?
- 是否使用内存缓存?
- 是否使用堆外内存?
- 缓存前是否先序列化?
- 是否需要有副本?
| 缓存级别 | userDisk是否使用磁盘 | useMemory是否使用内存 | useOffHeap是否使用堆外内存 | deserialized是否以反序列化形式存储 | replication副本数 |
|---|---|---|---|---|---|
NONE | false | false | false | false | 1 |
DISK_ONLY | true | false | false | false | 1 |
DISK_ONLY_2 | true | false | false | false | 2 |
MEMORY_ONLY | false | true | false | true | 1 |
MEMORY_ONLY_2 | false | true | false | true | 2 |
MEMORY_ONLY_SER | false | true | false | false | 1 |
MEMORY_ONLY_SER_2 | false | true | false | false | 2 |
MEMORY_AND_DISK | true | true | false | true | 1 |
MEMORY_AND_DISK | true | true | false | true | 2 |
MEMORY_AND_DISK_SER | true | true | false | false | 1 |
MEMORY_AND_DISK_SER_2 | true | true | false | false | 2 |
OFF_HEAP | true | true | true | false | 1 |
如何选择分区级别
- Spark 的存储级别的选择，核心问题是在 memory 内存使用率和 CPU 效率之间进行权衡。建议按下面的过程进行存储级别的选择:
- 如果您的 RDD 适合于默认存储级别（MEMORY_ONLY），leave them that way。这是 CPU 效率最高的选项，允许 RDD 上的操作尽可能快地运行.
- 如果不是，试着使用 MEMORY_ONLY_SER 和 selecting a fast serialization library 以使对象更加节省空间，但仍然能够快速访问。(Java和Scala)
- 不要溢出到磁盘，除非计算您的数据集的函数是昂贵的，或者它们过滤大量的数据。否则，重新计算分区可能与从磁盘读取分区一样快.
- 如果需要快速故障恢复，请使用复制的存储级别（例如，如果使用 Spark 来服务 来自网络应用程序的请求）。All 存储级别通过重新计算丢失的数据来提供完整的容错能力，但复制的数据可让您继续在 RDD 上运行任务，而无需等待重新计算一个丢失的分区.
Checkpoint
Checkpoint 的作用
Checkpoint 的主要作用是斩断 RDD 的依赖链, 并且将数据存储在可靠的存储引擎中, 例如支持分布式存储和副本机制的 HDFS.
Checkpoint 的方式
可靠的 将数据存储在可靠的存储引擎中, 例如 HDFS
本地的 将数据存储在本地
Checkpoint 和 Cache 的区别
Checkpoint 可以保存数据到 HDFS 这类可靠的存储上, Persist 和 Cache 只能保存在本地的磁盘和内存中
Checkpoint 可以斩断 RDD 的依赖链, 而 Persist 和 Cache 不行
因为 CheckpointRDD 没有向上的依赖链, 所以程序结束后依然存在, 不会被删除. 而 Cache 和 Persist 会在程序结束后立刻被清除.
Checkpoint的使用
val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")val sc = new SparkContext(conf)sc.setCheckpointDir("checkpoint")val interimRDD = sc.textFile("dataset/access_log_sample.txt").map(item => (item.split(" ")(0), 1)).filter(item => StringUtils.isNotBlank(item._1)).reduceByKey((curr, agg) => curr + agg)//在checkpoint之前先cache保守操作,因为checkpoint会重新计算整个RDD的数据.cache()interimRDD.checkpoint()interimRDD.collect().foreach(println(_))sc.stop()
在使用 Checkpoint 之前需要先设置 Checkpoint 的存储路径, 而且如果任务在集群中运行的话, 这个路径必须是 HDFS 上的路径
应该在 checkpoint 之前先 cache 一下, 因为 checkpoint 会重新计算整个 RDD 的数据然后再存入 HDFS 等地方.
所以上述代码中如果 checkpoint 之前没有 cache, 则整个流程会被计算两次, 一次是 checkpoint, 另外一次是 collect