datax(6):启动步骤解析

全栈程序员站长

发布于 2022-08-29 15:48:31

5820

发布于 2022-08-29 15:48:31

文章被收录于专栏：全栈程序员必看

大家好，又见面了，我是你们的朋友全栈君。

通过前面 datax(2): 通过idea搭建源码阅读+调试环境已经知道了idea下阅读源码的步骤，现在看下 DataX启动步骤解析

一、启动java类(主入口)

/**
 * Engine是DataX入口类，该类负责初始化Job或者Task的运行容器，并运行插件的Job或者Task逻辑
 */
com.alibaba.datax.core.Engine

二、启动的步骤

1、解析用户输入的参数： job（datax的json），jobId（默认-1,），running_mode

2、配置额外的参数，打印vm的信息，打印过滤的配置信息（过滤敏感字符），校验配置

3、配置conf传入Engine.start()，启动程序

4、绑定字段信息，初始化插件加载器

5、判断任务类型（taskGroup还是job），生成不同的container（JobContainer或TaskGroupContainer）

6、打开各种追踪器，报告器，用于任务的运行状况收集

7、container.start() 对应的容器开始运行任务

Datax的执行过程

过程详细说明如下：

DataX完成单个数据同步的作业，我们称之为Job，DataX接受到一个Job之后，将启动一个进程来完成整个作业同步过程。DataX Job模块是单个作业的中枢管理节点，承担了数据清理、子任务切分(将单一作业计算转化为多个子Task)、TaskGroup管理等功能。
DataXJob启动后，会根据不同的源端切分策略，将Job切分成多个小的Task(子任务)，以便于并发执行。Task便是DataX作业的最小单元，每一个Task都会负责一部分数据的同步工作。
切分多个Task之后，DataX Job会调用Scheduler模块，根据配置的并发数据量，将拆分成的Task重新组合，组装成TaskGroup(任务组)。每一个TaskGroup负责以一定的并发运行完毕分配好的所有Task，默认单个任务组的并发数量为5。
每一个Task都由TaskGroup负责启动，Task启动后，会固定启动Reader—>Channel—>Writer的线程来完成任务同步工作。
DataX作业运行起来之后， Job监控并等待多个TaskGroup模块任务完成，等待所有TaskGroup任务完成后Job成功退出。否则，异常退出，进程退出值非0

简单总结过程如下：

一个DataX Job会切分成多个Task，每个Task会按TaskGroup进行分组，一个Task内部会有一组Reader->Channel->Writer。Channel是连接Reader和Writer的数据交换通道，所有的数据都会经由Channel进行传输

三、启动时序图

四、主要方法

五、代码详细

package com.alibaba.datax.core;

**
 * Engine是DataX入口类，该类负责初始化Job或者Task的运行容器，并运行插件的Job或者Task逻辑
 */
public class Engine { 
   

  private static final Logger LOG = LoggerFactory.getLogger(Engine.class);

  private static String RUNTIME_MODE;

  /**
   * 真正开始执行任务的地方 check job model (job/task) first
   *
   * @param allConf Configuration
   */
  public void start(Configuration allConf) { 
   

    // 绑定column转换信息
    ColumnCast.bind(allConf);

    //初始化PluginLoader，可以获取各种插件配置
    LoadUtil.bind(allConf);

    boolean isJob = !("taskGroup"
        .equalsIgnoreCase(allConf.getString(CoreConstant.DATAX_CORE_CONTAINER_MODEL)));
    //JobContainer会在schedule后再行进行设置和调整值
    int channelNumber = 0;
    AbstractContainer container;
    long instanceId;
    int taskGroupId = -1;
    if (isJob) { 
   
      allConf.set(CoreConstant.DATAX_CORE_CONTAINER_JOB_MODE, RUNTIME_MODE);
      container = new JobContainer(allConf);
      instanceId = allConf.getLong(CoreConstant.DATAX_CORE_CONTAINER_JOB_ID, 0);
    } else { 
   
      container = new TaskGroupContainer(allConf);
      instanceId = allConf.getLong(CoreConstant.DATAX_CORE_CONTAINER_JOB_ID);
      taskGroupId = allConf.getInt(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_ID);
      channelNumber = allConf.getInt(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL);
    }

    //缺省打开perfTrace
    boolean traceEnable = allConf.getBool(CoreConstant.DATAX_CORE_CONTAINER_TRACE_ENABLE, true);
    boolean perfReportEnable = allConf.getBool(CoreConstant.DATAX_CORE_REPORT_DATAX_PERFLOG, true);

    //standalone 模式的 datax shell任务不进行汇报
    if (instanceId == -1) { 
   
      perfReportEnable = false;
    }

    int priority = 0;
    try { 
   
      priority = Integer.parseInt(System.getenv("SKYNET_PRIORITY"));
    } catch (NumberFormatException e) { 
   
      LOG.warn("priority set to 0, because NumberFormatException, the value is: {}",
          System.getProperty("PROIORY"));
    }

    Configuration jobInfoConfig = allConf.getConfiguration(CoreConstant.DATAX_JOB_JOBINFO);
    //初始化PerfTrace
    PerfTrace perfTrace = PerfTrace
        .getInstance(isJob, instanceId, taskGroupId, priority, traceEnable);
    perfTrace.setJobInfo(jobInfoConfig, perfReportEnable, channelNumber);
    container.start();
  }


