成都网站建设网站建设,阜阳水建公司官网,网站建设与管理介绍,南京 外贸网站建设Spark
一、Spark简介
Spark 是一种由 Scala 语言开发的快速、通用、可扩展的大数据分析引擎Spark Core 中提供了 Spark 最基础与最核心的功能Spark SQL 是 Spark 用来操作结构化数据的组件。通过 Spark SQL#xff0c;用户可以使用 SQL 或者 Apache Hive 版本的 SQL 方言用户可以使用 SQL 或者 Apache Hive 版本的 SQL 方言HQL来查询数据。Spark Streaming 是 Spark 平台上针对实时数据进行流式计算的组件提供了丰富的处理数据流的API。
Spark 和Hadoop 的根本差异是多个作业之间的数据通信问题 : Spark 多个作业之间数据通信是基于内存而 Hadoop 是基于磁盘。
hadoop file----map----data----reduce----file----map…
spark file----map----data----reduce----memory----map…
二、Spark运行架构 Spark独立部署下其Master相当于ResourceManagerWorker相当于NodeManager
Yarn模式下直接由ResourceManager进行调度Executor运行在NodeManager中
1. 两类工作节点Driver与Worker
——驱动器节点Driver
Driver 负责实际代码的执行工作在 Spark 作业执行时主要负责
➢ 将用户程序转化为作业job
➢ 在 Executor 之间调度任务(task)
➢ 跟踪 Executor 的执行情况
➢ 通过 UI 展示查询运行情况
——工作节点Worker
Spark Executor 是集群中工作节点Worker中的一个 JVM 进程负责在 Spark 作业中运行具体任务Task任务彼此之间相互独立。Spark 应用启动时Executor 节点被同时启动并且始终伴随着整个 Spark 应用的生命周期而存在。如果有 Executor 节点发生了 故障或崩溃Spark 应用也可以继续执行会将出错节点上的任务调度到其他 Executor 节点 上继续运行。
➢ 负责运行组成 Spark 应用的任务并将结果返回给驱动器进程
➢ 它们通过自身的块管理器Block Manager为用户程序中要求缓存的 RDD 提供内存式存储。RDD 是直接缓存在Executor进程内的因此任务可以在运行时充分利用缓存数据加速运算。
2. Yarn模式下两种部署方式Client与Cluster
——Client 模式将用于监控和调度的Driver模块在客户端执行而不是在Yarn中
➢ Driver 在任务提交的本地机器上运行
➢ Driver 启动后会和 ResourceManager 通讯申请启动 ApplicationMaster
➢ ResourceManager分配container在合适的NodeManager上启动 ApplicationMaster负责向ResourceManager申请Executor内存
➢ ResourceManager接到ApplicationMaster 的资源申请后会分配 container然后 ApplicationMaster 在资源分配指定的 NodeManager 上启动 Executor 进程
➢ Executor 进程启动后会向 Driver 反向注册Executor 全部注册完成后 Driver 开始执行 main 函数
➢ 执行到 Action 算子时触发一个 Job并根据宽依赖开始划分 stage每个stage生成对应TaskSet之后将task分发到各个Executor上执行。
——Cluster 模式将用于监控和调度的 Driver 模块启动在 Yarn 集群资源中执行。
➢ 在 YARN Cluster 模式下任务提交后会和 ResourceManager 通讯申请启动 ApplicationMaster
➢ 随后 ResourceManager 分配 container在合适的 NodeManager 上启动 ApplicationMaster 此时的 ApplicationMaster 就是 Driver。
➢ Driver 启动后向 ResourceManager 申请 Executor 内存ResourceManager 接到 ApplicationMaster 的资源申请后会分配 container然后在合适的 NodeManager 上启动 Executor 进程
➢ Executor 进程启动后会向 Driver 反向注册Executor 全部注册完成后 Driver 开始执行 main 函数
➢ 之后执行到 Action 算子时触发一个 Job并根据宽依赖开始划分 stage每个 stage 生 成对应的 TaskSet之后将 task 分发到各个 Executor 上执行。
三、Spark核心编程
Spark封装了三大数据结构用于 处理不同的应用场景。三大数据结构分别是
➢ RDD : 弹性分布式数据集
➢ 累加器分布式共享只写变量
➢ 广播变量分布式共享只读变量
1RDDResilient Distributed Dataset 弹性分布式数据集是Spark中最基本的数据抽象它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中后续的查询能够重用工作集这极大地提升了查询速度。
上述图中四个步骤的具体关系如图所示RDD与IO流类似都属于装饰者模式。如下图中每一个RDD都是封装关系。 RDD在运行流程中不存储数据 RDD只有在执行collect函数时才会真正开始执行业务逻辑 RDD五大属性 分区列表用于执行并行任务 分区计算函数用于对分区进行计算 RDD间依赖关系如上图所示的封装关系 分区器对KV类型的数据进行自定义分区 首选位置保障计算任务下发到数据近邻节点
RDD运行流程
RDD算子方法
注意1Spark中涉及到shuffle操作必须进行落盘处理因为在内存中等待可能造成内存溢出。解释——比如groupByKey操作下分区一的数据已经处理完毕而分区二的数据还未处理完成就会等待二中的数据全部处理完成才算操作完成。这个等待过程可能消耗大量内存。
2从计算的角度, 算子以外的代码都是在 Driver 端执行, 算子里面的代码都是在 Executor 端执行。那么在 scala 的函数式编程中就会导致算子内经常会用到算子外的数据这样就 形成了闭包的效果如果使用的算子外的数据无法序列化就意味着无法传值给 Executor 端执行就会发生错误所以需要在执行任务计算前检测闭包内的对象是否可以进行序列化这个操作我们称之为闭包检测。
3涉及shuffle的操作会将分区的数据打乱重组。所以针对操作如果涉及shuffle我们将其分为不同执行阶段。每一个阶段的最后一个RDD的分区个数就是执行的任务个数。
一个SparkContext对应一个Application一个Action算子对应一个Job一个shuffle操作对应一个新的Stage一个Stage中最后一个RDD的分区个数对应Task个数
——Value类型
方法名作用map(func)将处理的数据逐条进行映射转换这里的转换可以是类型的转换也可以是值的转换每次处理一条数据mapPartitions(Iterator)将待处理的数据以分区为单位发送到计算节点进行处理这里的处理是指可以进行任意的处理可以理解为缓冲区每次将一个分区的数据引用进内存mapPartitionsWithIndex(Iterator)将待处理的数据以分区为单位发送到计算节点进行处理在处理时同时可以获取当前分区索引flatMap(func)将处理的数据进行扁平化后再进行映射处理所以算子也称之为扁平映射比如将一个完整的数组拆分成一个一个数再进行映射。该函数需要的是一个List也即map的运算结果glom()将同一个分区的数据直接转换为相同类型的内存数组进行处理分区不变groupby(func)将数据根据指定的规则进行分组, 数据会被打乱重新组合我们将这样的操作称之为 shuffle。极限情况下数据可能被分在同一个分区中一个组的数据在一个分区中但是并不是说一个分区中只有一个组即分组与分区没有必然联系filter(func)数据根据指定的规则进行筛选符合规则的数据保留不符合规则的数据丢弃。 当数据进行筛选过滤后分区不变但是分区内的数据可能不均衡生产环境下可能会出现数据倾斜。sample()根据指定的规则从数据集中抽取数据distinct()将数据集中重复的数据去重coalesce()根据数据量缩减分区用于大数据集过滤后提高小数据集的执行效率 当 spark 程序中存在过多的小任务的时候可以通过 coalesce 方法收缩合并分区减少 分区的个数减小任务调度成本repartition()coalesce一般用于减少分区repartition一般用于扩大分区sortBy(func)在排序之前可以将数据通过 f 函数进行处理之后按照 f 函数处理 的结果进行排序默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一 致。中间存在 shuffle 的过程因为会重排数据序列所以分区会改变mapValue(value)key不变只对value进行操作
——双Value类型
方法名作用intersection()对源 RDD 和参数 RDD 求交集后返回一个新的 RDDunion()对源 RDD 和参数 RDD 求并集后返回一个新的 RDDsubtract()以一个 RDD 元素为主去除两个 RDD 中重复元素将其他元素保留下来zip()将两个 RDD 中的元素以键值对的形式进行合并。其中键值对中的 Key 为第 1 个 RDD 中的元素Value 为第 2 个 RDD 中的相同位置的元素。
——key-value类型
方法名作用partitionBy()将数据按照指定 Partitioner 重新进行分区reduceByKey()可以将数据按照相同的 Key 对 Value 进行聚合。支持分区内预聚合操作分区内聚合再进行区间聚合减少shuffle落盘和再读取的数据量。groupByKey()将数据源的数据根据 key 对 value 进行分组(会导致数据分区被打乱。即存在shuffle操作)aggregateByKey()将数据根据不同的规则进行分区内计算和分区间计算foldByKey()当分区内计算规则和分区间计算规则相同时aggregateByKey 就可以简化为 foldByKeycombineByKey()对 key-value 型 rdd 进行聚集操作的聚集函数aggregation functionsortByKey()在一个(K,V)的 RDD 上调用K 必须实现 Ordered 接口(特质)返回一个按照 key 进行排序的rddjoin()在类型为(K,V)和(K,W)的 RDD 上调用返回一个相同 key 对应的所有元素连接在一起的 (K,(V,W))的 RDDleftOuterJoin()SQL 语句的左外连接rightOuterJoin()右外连接cogrpup()在类型为(K,V)和(K,W)的 RDD 上调用先进行分组后进行左外连接
——行动算子触发整个任务的执行
方法名作用reduce()聚集 RDD 中的所有元素先聚合分区内数据再聚合分区间数据collect()在驱动程序中以数组 Array 的形式返回数据集的所有元素count()返回 RDD 中元素的个数first()返回 RDD 中的第一个元素take()返回 RDD 中的第一个元素takeOrdered()返回该 RDD 排序后的前 n 个元素组成的数组aggregate()分区的数据通过初始值和分区内的数据进行聚合然后再和初始值进行分区间的数据聚合fold()折叠操作aggregate 的简化版操作countByKey()统计每种 key 的个数saveAsTextFile()、saveAsObjectFile(、saveAsSequenceFile(将数据保存到不同格式的文件中:文本文件、对象序列化后保存的文件、二进制形式的key-value平面文件foreach()分布式遍历 RDD 中的每一个元素调用指定函数foreachPartition()按分区得到一个RDD迭代序列
rddList.foreach { rdd {val conn JDBCUtil.getConnection}
}
//上述操作将为每一个RDD建立一个数据库连接对象浪费资源
rddList.foreach { val conn JDBCUtil.getConnectionrdd {}
}
//上述操作foreach作为一个算子之外的操作会在Driver端执行而之内的将在Executor端执行这样涉及序列化操作而连接对象不支持序列化
rddList.foreachPartition { iter {val conn JDBCUtil.getConnectioniter.foreach(iter {})}
}
//所以我们使用foreachPartition返回一个分区列表而在分区列表中我们可以为每个分区建立一个连接对象减少资源消耗RDD依赖关系 RDD不会保存数据但会存储RDD间的血缘关系如RDD2会保存2-4的所有操作提高容错性 旧RDD的每个分区只被新RDD的一个分区使用(OneToOneDependency窄依赖上图)旧RDD的每个分区被新RDD的多个分区使用(ShuffleDependency宽依赖下图) 方法名作用toDebugString()RDD保存的血缘关系dependenciesRDD保存的依赖关系
RDD持久化
val rdd sc.makeRDD(List(1,2,3,4), 2)
val res1 rdd.map(num num 1)
val res2 res1.map(num num 1)
val res3 res1.map(num num 2) 针对以上代码发现res2res3结果都是正确的这是因为res3在执行时又重新执行了一遍rdd-res1-res3的操作。结论如下 RDD不存放数据只存放操作 RDD对象可以重用但是数据不可以重用 如果多次用同一个RDD则每一次调用都会从头计算一遍 因此如果需要重复利用一个RDD需要对其进行缓存。 方法名作用RDD.cache()存放RDD数据到cache缓存RDD.persist()更改缓存级别
RDD检查点
所谓的检查点其实就是通过将 RDD 中间结果写入磁盘。
方法名作用RDD.checkpointRDD数据落盘
1Cache 缓存只是将数据保存起来不切断血缘依赖。Checkpoint 检查点切断血缘依赖。因为checkpoint会将数据落盘而血缘关系正是为了记录操作从而恢复数据所以无需记录血缘。
2Cache 缓存的数据通常存储在磁盘、内存等地方可靠性低。Checkpoint 的数据通常存储在 HDFS 等容错、高可用的文件系统可靠性高。
3建议对 checkpoint()的 RDD 使用 Cache 缓存这样 checkpoint 的 job 只需从 Cache 缓存中读取数据即可否则checkpoint为了得到数据需要再从头计算一次 RDD。
RDD分区器
Spark 目前支持 Hash 分区和 Range 分区和用户自定义分区。
Hash 分区对于给定的 key计算其 hashCode,并除以分区个数取余Range 分区将一定范围内的数据映射到一个分区中尽量保证每个分区数据均匀而且分区间有序自定义分区
rdd.partitionBy(new Mypartitioner)
class Mypartitioner extends Partitioner{override def numPartitions: Int 3override def getPartition(key:Any): Int {key match{case my 0case you 1case we 2}}
}2累加器分布式共享只写变量
累加器的作用就是将sum在各Executor执行后传回Driver端再进行整体sum,比如Executor1将sum3Executor2将sum7传回Driver后就是sum3710
注意 只有行动算子可以触发累加器所以累加器一般放在行动算子中 可以自定义累加器1. 继承 AccumulatorV2并设定泛型 2. 重写累加器的抽象方法
val sum1 sc.longAccumulator(name Sum)
//sc.doubleAccumulator sc.collectionAccumulator
rdd.foreach(num {sum1.add(num)
})
println(sum1.value)3广播变量分布式共享只读变量
广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值以供一个或多个 Spark 操作使用。
闭包数据都以Task为单位发送每个任务都包含闭包数据当一个Executor执行了多个Task会导致有大量重复数据可以将任务中的闭包数据保存到内存中实现共享的目的减少空间占用
具体代码可以参考 https://github.com/Ostrich5yw/java4BigData
