其他分享
首页 > 其他分享> > spark调优篇-oom 优化(汇总)

spark调优篇-oom 优化(汇总)

作者:互联网

spark 之所以需要调优,一是代码执行效率低,二是经常 OOM 

 

内存溢出

内存溢出无非两点

1. Driver 内存不够

2. Executor 内存不够

Driver 内存不够无非两点

1. 读取数据太大

2. 数据回传

Executor 内存不够无非两点

1. map 类操作产生大量数据,包括 map、flatMap、filter、mapPartitions 等

2. shuffle 后产生数据倾斜

 

Executor 内存不够

有个通用的解决办法就是增加 Executor 内存

--executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).

但这并不一定是最好的办法

 

map 过程产生大量对象

造成 Executor 内存溢出

解决思路是减少每个 task 的大小,从而减少每个 task 的输出;

具体做法是在 会产生大量对象的 map 操作前 添加 repartition(重新分区) 方法,分区成更小的块传入 map

rdd.flatMap(lambda x: ['%d'%x*50 for _ in range(100000000)]).count()      # 100 * 100000000 个对象,内存溢出
rdd.flatMap(lambda x: len(['%d'%x*50 for _ in range(100000000)])).sum()     # 内存溢出

rdd.repartition(1000000).flatMap(lambda x: ['%d'%x*50 for _ in range(100000000)]).count()   # 可执行

 

数据倾斜

参考我的博客 数据倾斜

 

standalone 模式资源分配不均

该模式下配置了 

--total-executor-cores NUM  (Total cores for all executors.)   集群 executor 核数

--executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).  每个 executor 内存

而没有配置

--executor-cores NUM Number of cores per executor. (Default: 1 in YARN mode,
or all available cores on the worker in standalone mode)  每个 executor 核数

 

假如各个 executor 核数不一样,核数多的 executor 执行的 task 就多,内存就容易溢出

解决方法是配置参数 --executor-cores,或者是在 spark 中配置 spark.executor.cores

 

在 RDD 中共用对象

rdd = sc.parallelize(range(100))
def myfunc(x): return x
rdd.flatMap(lambda x: [('k', 'v') for _ in range(200000000)]).foreach(myfunc)     # 每次生成一个 tuple 对象,内存溢出
rdd.flatMap(lambda x: ['k'+'v' for _ in range(2000000)]).count()        # 无需生成新的 string 对象,可执行

tuple 为不可变对象,不过字符串也是可变对象

此条方法有待进一步验证

 

Driver 中需要读取大量数据

造成 Driver 内存溢出

解决思路是增加 Driver 内存,具体做法为设置参数

--driver-memory MEM         Memory for driver (e.g. 1000M, 2G) (Default: 1024M). 

示例

from pyspark import SparkContext
sc = SparkContext(master='yarn')
rdd = sc.parallelize(range(300000000))
# spark-submit --master yarn-client  --driver-memory 512M  driver_oom.py    内存溢出
# spark-submit --master yarn-client  --driver-memory 3G  driver_oom.py  可以执行

 

collect

大量数据回传 Driver,造成内存溢出

解决思路是分区输出,具体做法是 foreach

rdd = sc.parallelize(range(100))
rdd.flatMap(lambda x: ['%d'%x*50 for _ in range(100000)]).collect()     # 内存溢出

def func(x): print(x)
rdd.flatMap(lambda x: ['%d'%x*50 for _ in range(100000)]).foreach(func) # 分区输出

或者增加 Driver 内存

 

代码优化

mapPartitions

1. 批处理

2. 减少中间输出

用 mapPartitions 替代多个 map,减少 Executor 内存压力

from pyspark import SparkContext
sc = SparkContext(master='yarn')
data = range(10)
rdd = sc.parallelize(data, 2)

##### map
rdd.map(lambda x: x % 3).filter(lambda x : x>1 ).countByValue().values()        # [3]

##### mapPartitions
# 避免了中间 RDD 的产生,节约内存,防止 oom
def myfunc(datas):
    # datas type is itertools.chain
    for data in datas:
        value = data % 3
        if value > 1:
            yield value

print rdd.mapPartitions(myfunc).countByValue().values()     # [3]
# spark-submit --master yarn-client mapVSmapPartitions.py   python 只支持 client 模式

 

DataFrame 代替 RDD

任务被划分成多个 stage,在每个 stage 内部,RDD 是无法自动优化的,

rdd.map(lambda x: x+1).map(lambda x: x+1)   ==  rdd.map(lambda x: x+2)

如上面两个操作是等价的,但是 RDD 并不会自动优化,

而 DataFrame 使用 sql 查询,自带 sql 优化器,可自动找到最优方案

 

broadcast join

在分布式计算中,数据跨节点移动是非常影响性能的,网络传输耗时,多次传输消耗内存,broadcast 在某些场景可以减少数据移动;

如 一个 小RDD 要和 一个 大RDD 进行 join 操作,常规情况下要互传 RDD,由于多个 task,故需多次传输,    【注意必须是有个小 RDD,否则这种做法意义不大,因为后面要遍历这个广播变量】

如果把 小RDD 变成 broadcast 变量,就不用传输 大RDD,把 broadcast(小RDD) 缓存在对应 Executor 上即可

对 大RDD 进行 map 操作,在 map 函数中调用 小RDD 的 value,遍历 小RDD

map(lambda x: i for i in smallRDD.value if x == i)

 

filter 之后再 join

就是所谓的谓词下推,在 sparkSQL 中会自动这么操作;

如果是自己操作 RDD,可以减少 shuffle 的数据量 

 

cache and persist

缓存 RDD 既可以节省内存,也可以提高性能;

cahce 是缓存到内存,等同于 persist(Storage.MEMORY_ONLY),在内存不足时,这种缓存方式会丢失数据,再次使用时会重新计算;

rdd.persist(StorageLevel.MEMORY_AND_DISK_SER) 在内存不足时会写到磁盘,避免重复,只是耗费一点 IO 时间

 

combineByKey

在 hadoop 中也有 combine,有 combine 比 没有combine 效率高;

比如 reduceByKey (combine操作) 就比 groupyByKey (非combine操作) 效率高

import time
from pyspark import SparkContext

sc = SparkContext(master='yarn')

strs = list('abcd')*10000000
rdd = sc.parallelize(strs)

time.clock()
print rdd.map(lambda x: (x, 1)).reduceByKey(lambda x, y: x+y).collect()     # combinByKey 操作耗时少3.2s
# print rdd.map(lambda x: (x, 1)).groupByKey().mapValues(sum).collect()        # 非 combinByKey 操作耗时3.6s
# 二者结果一样
print(time.clock())

strs = list('abcd')*10000000
for i in strs:i = (i, 1)    # 6s,单机for循环做更少的事情,耗时更多

图解如下

 

 

 

 

参考资料:

https://blog.csdn.net/yhb315279058/article/details/51035631

标签:map,RDD,oom,executor,rdd,内存,调优篇,spark,lambda
来源: https://www.cnblogs.com/yanshw/p/11988347.html