之前也遇到过几次关于OOM(堆外内存溢出)的问题，但都只是大体上看了看，没有细致的总结，目前了解的还不是特别清楚，只好总结一下我觉得可行的处理方案，另外贴一些原理。
首先是当时的一些处理方案：
第一次OOM：第一次遇到这个问题时，上网查，发现很多人都说要更改运行内存之类的，但本人是个小白，而且这个job是提交到集群上的，也不敢乱改，一般来说都是够的。我的数据量大约是在千万，接近亿级的数据，其中主要的操作是以userId为key的reduceByKey，询问了一下同事，他提出有可能是数据不平衡导致的节点资源分布不均匀（id开头为1,2的明显比较多），有些节点负载过大，而其他节点负载很小，导致分配内存时浪费了许多。
解决方案：将userId做了反转，基本可以保证数据均衡，因为末尾数字出现频率大约一致。
第二次OOM：运行了一段时间后再次OOM，这次原因就比较显而易见了，我在存文件之前将所有数据repartion(1)了，改大就好了。
解决方案：增大partition数量。当我改到16的时候，就不再出现OOM的问题了。
第三次OOM：这次有点手足无措了，一开始以为是保存的时候出的问题，后来仔细查看了spark job每一个步骤所使用的资源，shuffle read和right的数据量，以及失败次数，才发现是我在收集完5~6个数据源的数据后，做了这样一个操作
Await
.result(futures, 5.hours)
.reduce(_ union _)解决方案： 我把这段改成了以下这样
val resList = Await
.result(futures, 5.hours)
var res = resList.head
resList.foreach(data => {
if (data != resList.head)
res = res
.union(data)
.reduceByKey(reduceUserData(_: VipUserData, _: VipUserData))
})
--------------------------------我是讲解的分割线--------------------------------
OOM可能的原因有以下几点：
off heap: 资源释放不当, 例如加载文件资源次数过多, 且不正常关闭, 例如多次调用ClassLoader().getResourceAsStream
DirectMemory: 拉取Executor端Task Result数据回Driver节点时, 此处消耗的DirectMemory内存 = conf.getInt("spark.resultGetter.threads", 4) * TaskResultSize
Executor可能消耗的情况如下:
因为会涉及到拉取remote RDD Block时出现Direct OOM, 此时消耗的Direct Memory = 拉取的RDDBlockSize.
Tips: 查看RDD Block Size步骤: SparkUI->Storage Tabs -> 看众多RDD中Memory或者Disk中的totalSize/cached Partitions中最大的RDD, 点进去看详情页, 然后对RDD的大小 按照Memory或者Disk排序, 找到最大的RDD Block....
如下图标记处: 
此时消耗的Direct Memory 通常= max(某个Shuffle Block的size, 50MB)
Tips: 可以在抛出该Direct OOM的Executor节点上检查是否有如下日志: Spark会在如果单个shuffleBlock的大小>1MB时输出该语句.
还有一种预估的方式, 前一阶段Stage 对其内的每个Task的Shuffle Write排序, 找到最大的Shuffle Write / 下一stage的task个数, 即为一个预估的shuffle Block大小.
一般情况最大的值是整个作业中最大的disk level的RDD Block的size. 但除了些许特殊操作: zip类操作(包括相同partitoner的RDD做union操作, 因为会被Spark后台优化成zip操作) 使用的size等于该操作zip的rdd中涉及的所有disk level rdd block size之和.
此部分Size如第(1)点所示, 找到Disk中最大的RDD Block即可.
堆外内存的使用总量 = jvmOverhead(off heap) + directMemoryOverhead(direct memory) + otherMemoryOverhead
| 参数 | 描述 | 默认值 |
|---|---|---|
| spark.yarn.executor.jvmMemoryOverhead | off heap内存控制 | max(0.1 * executorMemory, 384MB) |
| spark.yarn.executor.directMemoryOverhead | Direct Memory的控制参数 | 256MB |
| spark.yarn.driver.jvmMemoryOverhead | 同Executor | |
| spark.yarn.driver.directMemoryOverhead | 同Executor | |
| spark.yarn.executor.memoryOverhead | 统筹参数, 如果设置了该值m, 会自动按比例分配off heap给jvmOverhead和directMemory, 分配比例为jvmOverhead = max(0.1 * executorMemory, 384MB), directMemoryOverhead =m - jvmOverhead | 无 |
| spark.yarn.driver.memoryOverhead | 同Executor |
合理的参数推荐:
一般推荐总值:
spark.yarn.executor.directMemoryOverhead =
{ if 存在memory level or disk level 的 block then 第1点的Size else 0 } +
{if Shuffle阶段抛出Direct OOM then 第2点的Size else 0} +
｛if 存在Disk level的Block then 第3点的192MB else 0｝ +
{ if 存在其他框架的 then 其他框架的size else 0} +
256MB
spark.yarn.executor.jvmOverhead =
{ if 存在disk level的Block then 第4点的Size else 0 } +
{ if 存在其他框架的 then 其他框架的size else 0} +
max(executor-memory * 0.1, 384)
// 如果没有Executor表现为堆外内存使用超出, 则不需要手动调整.