datax(6):启动步骤解析

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

通过前面 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的执行过程

过程详细说明如下：

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

注：

1. 对源码进行略微改动，主要修改为 1 阿里代码规约扫描出来的，2 clean code；

2. 所有代码都已经上传到github（master分支和dev），可以免费白嫖