DataX源码解析-调度流程

tyrantlucifer

发布于 2022-08-30 14:25:02

1.1K0

发布于 2022-08-30 14:25:02

文章被收录于专栏：Tyrant LuciferTyrant Lucifer

前言

书接上回，继续来聊一聊DataX源码，在上篇文章中我们已经对DataX的整体架构以及运行流程有了一个比较细致的了解，这篇文章我们将更深层次的研究DataX在调度方面的细节。

调度流程解析

确认最终任务需要的channel数量

「注：channel是子任务数据传输的内存模型，后续文章将详细剖析，在这里可以暂且认为就是任务分片数量」

在任务周期中含有一个split()阶段，在这个阶段做了两件事情：

通过配置项计算出建议的并发channel数量
执行reader插件中的的实际切片逻辑，并根据数量切分configuration，请注意，这一步计算出的数量可能小于第一步配置的并发数

所以在真正调度阶段，需要根据split()阶段中计算的两个值，计算出最终的channel数量

image-20220427120240736

通过channel数量分配taskGroup

在计算出真正需要的channel数量之后，根据每个TaskGroup应该被分配任务的个数，计算TaskGroup的个数：

image-20220427144829662

由上图可知，任务的分配是由JobAssignUtil去进行，而且从方法名称assignFairly也可以知晓，分配的逻辑是公平分配，使用的Round Robin算法，轮询分配到每个TaskGroup中，在此就简单贴一下分配的核心源码：

    /**
     * <pre>
     * 需要实现的效果通过例子来说是：
     * a 库上有表：0, 1, 2
     * a 库上有表：3, 4
     * c 库上有表：5, 6, 7
     *
     * 如果有 4个 taskGroup
     * 则 assign 后的结果为：
     * taskGroup-0: 0,  4,
     * taskGroup-1: 3,  6,
     * taskGroup-2: 5,  2,
     * taskGroup-3: 1,  7
     * </pre>
     * @param resourceMarkAndTaskIdMap 对所有任务进行编号的map
     * @param jobConfiguration job的配置
     * @param taskGroupNumber 一共需要的taskGroup的数量
     * @return taskGroup的配置
     */
    private static List<Configuration> doAssign(LinkedHashMap<String, List<Integer>> resourceMarkAndTaskIdMap, Configuration jobConfiguration, int taskGroupNumber) {

        // 获取job配置
        List<Configuration> contentConfig = jobConfiguration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT);

        // 初始化一份taskGroup配置
        Configuration taskGroupTemplate = jobConfiguration.clone();

        // 将配置中的content配置项去掉
        taskGroupTemplate.remove(CoreConstant.DATAX_JOB_CONTENT);

        // 初始化taskGroup配置
        List<Configuration> result = new LinkedList<Configuration>();
        List<List<Configuration>> taskGroupConfigList = new ArrayList<List<Configuration>>(taskGroupNumber);
        for (int i = 0; i < taskGroupNumber; i++) {
            taskGroupConfigList.add(new LinkedList<Configuration>());
        }

        int mapValueMaxLength = -1;

        // 获取任务编号，并给每个任务一个标记，每个任务的标记都一致，用一个map保存，标记 -> 编号
        List<String> resourceMarks = new ArrayList<String>();
        for (Map.Entry<String, List<Integer>> entry : resourceMarkAndTaskIdMap.entrySet()) {
            resourceMarks.add(entry.getKey());
            if (entry.getValue().size() > mapValueMaxLength) {
                mapValueMaxLength = entry.getValue().size();
            }
        }

        // 核心Round Robin分配逻辑
        int taskGroupIndex = 0;
        for (int i = 0; i < mapValueMaxLength; i++) {
            for (String resourceMark : resourceMarks) {
                if (resourceMarkAndTaskIdMap.get(resourceMark).size() > 0) {
                    int taskId = resourceMarkAndTaskIdMap.get(resourceMark).get(0);
                    taskGroupConfigList.get(taskGroupIndex % taskGroupNumber).add(contentConfig.get(taskId));
                    taskGroupIndex++;

                    resourceMarkAndTaskIdMap.get(resourceMark).remove(0);
                }
            }
        }

        Configuration tempTaskGroupConfig;
        for (int i = 0; i < taskGroupNumber; i++) {
            tempTaskGroupConfig = taskGroupTemplate.clone();
            tempTaskGroupConfig.set(CoreConstant.DATAX_JOB_CONTENT, taskGroupConfigList.get(i));
            tempTaskGroupConfig.set(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_ID, i);

            result.add(tempTaskGroupConfig);
        }

        return result;
    }

