Spark ExecutorLostFailure内存超出

Question

我一直试图获得一个火花工作,现在运行完成几天,我终于能够完成它但仍然有大量失败的任务,其中执行者被杀死的信息如下:

ExecutorLostFailure(执行程序77退出由其中一个正在运行的任务引起)原因:容器被YARN杀死超过内存限制.使用45.1 GB的44.9 GB物理内存.考虑提升spark.yarn.executor.memoryOverhead

这些是我传递给集群的属性:

[
    {
        "classification": "spark-defaults",
        "properties": {
            "spark.executor.memory": "41000m",
            "spark.driver.memory": "8000m",
            "spark.executor.cores": "6",
            "spark.shuffle.service.enabled": "true",
            "spark.executor.instances": "98",
            "spark.yarn.executor.memoryOverhead": "5000"
        }
    }
]

该集群包括20台机器,每台机器有32个内核和240G内存.我是否应该继续提高memoryOverhead,或者是否表明存在更深层次的问题.在将结果数据写入S3之前,此时的错误似乎发生在从5000个分区合并到500个分区期间.我猜测coalesce导致了一个shuffle,因为集群的内存已经很低,所以它推得太远了.

工作流程如下:

1. 将s3中的镶木地板文件加载到数据框中
2. 提取一组唯一键,使用sql查询对数据进行分组
3. 将数据帧转换为JavaRDD并应用多个映射函数
4. MapToPair的数据

5. 使用下面的combineByKey,实质上是通过键将单个对象合并到对象数组中

   combineByKey(新函数,添加函数,合并函数,新HashPartitioner(5000),false,null);

6. 更多地图

7. 对于几个唯一键中的每一个,过滤rdd以获得具有该键的元组,然后在合并后将每个子集保留到磁盘

另一个问题是如何导出上面的44.9数字.我认为最大内存将是执行程序内存+ memoryOverhead,这将是46G而不是44.9G.

Nathan,非常感谢任何帮助

Answer 1

根据我的经验,这表明更深层次的问题,从你发布的内容我看到了一些陷阱.

首先,您可能需要查看分区大小,因为combineByKey操作期间创建的数据偏差很容易导致OOM .也许有些钥匙非常频繁？

如果没有,我会查看coalesce函数调用.你还没有发布代码,所以我只能猜测正在生成的DAG,但我会知道coalesce函数和在同一个写阶段执行的其他操作.

Spark分阶段执行,从我从你的解释中可以看出,你coalesce之前调用write,所以取决于你进入最后阶段的分区数量,并且根据在这个阶段完成的转换,你实际上可能在更少的分区上运行因此导致OOM异常.

用文字解释有点复杂,但我会尝试给出一个可能发生的事情的简单例子.

想象一下这样一个简单的场景,你在一个键值对的文件中读取(Int, Double),然后你将一些函数应用于所有的值,比如说round.然后,您希望将输出写回单个文件,因此您调用coalesce(1)后跟write.代码看起来像这样:

val df = sqlContext.read.parquet("/path/to/my/file/")
df.map{case(key: Int, value: Double) => (key, round(value)}
  .toDF()
  .coalesce(1)
  .write
  .parquet("/my/output/path/")

现在有人可能认为map操作是在整个集群上并行执行的,但是如果你注意spark ui,你会注意到这个任务没有在你的集群中分布.因为coalesce(1),Spark知道一切都需要在一个分区中结束,所以它只是开始将所有数据收集到应用该map功能的一个分区中.您可以想象,这可能很容易在OOM异常中以更复杂的转换结束.

我希望这能为您提供一些关于在哪里寻找的指示.祝好运 :)