Hadoop3.2.1 【 YARN 】源码分析 : DeletionService解析

川长思鸟来 2022-10-28 03:52 120阅读 0赞

# 一 .前言 #

NodeManager将文件删除功能服务化,即提供一个专门的文件删除服务异步删除失效文件， 这样可避免同步删除文件带来的性能开销

# 二 .属性 #

// 延迟
      private int debugDelay;
    
      // ContainerExecutor
      private final ContainerExecutor containerExecutor;
    
      //存储
      private final NMStateStoreService stateStore;
    
      // 线程池
      private ScheduledThreadPoolExecutor sched;
    
      // 原子类, 获取 任务的id
      private AtomicInteger nextTaskId = new AtomicInteger(0);

# 三 .构造方法 #

没啥,都是赋值操作…

public DeletionService(ContainerExecutor exec) { 
        this(exec, new NMNullStateStoreService());
      }
    
    
      public DeletionService(ContainerExecutor containerExecutor,  NMStateStoreService stateStore) { 
        super(DeletionService.class.getName());
        this.containerExecutor = containerExecutor;
        this.debugDelay = 0;
        this.stateStore = stateStore;
      }

# 四 .serviceInit #

主要构建线程池HadoopScheduledThreadPoolExecutor …

@Override
      protected void serviceInit(Configuration conf) throws Exception { 
        ThreadFactory tf = new ThreadFactoryBuilder()
            .setNameFormat("DeletionService #%d")
            .build();
    
    
    
        if (conf != null) { 
    
          // 从配置中获取参数
          // yarn.nodemanager.delete.thread-count : 4
          sched = new HadoopScheduledThreadPoolExecutor(
              conf.getInt(YarnConfiguration.NM_DELETE_THREAD_COUNT,
                  YarnConfiguration.DEFAULT_NM_DELETE_THREAD_COUNT), tf);
    
          // yarn.nodemanager.delete.debug-delay-sec : 0
          debugDelay = conf.getInt(YarnConfiguration.DEBUG_NM_DELETE_DELAY_SEC, 0);
    
        } else { 
    
          sched = new HadoopScheduledThreadPoolExecutor(
              YarnConfiguration.DEFAULT_NM_DELETE_THREAD_COUNT, tf);
        }
    
        // true时，在执行shutdown方法后，当前正在等待的任务的和正在运行的任务需要被执行完，然后进程被销毁；
        // false时，表示放弃等待的任务，正在运行的任务一旦完成，则进程被销毁对与cheduleAtFisedRate方法和scheduleWithFixedDelay方法，则会打断循环，只执行当前正在执行的任务，不会对他们在进行循环执行
        sched.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
    
        //
        sched.setKeepAliveTime(60L, SECONDS);
        if (stateStore.canRecover()) { 
          recover(stateStore.loadDeletionServiceState());
        }
        super.serviceInit(conf);
      }

# 五 .delete方法 #

就是传入DeletionTask类型的任务, 执行删除操作…

public void delete(DeletionTask deletionTask) { 
        if (debugDelay != -1) { 
          if (LOG.isDebugEnabled()) { 
            String msg = String.format("Scheduling DeletionTask (delay %d) : %s",
                debugDelay, deletionTask.toString());
            LOG.debug(msg);
          }
          // 删除存储
          recordDeletionTaskInStateStore(deletionTask);
          // 执行删除任务..
          sched.schedule(deletionTask, debugDelay, TimeUnit.SECONDS);
        }
      }

# 六 .FileDeletionTask #

文件删除类型的任务, 父类是DeletionTask , DeletionTask 是一个抽象类, 用于删除资源文件任务…

直接看一下run方法就好了, 其实就是 删除文件的各种操作…

/** * Delete the specified file/directory as the specified user. */
      @Override
      public void run() { 
    
    
        if (LOG.isDebugEnabled()) { 
          // 打印log日志....
          String msg = String.format("Running DeletionTask : %s", toString());
          LOG.debug(msg);
        }
        boolean error = false;
        if (null == getUser()) { 
          if (baseDirs == null || baseDirs.size() == 0) { 
            if (LOG.isDebugEnabled()) { 
              LOG.debug("NM deleting absolute path : " + subDir);
            }
            try { 
              // 递归删除文件
              lfs.delete(subDir, true);
            } catch (IOException e) { 
              error = true;
              LOG.warn("Failed to delete " + subDir);
            }
          } else { 
            for (Path baseDir : baseDirs) { 
              Path del = subDir == null? baseDir : new Path(baseDir, subDir);
              if (LOG.isDebugEnabled()) { 
                LOG.debug("NM deleting path : " + del);
              }
              try { 
                // 递归删除文件
                lfs.delete(del, true);
              } catch (IOException e) { 
                error = true;
                LOG.warn("Failed to delete " + subDir);
              }
            }
          }
        } else { 
          try { 
            if (LOG.isDebugEnabled()) { 
              LOG.debug(
                  "Deleting path: [" + subDir + "] as user: [" + getUser() + "]");
            }
            if (baseDirs == null || baseDirs.size() == 0) { 
              getDeletionService().getContainerExecutor().deleteAsUser(
                  new DeletionAsUserContext.Builder()
                  .setUser(getUser())
                  .setSubDir(subDir)
                  .build());
            } else { 
              getDeletionService().getContainerExecutor().deleteAsUser(
                  new DeletionAsUserContext.Builder()
                  .setUser(getUser())
                  .setSubDir(subDir)
                  .setBasedirs(baseDirs.toArray(new Path[0]))
                  .build());
            }
          } catch (IOException|InterruptedException e) { 
            error = true;
            LOG.warn("Failed to delete as user " + getUser(), e);
          }
        }
        if (error) { 
          setSuccess(!error);
        }
        deletionTaskFinished();
      }