Spark TaskAttempt目录未删除问题排查

待我称王封你为后i 2021-11-10 19:58 278阅读 0赞

### 文章目录 ###

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

## 一、问题描述 ##

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMzMzIxMjQ_size_16_color_FFFFFF_t_70]

## 二、问题分析 ##

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

[Spark 任务输出目录\_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任务输出文件过程详解][Spark]

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

![在这里插入图片描述][20190803174654466.png]

![在这里插入图片描述][20190803174703371.png]

从上面的日志可以看出，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），说明这一特性还不是很成熟，可以先关了（部分任务确实需要的话可以自行开启）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMzMzIxMjQ_size_16_color_FFFFFF_t_70]: /images/20211109/6a65415fd3ef4958b8f508053d6c8cbf.png
[Spark _temporary]: https://blog.csdn.net/u013332124/article/details/90677530
[Spark]: https://blog.csdn.net/u013332124/article/details/92001346
[20190803174654466.png]: /images/20211109/2da2cbb12f964c568bb9458cd5a89c1b.png
[20190803174703371.png]: /images/20211109/9f38dfa51e83489fb0c53293930a1850.png