网站建设花费录什么费用,网站排名必做阶段性seo策略,皮具网站建设服装网站,wordpress 搭建个人博客前言#xff1a; #x1f49e;#x1f49e;大家好#xff0c;我是书生♡#xff0c;本阶段和大家一起分享和探索大数据技术Spark—RDD#xff0c;本篇文章主要讲述了#xff1a;RDD的依赖、Spark 流程、Shuffle 与缓存等等。欢迎大家一起探索讨论#xff01;#xff0… 前言 大家好我是书生♡本阶段和大家一起分享和探索大数据技术Spark—RDD本篇文章主要讲述了RDD的依赖、Spark 流程、Shuffle 与缓存等等。欢迎大家一起探索讨论 代码是你的画笔创新是你的画布用它们绘出属于你的精彩世界不断挑战无限可能 个人主页⭐ 书生♡ gitee主页♂闲客 专栏主页大数据开发 博客领域大数据开发java编程前端算法Python 写作风格超前知识点干货思路讲解通俗易懂 支持博主关注⭐点赞、收藏⭐、留言 目录 1. RDD缓存和checkpoint1.1 缓存机制1. 2 CheckPoint 机制 2. RDD依赖2.1 窄依赖Narrow Dependency2.2 宽依赖Wide Dependency2.3 管理依赖2.4 日志查看依赖关系和计算流程2.5 划分stage 3.Spark的运行流程(内核调度)4.spark的shuffle过程4.1 shuffle介绍4.2 Shuffle 的过程Shuffle 的影响 4.3 SparkShuffle配置 1. RDD缓存和checkpoint 在 Apache Spark 中缓存也称为持久化和 checkpoint 是两种用于优化性能和容错的重要机制。这两种机制可以帮助减少重复计算提高应用程序的效率。 RDD的缓存和checkpoint机制也是spark计算速度快的原因之一
1.1 缓存机制
定义 缓存是一种将 RDD 的计算结果存储在内存或磁盘上的机制。通过缓存Spark 可以避免重新计算已经处理过的数据从而显著提高应用程序的性能。
缓存是将RDD存储到内存上或者是本地磁盘上Linux缓存是临时持久化操作 用途
提高性能对于需要多次访问的数据集缓存可以避免重复计算显著加快执行速度。内存管理可以根据可用内存和数据的重要性选择合适的存储级别。保证RDD容错性应用程序运行过程中, 可能因为一些原因导致rdd计算失败需要重新计算
应用场景
计算时间长的rdd计算成本昂贵的RDD重复多次使用的RDD 注意点 应用程序结束后会自动清空缓存RDD缓存不会切断RDD之间的依赖关系(缓存的rdd有可能丢失, 丢失后还可以通过依赖关系计算得到)缓存的RDD需要通过aciton算子触发缓存任务, 触发缓存任务后的RDD才是从缓存中获取的触发缓存任务之前, 调用的rdd还是通过依赖关系计算得到的缓存级别缓存的RDD存储在哪里 默认存储在内存中, 也可以设置存储在内存和本地磁盘persist 只是定义了一个缓存任务并不是执行使用unpersist释放缓存 API
cache()默认将 RDD 存储在内存中如果内存不足则溢出到磁盘。persist(storageLevel)允许指定不同的存储级别如 MEMORY_ONLY, MEMORY_AND_DISK, DISK_ONLY 等。is_cached 查看缓存状态进行了缓存返回TRUE反之返回Falseunpersist释放缓存 缓存的级别缓存RDD默认存储在内存中 from pyspark.storagelevel import StorageLevel StorageLevel.DISK_ONLY # 将数据缓存到磁盘上StorageLevel.DISK_ONLY_2 #将数据缓存到磁盘上 保存两份StorageLevel.DISK_ONLY_3 # 将数据缓存到磁盘上 保存三份StorageLevel.MEMORY_ONLY # 将数据缓存到内存 默认StorageLevel.MEMORY_ONLY_2 #将数据缓存到内存 保存两份StorageLevel.MEMORY_AND_DISK # 将数据缓存到内存和磁盘 优先将数据缓存到内存上内存不足可以缓存到磁盘StorageLevel.MEMORY_AND_DISK_2 # 将数据缓存到内存和磁盘StorageLevel.OFF_HEAP # 基本不使用 缓存在系统管理的内存上 jvm内存在系统上运行系统内存StorageLevel.MEMORY_AND_DISK_ESER # 将数据缓存到内存和磁盘 序列化操作按照二进制存储节省空间 案例 这个案例中我们对 rdd_cnt 这个RDD对象进行了缓存操作 rdd_cnt.persist(storageLevelStorageLevel.MEMORY_ONLY) # 创建SparkContext对象
sc SparkContext()# 从HDFS读取文本文件
rdd_file sc.textFile(/data/stu.csv)
# 读取文件后对文件进行分割得到每个学生的信息
rdd_stu rdd_file.map(lambda x: x.split(,))
rdd_map rdd_stu.map(lambda x: (x[0], int(x[1])))# 查看是否对rdd进行缓存操作, 返回True或False
print(rdd_map.is_cached)
# 统计不同年龄段的人数 0-30 为青年30-60为中年大于60为老年
rdd_age rdd_map.map(lambda x: (青年,1) if x[1]30 else (中年,1) if x[1]60 and x[1]30 else (老年,1) )
rdd_cnt rdd_age.reduceByKey(lambda x,y: xy)# 对rdd_reducebykey进行缓存操作, 只是定义了一个缓存任务
# 如果实现对rdd进行缓存, 需要调用action算子触发缓存任务
# 触发缓存任务之前, 调用的当前rdd还是通过依赖关系计算得到的
rdd_cnt.persist(storageLevelStorageLevel.MEMORY_ONLY)
# 查看是否对rdd进行缓存操作, 返回True或False
print(rdd_cnt.is_cached)# 手动释放
rdd_cnt.unpersist()
print(rdd_cnt.is_cached)print(rdd_cnt.collect())内存限制确保有足够的内存来缓存数据否则可能会导致 OutOfMemoryError。 1. 2 CheckPoint 机制
定义 Checkpoint 是一种容错机制用于将 RDD 的数据持久化到可靠的存储系统如 HDFS、S3 等。与缓存不同Checkpoint 不仅用于提高性能还可以在节点故障时恢复数据。
checkpoint将RDD存储到HDFS(分块多副本)中checkpoint的RDD是不会随程序运行结果而被清空checkpoint会永久存储RDD, RDD之间的依赖关系会被删除checkpoint是永久持久化操作
用途
提高计算效率容错当 Spark 应用程序遇到故障时可以从 Checkpoint 恢复数据避免重新计算整个数据流。减少依赖通过 Checkpoint可以减少计算依赖的深度从而降低故障时需要重新计算的数据量。 注意 永久存储在HDFS上程序运行结束不会被删除会切断rdd之间的依赖关系需要通过action算子触发checkpoint任务 API
setCheckpointDir(path)将 RDD 的数据持久化到指定的目录。rdd.checkpoint() 对RDD进行checkpoint操作
使用 checkpoint:将rdd存储在HDFS中使用: ①设置checkpoint目录路径 sc.setCheckpointDir()②rdd.checkpoint() 案例 checkpoint:将rdd存储在HDFS中
使用:①设置checkpoint目录路径 sc.setCheckpointDir() ②rdd.checkpoint()
注意点:①永久存储在HDFS上 ②程序运行结束不会被删除 ③会切断rdd之间的依赖关系 ④需要通过action算子触发checkpoint任务# 统计不同词出现的次数 - 分组聚合操作
from pyspark import SparkContext
# 导入缓存级别类
from pyspark import StorageLevel# 创建sc对象
sc SparkContext()
# 设置checkpoint目录
sc.setCheckpointDir(/checkpoint)
# 读取hdfs文件数据,转换成rdd对象
rdd_words sc.textFile(/data/words.txt)
rdd_flatmap rdd_words.flatMap(lambda x: x.split(,))
rdd_map rdd_flatmap.map(lambda x: (x, 1))
print(rdd_map.is_checkpointed)
rdd_reducebykey rdd_map.reduceByKey(lambda x, y: x y)
# 对rdd_reducebykey进行checkpoint操作, 只是定义了一个checkpoint任务
# 如果实现对rdd进行checkpoint, 需要调用action算子触发checkpoint任务
# 触发checkpoint任务之前, 调用的rdd还是通过依赖关系计算得到的
rdd_reducebykey.checkpoint()
# 查看是否对rdd进行checkpoint操作, 返回True或False
print(rdd_reducebykey.is_checkpointed)
# 根据词出现的次数进行降序操作
# rdd_reducebykey是通过依赖关系计算得到的, 不是从checkpoint中获取的
rdd_sortby rdd_reducebykey.sortBy(lambda x: x[1], ascendingFalse)
print(rdd_sortby.collect())
# rdd_reducebykey是从checkpoint中获取的
rdd_sortby2 rdd_reducebykey.sortBy(lambda x: x[1], ascendingTrue)
print(rdd_sortby2.collect())注意事项 存储空间确保 Checkpoint 目录有足够的存储空间。 性能开销频繁地执行 Checkpoint 会增加额外的 I/O 开销。 2. RDD依赖 在 Apache Spark 中弹性分布式数据集Resilient Distributed Dataset, RDD之间的依赖关系是 Spark 计算模型的核心部分。依赖关系决定了数据的处理顺序和粒度同时也影响着 Spark 作业的执行效率和容错性。 定义 RDD 之间的依赖关系指的是一个 RDD 如何依赖于另一个 RDD 的数据。相邻RDD之间存在的因果关系, 可以称为依赖关系----》新RDD一定是由旧RDD计算得到, RDD1-RDD2-RDD3这种依赖关系决定了数据流的方向和数据处理的顺序。RDD特性之一依赖关系可以保证RDD计算的容错性, 如果rdd因为某些原因计算失败, 可以根据依赖关系重新计算 类型 窄依赖Narrow Dependency每个父 RDD 分区最多被一个子 RDD 分区使用。宽依赖Wide Dependency一个子 RDD 分区可能依赖于多个父 RDD 分区。 影响 窄依赖允许 Spark 在流水线中并行执行任务。宽依赖导致数据重分布增加了计算成本。 2.1 窄依赖Narrow Dependency 定义 窄依赖指的是父 RDD 的每个分区最多被一个子 RDD 分区使用。一对一或者多对一关系窄依赖通常出现在 map、filter、union 等操作中。 触发窄依赖关系的算子 map()flatMap()filter()mapValues()mapPartitions() 特点 窄依赖操作可以并行执行因为它们不需要重新分布数据。通常不会触发 shuffle 操作。 示例 使用 map 操作将每个元素乘以 2rdd sc.parallelize([1, 2, 3, 4, 5])
doubled_rdd rdd.map(lambda x: x * 2)2.2 宽依赖Wide Dependency 定义 宽依赖指的是一个子 RDD 分区可能依赖于多个父 RDD 分区。宽依赖通常出现在 groupByKey、reduceByKey、join 等操作中。 触发宽依赖关系的算子 groupBy()groupByKey()reduceByKey()sortBy()sortByKey()distinct() 特点 宽依赖操作需要触发 shuffle 操作即数据需要在节点间进行重分布。通常会导致较高的计算成本。 示例 使用 reduceByKey 对数据进行聚合rdd sc.parallelize([(1, 2), (1, 3), (2, 4)])
result rdd.reduceByKey(lambda x, y: x y)影响 计算效率 窄依赖通常比宽依赖更高效因为它们不需要 shuffle 数据。宽依赖可能会导致更多的磁盘 I/O 和网络传输从而降低性能。 容错性 RDD 的容错性是通过 lineage 信息来实现的。当数据丢失时Spark 可以根据依赖关系重新计算丢失的数据。 执行计划 Spark 的 DAGScheduler 会根据依赖关系构建执行计划。宽依赖会导致新的 stage 的形成而窄依赖则可以在同一个 stage 内执行。
示例
假设我们有一个简单的 RDD rdd1并执行了一系列的操作来创建新的 RDD。
rdd1 sc.parallelize([1, 2, 3, 4, 5])# 窄依赖
rdd2 rdd1.map(lambda x: x * 2)# 宽依赖
rdd3 rdd2.groupByKey()# 窄依赖
rdd4 rdd3.flatMap(lambda x: x)# 宽依赖
rdd5 rdd4.reduceByKey(lambda x, y: x y)在这个例子中rdd2 和 rdd4 之间的依赖关系是窄依赖而 rdd3 和 rdd5 之间的依赖关系是宽依赖。宽依赖操作如 groupByKey 和 reduceByKey会导致数据重分布而窄依赖操作如 map 和 flatMap则不需要重分布数据。
2.3 管理依赖 通过DAG有向无环图图计算算法管理RDD之间的依赖关系 DAG称为有向无环图(有方向没有闭环), 是图计算中的一种算法DAG有向无环图作用 管理RDD之间的依赖关系, 保证RDD按照依赖关系进行有序地计算根据RDD之间的依赖关系对计算任务划分成多个计算步骤, 每个步骤称为stage阶段 触发宽依赖关系的算子会产生新的stage阶段窄依赖关系的算子计算步骤是在同一个stage阶段进行 2.4 日志查看依赖关系和计算流程
app spark应用程序 appID 就是这个程序的IDAPP Name 就是这个spark程序的名字别名 job - 计算任务(一个app中是可能有多个job), 执行action算子时才会产生job stage - 计算步骤/阶段, DAG根据宽依赖关系划分成多个stage task - task线程任务, 真正执行的计算任务有多少个分区就有多个task线程任务 2.5 划分stage
怎么划分stageDAG根据宽依赖关系划分成多个stage
为甚要划分成多个stage呢
spark的task任务是以线程方式实现多任务计算, 线程多任务会有一个资源抢夺问题, 导致计算不准确spark中同一个stage中的多个task任务是并行计算的, 下一个stage中的多个task任务要想并行计算, 需要等上一个stage计算步骤完成后才能并行计算为什么要等待上一个stage计算完成? 宽依赖是会进行shuffle过程, 数据需要重新洗牌, 等待过程就是洗牌过程 如何划分stage? 查看rdd之间是否存在宽依赖关系触发宽依赖关系的算子通过日志查看DAG有向无环图‘ 注意在一个stage中会有多个线程也就是多个task任务是并行计算那么就会有资源竞争有的任务执行快有的任务执行慢当执行快的任务执行完的时候慢的任务刚刚执行了一旦这个时候通过宽依赖进行计算就会出现数据缺失的的问题因此划分成多个stage让执行快的任务等慢的任务执行完之后一起一起执行宽依赖的算子计算这样子数据就不会缺失了。 3.Spark的运行流程(内核调度) Spark框架中封装了三个Scheduler类完成整个spark的计算过程 Spark的核心是根据RDD来实现的Spark Scheduler则为Spark核心实现的重要一环其作用就是任务调度。Spark的任务调度就是如何组织任务去处理RDD中每个分区的数据根据RDD的依赖关系构建DAG基于DAG划分Stage将每个Stage中的任务发到指定节点运行。基于Spark的任务调度原理可以合理规划资源利用做到尽可能用最少的资源高效地完成任务计算。
DAGScheduler 根据rdd间的依赖关系将提交的job划分成多个stage。对每个stage中的task进行描述task编号task执行的rdd算子 TaskScheduler 获取DAGScheduler提交的task调用SchedulerBackend获取executor的资源信息给task分配资源维护task和executor对应关系管理task任务队列将task给到SchedulerBackend然后由SchedulerBackend分发对应的executor执行 SchedulerBackend 向RM申请资源获取executor信息分发task任务 执行spark应用程序, 创建driver进程, driver进程调用三个scheduler类创建三个scheduler对象schedulerbackend向资源调度工具主服务申请计算资源, 创建executor进程执行了action算子触发计算任务, DAGscheduler根据宽依赖关系划分stage, 同时分析stage中的task描述, 将taskSet提交给TaskschedulerTaskscheduler将计算资源分配给task, 同时维护task和executor之间关系, 管理task执行顺序, 将分配计算资源的task交给schedulerbackendschedulerbackend将task线程任务给到executor进程执行 4.spark的shuffle过程
4.1 shuffle介绍 在 Apache Spark 中Shuffle 是一个关键的概念它涉及到数据的重新分布通常发生在宽依赖操作中例如 groupByKey, reduceByKey, join 等。 mapreduce的shuffle作用 将map计算后的数据传递给reduce使用 mapreduce的shuffle过程 分区排序合并规约 Shuffle 的定义 Shuffle 是指在 Spark 中对数据进行重新分布的过程通常涉及到将数据从一个节点移动到另一个节点。这个过程发生在宽依赖操作中因为这些操作需要将具有相同 key 的数据聚集在一起而这些数据可能最初分布在不同的节点上。 Shuffle 的原因 Shuffle 发生的主要原因是需要将数据重新分布到不同的分区中以便进行聚合或连接等操作。例如在 groupByKey 操作中具有相同 key 的所有元素需要被聚集在一起以进行聚合计算。 作用不同阶段的数据传递 无论是spark shuffle还是mapreduce shuffle本质都是传递数据 spark的shuffle分成两个阶段 map阶段: shuffle write, 将上一个stage的数据保存到磁盘文件中reduce阶段: shuffle read, 将磁盘文件中的数据保存到下一个stage中 spark的shuffle方法类 是spark封装好的处理shuffle的方法 hashshuffle spark1.2版本之前使用, 在spark2.0版本删除hash(key)%分区数结果值…余数, 余数相同的数据放到一起未优化的hashshuffle - 有多少个buffer有多少个磁盘小文件优化后的hashshuffle - 有多少个分区有多少个磁盘小文件 sortshuffle spark2.0/3.0版本使用的都是sortshuffle普通模式 - 使用排序方式将数据划分 将分区数据存储在5M大小的memory中, 从memory取1w条数据进行排序 bypass模式 类似于优化后的hashshufflehash(key)%分区数结果值…余数, 余数相同的数据放到一起 无论是hash还是排序都是将相同key值放在一起处理 [(‘a’,1),(‘b’,2),(‘a’,1)]hash(key)%分区数相同的key数据余数是相同的会放一起交给同一个分区进行处理按照key排序相同key的数据也会放在一起 然后交给同一分区处理 4.2 Shuffle 的过程
Shuffle 的过程可以分为以下几个主要阶段 Map 阶段 Map 阶段通常涉及对输入数据进行转换例如应用 map 或 flatMap 等操作。在宽依赖操作中数据会被标记为需要参与 shuffle。Map 阶段还会进行一些优化例如将部分结果写入本地磁盘。 Shuffle write Map 阶段产生的数据会被写入本地磁盘上的 shuffle 文件。每个 map 任务都会产生一个或多个 shuffle 文件这些文件按 key 进行分区。Shuffle 文件通常会被压缩以节省存储空间和传输时间。 Shuffle read 在 shuffle 读阶段reduce 任务会从所有 map 任务产生的 shuffle 文件中读取数据。读取数据时reduce 任务会根据 key 去定位相应的 shuffle 文件并从中读取数据。数据可能需要在网络上传输这取决于数据的存储位置。 Reduce 阶段 Reduce 阶段处理从 shuffle 文件中读取的数据。对于每个 keyreduce 任务会执行相应的聚合或连接操作。最终结果会被输出到内存或磁盘上。
Shuffle 的影响
Shuffle 过程可能会显著影响 Spark 应用程序的性能因为它涉及到大量的磁盘 I/O 和网络传输。为了减少 shuffle 的影响可以采取以下措施
减少 shuffle 的数量尽量使用窄依赖操作来减少 shuffle 的需求。调整并行度通过设置 spark.sql.shuffle.partitions 来调整 shuffle 的并行度。优化数据分布确保数据在节点之间均匀分布以减少数据倾斜。启用压缩通过启用 shuffle 文件的压缩来减少传输的数据量。使用高效的序列化方式例如使用 Kryo 序列化器来提高序列化和反序列化的效率。
示例 假设我们有一个简单的 RDD我们想要使用 reduceByKey 来计算每个 key 的总和。
from pyspark import SparkContextsc SparkContext(local, Shuffle Example)# 创建一个 RDD
rdd sc.parallelize([(1, 2), (1, 3), (2, 4)])# 使用 reduceByKey 进行聚合
result rdd.reduceByKey(lambda x, y: x y)print(Result:, result.collect())在这个例子中reduceByKey 操作会导致 shuffle因为需要将具有相同 key 的元素聚集在一起。
4.3 SparkShuffle配置
spark.shuffle.file.buffer 参数说明 该参数用于设置shuffle write task的BufferedOutputStream的buffer缓冲大小默认是32K。将数据写到磁盘文件之前会先写入buffer缓冲中待缓冲写满之后才会溢写到磁盘。 2倍 3倍 调优建议如果作业可用的内存资源较为充足的话可以适当增加这个参数的大小比如64k从而减少shuffle write过程中溢写磁盘文件的次数也就可以减少磁盘IO次数进而提升性能。在实践中发现合理调节该参数性能会有1%~5%的提升 spark.reducer.maxSizeInFlight 参数说明 该参数用于设置shuffle read task的buffer缓冲大小而这个buffer缓冲决定了每次能够拉取多少数据。(默认48M) 调优建议如果作业可用的内存资源较为充足的话可以适当增加这个参数的大小比如96m从而减少拉取数据的次数也就可以减少网络传输的次数进而提升性能。在实践中发现合理调节该参数性能会有1%~5%的提升。 spark.shuffle.io.maxRetries and spark.shuffle.io.retryWai spark.shuffle.io.maxRetries shuffle read task从shuffle write task所在节点拉取属于自己的数据时如果因为网络异常导致拉取失败是会自动进行重试的。该参数就代表了可以重试的最大次数。默认是3次 spark.shuffle.io.retryWait 该参数代表了每次重试拉取数据的等待间隔。默认为5s 调优建议一般的调优都是将重试次数调高不调整时间间隔。 spark.shuffle.memoryFraction10 参数说明 该参数代表了Executor 1G内存中分配给shuffle read task进行聚合操作内存比例。 spark.shuffle.manager 参数说明该参数用于设置shufflemanager的类型默认为sort Hashspark1.x版本的默认值HashShuffleManager Sortspark2.x版本的默认值普通机制。当shuffle read task 的数量小于等于200采用bypass机制 spark.shuffle.sort.bypassMergeThreshold200 根据task数量决定sortshuffle的模式task数量小于等于200 就采用bypass task大于200就采用普通模式 参数说明 当ShuffleManager为SortShuffleManager时如果shuffle read task的数量小于这个阈值默认是200则shuffle write过程中不会进行排序操作。 调优建议当你使用SortShuffleManager时如果的确不需要排序操作那么建议将这个参数调大一些 交互式开发中使用
pyspark --master yarn --name shuffle_demo --conf spark.shuffle.sort.bypassMergeThreshold300通过--conf 来配置参数
脚本中配置参数 创建conf对象, 实现spark参数设置对象名类名()调用set()返回对象本身 conf (SparkConf().set(spark.shuffle.sort.bypassMergeThreshold, 300).set(spark.shuffle.io.maxRetries, 5))创建sc对象传递conf对象sc SparkContext(master‘yarn’, appName‘shuffle_demo’, confconf) from pyspark import SparkContext
from pyspark import SparkConf# 创建conf对象, 实现spark参数设置
# 对象名类名()
# 调用set()返回对象本身
conf (SparkConf().set(spark.shuffle.sort.bypassMergeThreshold, 300).set(spark.shuffle.io.maxRetries, 5))# 创建sc对象
# 传递conf对象
sc SparkContext(masteryarn, appNameshuffle_demo, confconf)
# 读取hdfs文件数据,转换成rdd对象
rdd_words sc.textFile(/data/words.txt)
print(rdd_words.take(num3))
rdd_flatmap rdd_words.flatMap(lambda x: x.split(,))
print(rdd_flatmap.collect())
rdd_map rdd_flatmap.map(lambda x: (x, 1))
print(rdd_map.collect())
rdd_reducebykey rdd_map.reduceByKey(lambda x, y: x y)
print(rdd_reducebykey.collect())
# 根据词出现的次数进行降序操作
rdd_sortby rdd_reducebykey.sortBy(lambda x: x[1], ascendingFalse)
print(rdd_sortby.collect())查看历史服务发现配置生效