  /**
   * 过滤job配置信息
   *
   * @param configuration Configuration
   * @return String
   */
  public static String filterJobConfiguration(final Configuration configuration) { 
   
    Configuration jobConfWithSetting = configuration.getConfiguration("job").clone();
    Configuration jobContent = jobConfWithSetting.getConfiguration("content");
    jobConfWithSetting.set("content", filterSensitiveConfiguration(jobContent));
    return jobConfWithSetting.beautify();
  }

  /**
   * 屏蔽敏感信息
   *
   * @param conf Configuration
   * @return Configuration
   */
  public static Configuration filterSensitiveConfiguration(Configuration conf) { 
   
    Set<String> keys = conf.getKeys();
    for (final String key : keys) { 
   
      boolean isSensitive =
          endsWithIgnoreCase(key, "password") || endsWithIgnoreCase(key, "accessKey");
      if (isSensitive && conf.get(key) instanceof String) { 
   
        conf.set(key, conf.getString(key).replaceAll(".", "*"));
      }
    }
    return conf;
  }

  /**
   * @param args String[]
   * @throws Throwable
   */
  public static void entry(final String[] args) throws Throwable { 
   
    Options options = new Options();
    options.addOption("job", true, "Job config.");
    options.addOption("jobid", true, "Job unique id.");
    options.addOption("mode", true, "Job runtime mode.");

    BasicParser parser = new BasicParser();
    CommandLine cl = parser.parse(options, args);
    String jobPath = cl.getOptionValue("job");
    // 如果用户没有明确指定jobid, 则 datax.py 会指定 jobid 默认值为-1
    String jobIdString = cl.getOptionValue("jobid");
    RUNTIME_MODE = cl.getOptionValue("mode");
    Configuration conf = ConfigParser.parse(jobPath);
    long jobId;
    String defaultJobId = "-1";
    if (!defaultJobId.equals(jobIdString)) { 
   
      //  如果jobId相同，会怎样？
      jobId = Long.parseLong(jobIdString);
    } else { 
   
      // 如果用户没有指定jobId，或jobId==1，执行后面逻辑
      // only for dsc & ds & datax 3 update
      String dscJobUrlPatternStr = "/instance/(\\d{1,})/config.xml";
      String dsJobUrlPatternStr = "/inner/job/(\\d{1,})/config";
      String dsTaskGroupUrlPatternStr = "/inner/job/(\\d{1,})/taskGroup/";
      List<String> patterns = Arrays
          .asList(dscJobUrlPatternStr, dsJobUrlPatternStr, dsTaskGroupUrlPatternStr);
      jobId = parseJobIdFromUrl(patterns, jobPath);
    }

    boolean isStandAloneMode = "standalone".equalsIgnoreCase(RUNTIME_MODE);
    if (!isStandAloneMode && jobId == -1) { 
   
      // 如果不是 standalone 模式，那么 jobId 一定不能为-1
      throw DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR,
          "非 standalone 模式必须在 URL 中提供有效的 jobId.");
    }
    conf.set(CoreConstant.DATAX_CORE_CONTAINER_JOB_ID, jobId);
    //打印vmInfo
    VMInfo vmInfo = VMInfo.getVmInfo();
    if (vmInfo != null) { 
   
      LOG.info(vmInfo.toString());
    }

    LOG.info("\n" + filterJobConfiguration(conf) + "\n");
    LOG.debug(conf.toJSON());
    ConfigurationValidate.doValidate(conf);
    Engine engine = new Engine();
    engine.start(conf);
  }


  /**
   * -1 表示未能解析到 jobId
   * <p>
   * only for dsc & ds & datax 3 update
   */
  private static long parseJobIdFromUrl(List<String> patternStringList, String url) { 
   
    long result = -1;
    for (String patternString : patternStringList) { 
   
      result = doParseJobIdFromUrl(patternString, url);
      if (result != -1) { 
   
        return result;
      }
    }
    return result;
  }

  private static long doParseJobIdFromUrl(String patternString, String url) { 
   
    Pattern pattern = Pattern.compile(patternString);
    Matcher matcher = pattern.matcher(url);
    if (matcher.find()) { 
   
      return Long.parseLong(matcher.group(1));
    }
    return -1;
  }

  public static void main(String[] args) { 
   
    int exitCode = 0;
    try { 
   
      Engine.entry(args);
    } catch (Throwable e) { 
   
      exitCode = 1;
      LOG.error("\n\n经DataX智能分析,该任务最可能的错误原因是:\n {}", ExceptionTracker.trace(e));
      if (e instanceof DataXException) { 
   
        DataXException tempException = (DataXException) e;
        ErrorCode errorCode = tempException.getErrorCode();
        if (errorCode instanceof FrameworkErrorCode) { 
   
          FrameworkErrorCode tempErrorCode = (FrameworkErrorCode) errorCode;
          exitCode = tempErrorCode.toExitValue();
        }
      }
      System.exit(exitCode);
    }
    System.exit(exitCode);
  }

}

注：