美文网首页
Spark异常处理——Shuffle FetchFailedEx

Spark异常处理——Shuffle FetchFailedEx

作者: 西二旗老司机 | 来源:发表于2018-07-23 12:19 被阅读0次

    在大规模数据处理中,这个错误比较常见。一般发生在有大量shuffle操作的时候,task不断的failed,然后又重执行,一直循环下去,直到application失败。

    报错方式

    • missing output location
    • shuffle fetch faild

    SparkSQL shuffle报错样例

    org.apache.spark.shuffle.MetadataFetchFailedException: 
    Missing an output location for shuffle 0
    
    org.apache.spark.shuffle.FetchFailedException:
    Failed to connect to hostname/192.168.xx.xxx:50268
    

    RDD shuffle报错样例

    WARN TaskSetManager: Lost task 17.1 in stage 4.1 (TID 1386, spark050013): java.io.FileNotFoundException: /data04/spark/tmp/blockmgr-817d372f-c359-4a00-96dd-8f6554aa19cd/2f/temp_shuffle_e22e013a-5392-4edb-9874-a196a1dad97c (没有那个文件或目录)
    
    FetchFailed(BlockManagerId(6083b277-119a-49e8-8a49-3539690a2a3f-S155, spark050013, 8533), shuffleId=1, mapId=143, reduceId=3, message=
    org.apache.spark.shuffle.FetchFailedException: Error in opening FileSegmentManagedBuffer{file=/data04/spark/tmp/blockmgr-817d372f-c359-4a00-96dd-8f6554aa19cd/0e/shuffle_1_143_0.data, offset=997061, length=112503}
    

    原因

    shuffle分为shuffle writeshuffle read两部分:

    • shuffle write:分区数由上一阶段的RDD分区数控制
      类似于saveAsLocalDiskFile的操作,将计算的中间结果按某种规则,临时存放到各个executor所在的本地磁盘上。
    • shuffle read:分区数由Spark提供的参数控制
      如果这个参数值设置的很小,同时shuffle read量很大,那么单个task处理的数据量也会很大,这可能导致JVM crash,从而获取shuffle数据失败,同时executor也丢失了,看到Failed to connect to host的错误,也就是executor lost的意思。
      有时候即使不会导致JVM crash也会造成长时间的gc。

    解决办法

    解决办法主要从 shuffle的数据量处理shuffle数据的分区数 两个角度入手。

    1. 减少shuffle数据
    • 是否可以使用map side join或是broadcast join来规避shuffle。
    • 在shuffle前将不必要的数据过滤掉。比如原始数据有20个字段,只要选取需要的字段进行处理即可,将会减少一定的shuffle数据。
    2. SparkSQL和DataFrame的join,group by等操作

    通过spark.sql.shuffle.partitions控制分区数,默认为200,根据shuffle的量以及计算的复杂度提高这个值。

    3. Rdd的join,groupBy,reduceByKey等操作

    通过spark.default.parallelism控制shuffle readreduce处理的分区数,默认为运行任务的core的总数(mesos细粒度模式为8个,local模式为本地的core总数),官方建议设置成运行任务的core的2-3倍。

    4. 提高executor的内存

    通过spark.executor.memory适当提高executor的内存
    通过spark.executor.cores增加每个executor的cpu,这样不会减少task并行度

    5. 是否存在数据倾斜的问题
    • 空值是否已经过滤?
    • 异常数据(某个key数据特别大)是否可以单独处理?
    • 考虑改变数据分区规则

    相关文章

      网友评论

          本文标题:Spark异常处理——Shuffle FetchFailedEx

          本文链接:https://www.haomeiwen.com/subject/hjwdmftx.html