启动调度

ok，已经确定完了taskGroup的个数以及每个taskGroup的channel数，接下来就到了真正启动任务的环节：

        // 初始化运行模式
        ExecuteMode executeMode = null;

        // 初始化调度器
        AbstractScheduler scheduler;
        try {
            // 在这里可以看到，DataX官方写死了运行模式，只支持STANDALONE模式，不支持分布式
         executeMode = ExecuteMode.STANDALONE;

            // 初始化调度器
            scheduler = initStandaloneScheduler(this.configuration);

            // 设置运行模式
            for (Configuration taskGroupConfig : taskGroupConfigs) {
                taskGroupConfig.set(CoreConstant.DATAX_CORE_CONTAINER_JOB_MODE, executeMode.getValue());
            }

            // 看到这里我彻底笑了 ^_^ ^_^ ^_^
            if (executeMode == ExecuteMode.LOCAL || executeMode == ExecuteMode.DISTRIBUTE) {
                if (this.jobId <= 0) {
                    throw DataXException.asDataXException(FrameworkErrorCode.RUNTIME_ERROR,
                            "在[ local | distribute ]模式下必须设置jobId，并且其值 > 0 .");
                }
            }

            LOG.info("Running by {} Mode.", executeMode);

            // 记录任务开始时间
            this.startTransferTimeStamp = System.currentTimeMillis();

            // 开始调度
            scheduler.schedule(taskGroupConfigs);

            // 记录任务结束时间
            this.endTransferTimeStamp = System.currentTimeMillis();
        } catch (Exception e) {
            LOG.error("运行scheduler 模式[{}]出错.", executeMode);
            this.endTransferTimeStamp = System.currentTimeMillis();
            throw DataXException.asDataXException(
                    FrameworkErrorCode.RUNTIME_ERROR, e);
        }

        // 检查任务执行情况
        this.checkLimit();

整个任务从scheduler.schedule(taskGroupConfigs)处被启动，在这个方法中，又调用了startAllTaskGroup(configurations)：

image-20220427180441170

    @Override
    public void startAllTaskGroup(List<Configuration> configurations) {
        // 生成线程池，对于线程池的个数是由taskGroup的数量进行控制
        this.taskGroupContainerExecutorService = Executors
                .newFixedThreadPool(configurations.size());

        // 注册taskGroup任务到线程池中并启动
        for (Configuration taskGroupConfiguration : configurations) {
            TaskGroupContainerRunner taskGroupContainerRunner = newTaskGroupContainerRunner(taskGroupConfiguration);
            this.taskGroupContainerExecutorService.execute(taskGroupContainerRunner);
        }
        // 关闭线程池提交线程的入口
        this.taskGroupContainerExecutorService.shutdown();
    }

实际上DataX底层对于每个taskGroup都启动了一个线程TaskGroupContainerRunner，采用线程池的方式实现并发操作

调度子单位解析

TaskGroupContainerRunner

TaskGroupContainerRunner是线程子单位，作为最上层的封装，直接提交到线程池中执行：

image-20220427182500027

public class TaskGroupContainerRunner implements Runnable {

 private TaskGroupContainer taskGroupContainer;

 private State state;

 public TaskGroupContainerRunner(TaskGroupContainer taskGroup) {
  this.taskGroupContainer = taskGroup;
  this.state = State.SUCCEEDED;
 }

 @Override
 public void run() {
  try {
            Thread.currentThread().setName(
                    String.format("taskGroup-%d", this.taskGroupContainer.getTaskGroupId()));
            // 启动真正的并发子单位
            this.taskGroupContainer.start();
   this.state = State.SUCCEEDED;
  } catch (Throwable e) {
   this.state = State.FAILED;
   throw DataXException.asDataXException(
     FrameworkErrorCode.RUNTIME_ERROR, e);
  }
 }

 public TaskGroupContainer getTaskGroupContainer() {
  return taskGroupContainer;
 }

 public State getState() {
  return state;
 }

 public void setState(State state) {
  this.state = state;
 }
}

从代码上看，真正核心的并发子单位是TaskGroupContainer