在火花流中随机获取 LeaseExpiredException答案

【问题标题】：Getting LeaseExpiredException in spark streaming randomly在火花流中随机获取 LeaseExpiredException
【发布时间】：2018-12-10 19:01:08
【问题描述】：

我有一个火花流（2.1.1 和 cloudera 5.12）。带有输入 kafka 和输出 HDFS（拼花格式）问题是，我随机收到 LeaseExpiredException（不是所有小批量）

org.apache.hadoop.ipc.RemoteException（org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException）：/user/qoe_fixe/data_tv/tmp/cleanData/_temporary/0/_temporary/attempt_20180629132202_0215_m_000000_0/上没有租约年=2018/月=6/天=29/小时=11/来源=LYO2/part-00000-c6f21a40-4088-4d97-ae0c-24fa463550ab.snappy.parquet（inode 135532024）：文件不存在。持有者 DFSClient_attempt_20180629132202_0215_m_000000_0_-1048963677_900 没有任何打开的文件。

我正在使用数据集 API 写入 hdfs

      if (!InputWithDatePartition.rdd.isEmpty() ) InputWithDatePartition.repartition(1).write.partitionBy("year", "month", "day","hour","source").mode("append").parquet(cleanPath)

由于这个错误，我的工作在几个小时后失败了

【问题讨论】：

您确定没有其他作业正在尝试更新/删除路径"cleanPath"？
我有两个工作流写入这个文件夹，但我添加了“源”作为分区，所以他们会写入不同的分区（文件夹）。当我将干净路径（父文件夹）更改为这两个工作的不同时，我没有遇到这个问题
目录路径是临时位置。能否给个具体的路径，看看是否存在这个问题？

标签： apache-spark hadoop hdfs spark-streaming parquet

【解决方案1】：

写入同一目录的两个作业共享同一 _temporary 文件夹。

因此，当第一个作业完成时，将执行此代码（FileOutputCommitter 类）：

  public void cleanupJob(JobContext context) throws IOException {
    if (hasOutputPath()) {
      Path pendingJobAttemptsPath = getPendingJobAttemptsPath();
      FileSystem fs = pendingJobAttemptsPath
          .getFileSystem(context.getConfiguration());
      // if job allow repeatable commit and pendingJobAttemptsPath could be
      // deleted by previous AM, we should tolerate FileNotFoundException in
      // this case.
      try {
        fs.delete(pendingJobAttemptsPath, true);
      } catch (FileNotFoundException e) {
        if (!isCommitJobRepeatable(context)) {
          throw e;
        }
      }
    } else {
      LOG.warn("Output Path is null in cleanupJob()");
    }
  }

它会在第二个作业仍在运行时删除 pendingJobAttemptsPath(_temporary) 这可能会有所帮助：

Multiple spark jobs appending parquet data to same base path with partitioning

【讨论】：