网站开发计算机配置,成都定制网站设,杭州工业设计公司排名前十强,wordpress 加载中动画RDD Cache
当同一个 RDD 被引用多次时#xff0c;就可以考虑进行 Cache#xff0c;从而提升作业的执行效率
// 用 cache 对 wordCounts 加缓存
wordCounts.cache
// cache 后要用 action 才能触发 RDD 内存物化
wordCounts.count// 自定义 Cache 的存储介质、存储形式、副本…RDD Cache
当同一个 RDD 被引用多次时就可以考虑进行 Cache从而提升作业的执行效率
// 用 cache 对 wordCounts 加缓存
wordCounts.cache
// cache 后要用 action 才能触发 RDD 内存物化
wordCounts.count// 自定义 Cache 的存储介质、存储形式、副本数量
wordCounts.persist(MEMORY_ONLY)Spark 的 Cache 机制 :
缓存的存储级别限定了数据缓存的存储介质如 : 内存、磁盘缓存的计算过程从 RDD 展开到分片 (Block)存储于内存或磁盘的过程缓存的销毁过程缓存数据主动或被动删除的内存或磁盘的过程
存储级别
Spark 支持的存储级别
RDD 缓存的默认存储级别MEMORY_ONLYDataFrame 缓存的默认存储级别MEMORY_AND_DISK
存储级别存储介质存储形式副本设置内存磁盘对象值序列化MEMORY_ONLY√√1MEMORY_ONLY_2√√2MEMORY_ONLY_SER√√1MEMORY_ONLY_SER_2√√2DISK_ONLY√√1DISK_ONLY_2√√2DISK_ONLY_3√√3MEMORY_AND_DISK√√√√1内存以对象值存储磁盘以序列化MEMORY_AND_DISK_2√√√√2MEMORY_AND_DISK_SER√√√1内存/磁盘都以序列化的字节数组存储MEMORY_AND_DISK_SER2√√√2
计算过程
缓存的计算过程 :
MEMORY_AND_DISK 先把数据集全部缓存到内存内存不足时才把剩余的数据落磁盘MEMORY_ONLY 只把数据往内存里塞
内存中的存储过程 销毁过程
缓存的销毁过程 :
缓存抢占 Execution Memory 空间会进行缓存释放
Spark 清除缓存的原则
LRU按元素的访问顺序优先清除那些最近最少访问的 BlockId、MemoryEntry 键值对在清除时同属一个 RDD 的 MemoryEntry 不会清除
Spark 实现 LRU 的数据结构LinkedHashMap , 内部有两个数据结构
HashMap : 用于快速访问根据指定的 BlockIdO(1) 返回 MemoryEntry双向链表 : 用于维护元素BlockId 和 MemoryEntry 键值对的访问顺序
Spark 会释放 LRU 的 MemoryEntry : Cache 注意点
用 Cache 的基本原则 :
当 RDD/DataFrame/Dataset 的引用数为 1坚决不用 Cache当引用数大于 1且运行成本超过 30%就考虑用 Cache
运行成本占比 : 计算某个分布式数据集要消耗的总时间与作业执行时间的比值
端到端的执行时间为 1 小时DataFrame 被引用了 2 次从读取数据源到生成该 DataFrame 花了 12 分钟该 DataFrame 的运行成本占比 12*2/60 40%
用 noop 计算 DataFrame 运行时间 :
df.write.format(noop).save().cache 是惰性操作在调用 .cache后要先用 count 才能触发缓存的完全物化
Cache 要遵循最小公共子集原则 :
val filePath: String _
val df: DataFrame spark.read.parquet(filePath)//Cache方式一
val cachedDF df.cache
//数据分析
cachedDF.filter(col2 0).select(col1, col2)
cachedDF.select(col1, col2).filter(col2 100)两个查询的 Analyzed Logical Plan 不一致无法缓存复用
//Cache方式二
df.select(col1, col2).filter(col2 0).cache
//数据分析
df.filter(col2 0).select(col1, col2)
df.select(col1, col2).filter(col2 100)Analyzed Logical Plan 完全一致能缓存复用
//Cache方式三
val cachedDF df.select(col1, col2).cache
//数据分析
cachedDF.filter(col2 0).select(col1, col2)
cachedDF.select(col1, col2).filter(col2 100)及时清理 Cache :
异步模式 (常用)调用 unpersist() 或 unpersist(False)同步模式调用 unpersist(True)
OOM
OOM的具体区域 :
发生 OOM 的 LOCLine Of Code代码位置OOM 发生在 Driver 端还是在 Executor 端在 Executor 端的哪片内存区域
Driver OOM
Driver 端的 OOM 位置 :
创建小规模的分布式数据集使用 parallelize、createDataFrame 创建数据集收集计算结果通过 take、show、collect 把结果收集到 Driver 端
Driver 端的 OOM 原因
创建的数据集超过内存上限收集的结果集超过内存上限
广播变量的创建与分发 : 广播变量的数据拉取就是用 collect 。当数据总大小超过 Driver 端内存时 , 就报 OOM :
java.lang.OutOfMemoryError: Not enough memory to build and broadcast对结果集尺寸预估适当增加 Driver 内存配置
Driver 内存大小 : spark.driver.memory
查看执行计划 :
val df: DataFrame _
df.cache.countval plan df.queryExecution.logical
val estimated: BigInt spark.sessionState.executePlan(plan).optimizedPlan.stats.sizeInBytesExecutor OOM
当 Executors OOM 时定位 Execution Memory 和 User Memory
User Memory OOM
User Memory 用于存储用户自定义的数据结构如: 数组、列表、字典
当自定义数据结构的总大小超出 User Memory 上限时就会报错
java.lang.OutOfMemoryError: Java heap space at java.util.Arrays.copyOf
java.lang.OutOfMemoryError: Java heap space at java.lang.reflect.Array.newInstance计算 User Memory 消耗时要考虑 Executor 的线程池大小
当 dict 大小为 #size , Executor 线程池大小为 #threadsdict 对 User Memory 的总消耗#size * #threads当总消耗超出 User Memory 上限就会 OOM
val dict List(spark, tune)
val words spark.sparkContext.textFile(~/words.csv)val keywords words.filter(word dict.contains(word))
keywords.map((_, 1)).reduceByKey(_ _).collect自定义数据分发 :
自定义的列表 dict 会随着 Task 分发到所有 Executors多个 Task 的 dict 会对 User Memory 产生重复消耗 解决 User Memory OOM 的思路 :
先对数据结构的消耗进行预估相应地扩大 User Memory
UserMemory 总大小 spark.executor.memory * (1 - spark.memory.fraction)
Execution Memory OOM
Execution Memory OOM 常见实例数据倾斜和 数据膨胀
配置说明:
2 个 CPU core每个 core 有两个线程内存大小为 1GBspark.executor.cores 3spark.executor.memory 900MBExecution Memory 180MBStorage Memory 180MBExecution Memory 上限 360MB
数据倾斜
3 个 Reduce Task 对应的数据分片大小分别是 100MB , 100MB , 300MB
当 Executor 线程池大小为 3所以每个 Reduce Task 最多 360MB * 1/3 120MB
数据倾斜导致OOM :
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-NHRj4w40-1677761995168)(…/…/png/%E5%86%85%E5%AD%98%E8%BF%90%E7%94%A8/image-20230209210919876.png)]
2 种调优思路
消除数据倾斜让所有的数据分片尺寸都小于 100MB调整 Executor 线程池、内存、并行度等相关配置提高 1/N 上限到 300MB
在 CPU 利用率高下解决 OOM :
维持并发度、并行度不变增大执行内存设置提高 1/N 上限到 300MB维持并发度、执行内存不变提升并行度把数据打散将所有的数据分片尺寸都缩小到 100MB 内
数据膨胀
3 个 Map Task 对应的数据分片大小都是 100MB
数据膨胀导致OOM : 2 种调优思路
把数据打散提高数据分片数量、降低数据粒度让膨胀后的数据量降到 100MB 左右加大内存配置结合 Executor 线程池调整提高 1/N 上限到 300MB
