Spark TaskAttempt目录未删除问题排查

Posted 2022-12-14 疯狂哈丘

文章目录

• • 一、问题描述
  • 二、问题分析
  • 三、总结

一、问题描述

有业务反馈spark任务结束后会遗留一些attempt目录在输出目录上，影响数据的读取。主要现象如下：

二、问题分析

之前排查过一个类似的问题，也是输出目录下有个遗留的_temporary目录未删除干净：

Spark 任务输出目录_temporary目录未删除问题排查

一开始以为就是这个问题，但是仔细分析了下，发现逻辑走不通。因此仔细做了下排查。

从目录的名称结构来看，我们很容易的可以看出这些都是spark task输出的临时目录，比如attempt_20190801042010_1478_r_000970_1则表示对应的RddId为1478，处理partitionId为970的task的第2次运行（开启推测执行后，spark会自动启动更多的task运行，这个数值从0开始累加，第一次运行就是0）。另外需要注意的是，这里的RddId并不等于stageId。

具体spark job是怎么输出临时目录并最终合并到输出目录的可以参考下这个文档：

Spark任务输出文件过程详解

既然知道是task输出的临时目录的问题，就可以找一个task去跟踪他的执行了。我们找了attempt_20190801042010_1478_r_000970_1这个task根据他的执行情况。最后跟到了partitonId=970这个task的两次运行记录所在的executor，看了下日志，发现以下两条关键日志：

从上面的日志可以看出，970这个task的两次attempt几乎在同时完成，时间都是04:20:37,364。也就是说他们执行commitTask的动作几乎是同时的。因此我们可以看一下FileOutputCommitter的commitTask方法：

public void commitTask(TaskAttemptContext context, Path taskAttemptPath)
throws IOException 
  TaskAttemptID attemptId = context.getTaskAttemptID();
  //判断该task是否需要输出
  if (hasOutputPath()) 
    context.progress();
    if(taskAttemptPath == null) 
      taskAttemptPath = getTaskAttemptPath(context);
    
    Path committedTaskPath = getCommittedTaskPath(context);
    FileSystem fs = taskAttemptPath.getFileSystem(context.getConfiguration());
    //判断该task的attempt目录是否存在
    if (fs.exists(taskAttemptPath)) 
      //判断之前是否有其他attempt的task已经提交过
      if(fs.exists(committedTaskPath)) 
        if(!fs.delete(committedTaskPath, true)) 
          throw new IOException("Could not delete " + committedTaskPath);
        
      
      //执行rename，将 attempt_20190801042010_1478_r_000970_1 重命名成 task_20190801042010_1478_r_000970
      if(!fs.rename(taskAttemptPath, committedTaskPath)) 
        throw new IOException("Could not rename " + taskAttemptPath + " to "
            + committedTaskPath);
      
        //输出日志，也就是我们刚才看到的那两条日志
      LOG.info("Saved output of task '" + attemptId + "' to " +
          committedTaskPath);
     else 
      LOG.warn("No Output found for " + attemptId);
    
   else 
    LOG.warn("Output Path is null in commitTask()");
  

这段代码的主要流程就是在task attempt执行完后，将attempt目录rename成commitedTask目录，这样就标志着这个task已经完成了。

但是我们可以看出，这个代码并没有做同步，假设同时有两个task attempt一起执行该方法，最差的情况就是两个线程都同时执行了 fs.rename(taskAttemptPath, committedTaskPath)。那么同时执行rename会有什么问题呢？

其实rename就是linux上的mv，逻辑基本是一样的。mv的逻辑是这样的，如果目标路径不存在，直接rename过去，修改一下文件的元数据就好了。如果目标路径已存在，则会将source文件移动目录路径下。

所以如果两个task attempt同时执行commitTask的rename，则会出现task970的输出目录task_20190801042010_1478_r_000970下有两个文件：

part-r-00970
attempt_20190801042010_1478_r_000970_1

最终spark在job结束后会统一将所有task输出目录下的所有文件都移动到输出目录下，因此才有了我们看到的现象。

三、总结

这个问题主要还是spark在处理推测执行的task时没处理好并发问题，目前可以想到的两种解决方案如下：

1、 在FileOutputCommitter的commitTask方法中加锁避免并发问题，这个需要修改hadoop的代码。但是如果不是对spark和hadoop的所有代码都很熟悉，不建议这么改，因为可能会引发其他的问题

2、 关闭推测执行，一劳永逸。目前发现了许多由于推测执行引发的问题（spark版本2.1.0），说明这一特性还不是很成熟，可以先关了（部分任务确实需要的话可以自行开启）