datax底层原理_Datax 任务分配原理
Datax 任务分配原理
Datax根首先据配置文件，确定好channel的并发数目。然后将整个job分成一个个小的task，然后划分成组。
确定channel数目
如果指定字节数限速，则计算字节限速后的并发数目。如果指定记录数限速，则计算记录数限速后的并发数目。再取两者中最小的channel并发数目。如果两者限速都没指定，则看是否配置文件指定了channel并发数目。
比如以下面这个配置为例：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21{
"core": {
"transport" : {
"channel": {
"speed": {
"record": 100,
"byte": 100
}
}
}
},
"job": {
"setting": {
"speed": {
"record": 500,
"byte": 1000,
"channel" : 1
}
}
}
}
首先计算按照字节数限速，channel的数目应该为 500 / 100 = 5
然后按照记录数限速， channel的数目应该为 1000 ／ 100 = 10
最后返回两者的最小值 5。虽然指定了channel为1， 但只有在没有限速的条件下，才会使用。
adjustChannelNumber方法，实现了上述功能
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74private void adjustChannelNumber(){
int needChannelNumberByByte = Integer.MAX_VALUE;
int needChannelNumberByRecord = Integer.MAX_VALUE;
// 是否指定字节数限速
boolean isByteLimit = (this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_BYTE, 0) > 0);
if (isByteLimit) {
// 总的限速字节数
long globalLimitedByteSpeed = this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_BYTE, 10 * 1024 * 1024);
// 单个Channel的字节数
Long channelLimitedByteSpeed = this.configuration
.getLong(CoreConstant.DATAX_CORE_TRANSPORT_CHANNEL_SPEED_BYTE);
if (channelLimitedByteSpeed == null || channelLimitedByteSpeed <= 0) {
DataXException.asDataXException(
FrameworkErrorCode.CONFIG_ERROR,
"在有总bps限速条件下，单个channel的bps值不能为空，也不能为非正数");
}
// 计算所需要的channel数目
needChannelNumberByByte =
(int) (globalLimitedByteSpeed / channelLimitedByteSpeed);
needChannelNumberByByte =
needChannelNumberByByte > 0 ? needChannelNumberByByte : 1;
}
// 是否指定记录数限速
boolean isRecordLimit = (this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_RECORD, 0)) > 0;
if (isRecordLimit) {
// 总的限速记录数
long globalLimitedRecordSpeed = this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_RECORD, 100000);
// 获取单个channel的限定的记录数
Long channelLimitedRecordSpeed = this.configuration.getLong(
CoreConstant.DATAX_CORE_TRANSPORT_CHANNEL_SPEED_RECORD);
if (channelLimitedRecordSpeed == null || channelLimitedRecordSpeed <= 0) {
DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR,
"在有总tps限速条件下，单个channel的tps值不能为空，也不能为非正数");
}
// 计算所需要的channel数目
needChannelNumberByRecord =
(int) (globalLimitedRecordSpeed / channelLimitedRecordSpeed);
needChannelNumberByRecord =
needChannelNumberByRecord > 0 ? needChannelNumberByRecord : 1;
LOG.info("Job set Max-Record-Speed to " + globalLimitedRecordSpeed + " records.");
}
// 取较小值
this.needChannelNumber = needChannelNumberByByte < needChannelNumberByRecord ?
needChannelNumberByByte : needChannelNumberByRecord;
// 返回最小值
if (this.needChannelNumber < Integer.MAX_VALUE) {
return;
}
// 是否指定了channel数目
boolean isChannelLimit = (this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_CHANNEL, 0) > 0);
if (isChannelLimit) {
this.needChannelNumber = this.configuration.getInt(
CoreConstant.DATAX_JOB_SETTING_SPEED_CHANNEL);
LOG.info("Job set Channel-Number to " + this.needChannelNumber
+ " channels.");
return;
}
throw DataXException.asDataXException(
FrameworkErrorCode.CONFIG_ERROR,
"Job运行速度必须设置");
}
切分任务
JobContainer 的split负责将整个job切分成多个task，生成task配置的列表
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30private int split(){
// 计算所需的channel数目
this.adjustChannelNumber();
if (this.needChannelNumber <= 0) {
this.needChannelNumber = 1;
}
// 生成任务的reader配置
List readerTaskConfigs = this
.doReaderSplit(this.needChannelNumber);
int taskNumber = readerTaskConfigs.size();
// 生成任务的writer配置
List writerTaskConfigs = this
.doWriterSplit(taskNumber);
// 生成任务的transformer配置
List transformerList = this.configuration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT_TRANSFORMER);
LOG.debug("transformer configuration: "+ JSON.toJSONString(transformerList));
// 合并任务的reader，writer，transformer配置
List contentConfig = mergeReaderAndWriterTaskConfigs(
readerTaskConfigs, writerTaskConfigs, transformerList);
LOG.debug("contentConfig configuration: "+ JSON.toJSONString(contentConfig));
// 将配置结果保存在job.content路径下
this.configuration.set(CoreConstant.DATAX_JOB_CONTENT, contentConfig);
return contentConfig.size();
}
Reader
doReaderSplit方法， 调用Reader.Job的split方法，返回Reader.Task的Configuration列表
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20/**
* adviceNumber, 建议的数目
*/
private List doReaderSplit(int adviceNumber){
// 切换ClassLoader
classLoaderSwapper.setCurrentThreadClassLoader(LoadUtil.getJarLoader(
PluginType.READER, this.readerPluginName));
// 调用Job.Reader的split切分
List readerSlicesConfigs =
this.jobReader.split(adviceNumber);
if (readerSlicesConfigs == null || readerSlicesConfigs.size() <= 0) {
throw DataXException.asDataXException(
FrameworkErrorCode.PLUGIN_SPLIT_ERROR,
"reader切分的task数目不能小于等于0");
}
LOG.info("DataX Reader.Job [{}] splits to [{}] tasks.",
this.readerPluginName, readerSlicesConfigs.size());
classLoaderSwapper.restoreCurrentThreadClassLoader();
return readerSlicesConfigs;
}
Writer
doWriterSplit方法， 调用Writer.JOb的split方法，返回Writer.Task的Configuration列表
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18private List doWriterSplit(int readerTaskNumber){
// 切换ClassLoader
classLoaderSwapper.setCurrentThreadClassLoader(LoadUtil.getJarLoader(
PluginType.WRITER, this.writerPluginName));
// 调用Job.Reader的split切分
List writerSlicesConfigs = this.jobWriter
.split(readerTaskNumber);
if (writerSlicesConfigs == null || writerSlicesConfigs.size() <= 0) {
throw DataXException.asDataXException(
FrameworkErrorCode.PLUGIN_SPLIT_ERROR,
"writer切分的task不能小于等于0");
}
LOG.info("DataX Writer.Job [{}] splits to [{}] tasks.",
this.writerPluginName, writerSlicesConfigs.size());
classLoaderSwapper.restoreCurrentThreadClassLoader();
return writerSlicesConfigs;
}
合并配置
合并reader，writer，transformer配置列表。并将任务列表，保存在配置job.content的值里。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37private List mergeReaderAndWriterTaskConfigs(
List readerTasksConfigs,
List writerTasksConfigs,
List transformerConfigs){
// reader切分的任务数目必须等于writer切分的任务数目
if (readerTasksConfigs.size() != writerTasksConfigs.size()) {
throw DataXException.asDataXException(
FrameworkErrorCode.PLUGIN_SPLIT_ERROR,
String.format("reader切分的task数目[%d]不等于writer切分的task数目[%d].",
readerTasksConfigs.size(), writerTasksConfigs.size())
);
}
List contentConfigs = new ArrayList();
for (int i = 0; i < readerTasksConfigs.size(); i++) {
Configuration taskConfig = Configuration.newDefault();
// 保存reader相关配置
taskConfig.set(CoreConstant.JOB_READER_NAME,
this.readerPluginName);
taskConfig.set(CoreConstant.JOB_READER_PARAMETER,
readerTasksConfigs.get(i));
// 保存writer相关配置
taskConfig.set(CoreConstant.JOB_WRITER_NAME,
this.writerPluginName);
taskConfig.set(CoreConstant.JOB_WRITER_PARAMETER,
writerTasksConfigs.get(i));
// 保存transformer相关配置
if(transformerConfigs!=null && transformerConfigs.size()>0){
taskConfig.set(CoreConstant.JOB_TRANSFORMER, transformerConfigs);
}
taskConfig.set(CoreConstant.TASK_ID, i);
contentConfigs.add(taskConfig);
}
return contentConfigs;
}
分配任务
分配算法首先根据指定的channel数目和每个Taskgroup的拥有channel数目，计算出Taskgroup的数目
根据每个任务的reader.parameter.loadBalanceResourceMark将任务分组
根据每个任务writer.parameter.loadBalanceResourceMark来讲任务分组
根据上面两个任务分组的组数，挑选出大的那个组
轮询上面步骤的任务组，依次轮询的向各个TaskGroup添加一个，直到所有任务都被分配完
这里举个实例：
目前有7个task，channel有20个，每个Taskgroup拥有5个channel。
首先计算出Taskgroup的数目， 20 ／ 5 = 4 。
根据reader.parameter.loadBalanceResourceMark，将任务分组如下：
1
2
3
4
5{
"database_a" : [task_id_1, task_id_2],
"database_b" : [task_id_3, task_id_4, task_id_5],
"database_c" : [task_id_6, task_id_7]
}
根据writer.parameter.loadBalanceResourceMark，将任务分组如下：
1
2
3
4{
"database_dst_d" : [task_id_1, task_id_2],
"database_dst_e" : [task_id_3, task_id_4, task_id_5, task_id_6, task_id_7]
}
因为readerResourceMarkAndTaskIdMap有三个组，而writerResourceMarkAndTaskIdMap只有两个组。从中选出组数最多的，所以这里按照readerResourceMarkAndTaskIdMap将任务分配。
执行过程是，轮询database_a, database_b, database_c，取出第一个。循环上一步
取出task_id_1 放入 taskGroup_1
取出task_id_3 放入 taskGroup_2
取出task_id_6 放入 taskGroup_3
取出task_id_2 放入 taskGroup_4
取出task_id_4 放入 taskGroup_1
………
最后返回的结果为
1
2
3
4
5
6{
"taskGroup_1": [task_id_1, task_id_4],
"taskGroup_2": [task_id_3, task_id_7],
"taskGroup_3": [task_id_6, task_id_5],
"taskGroup_4": [task_id_2]
}
代码解释
任务的分配是由JobAssignUtil类负责。使用者调用assignFairly方法，传入参数，返回TaskGroup配置列表
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23public final class JobAssignUtil{
/**
* configuration 配置
* channelNumber， channel总数
* channelsPerTaskGroup， 每个TaskGroup拥有的channel数目
*/
public static List assignFairly(Configuration configuration, int channelNumber, int channelsPerTaskGroup){
List contentConfig = configuration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT);
// 计算TaskGroup的数目
int taskGroupNumber = (int) Math.ceil(1.0 * channelNumber / channelsPerTaskGroup);
......
// 任务分组
LinkedHashMap> resourceMarkAndTaskIdMap = parseAndGetResourceMarkAndTaskIdMap(contentConfig);
// 调用doAssign方法，分配任务
List taskGroupConfig = doAssign(resourceMarkAndTaskIdMap, configuration, taskGroupNumber);
// 调整 每个 taskGroup 对应的 Channel 个数(属于优化范畴)
adjustChannelNumPerTaskGroup(taskGroupConfig, channelNumber);
return taskGroupConfig;
}
}
任务分组
按照task配置的reader.parameter.loadBalanceResourceMark和writer.parameter.loadBalanceResourceMark，分别对任务进行分组，选择分组数最高的那组，作为任务分组的源。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39/**
* contentConfig参数，task的配置列表
*/
private static LinkedHashMap> parseAndGetResourceMarkAndTaskIdMap(List contentConfig) {
// reader的任务分组，key为分组的名称，value是taskId的列表
LinkedHashMap> readerResourceMarkAndTaskIdMap = new LinkedHashMap>();
// writer的任务分组，key为分组的名称，value是taskId的列表
LinkedHashMap>
writerResourceMarkAndTaskIdMap = new LinkedHashMap>();
for (Configuration aTaskConfig : contentConfig) {
int taskId = aTaskConfig.getInt(CoreConstant.TASK_ID);
// 取出reader.parameter.loadBalanceResourceMark的值，作为分组名
String readerResourceMark = aTaskConfig.getString(CoreConstant.JOB_READER_PARAMETER + "." + CommonConstant.LOAD_BALANCE_RESOURCE_MARK);
if (readerResourceMarkAndTaskIdMap.get(readerResourceMark) == null) {
readerResourceMarkAndTaskIdMap.put(readerResourceMark, new LinkedList());
}
// 把 readerResourceMark 加到 readerResourceMarkAndTaskIdMap 中
readerResourceMarkAndTaskIdMap.get(readerResourceMark).add(taskId);
// 取出writer.parameter.loadBalanceResourceMark的值，作为分组名
String writerResourceMark = aTaskConfig.getString(CoreConstant.JOB_WRITER_PARAMETER + "." + CommonConstant.LOAD_BALANCE_RESOURCE_MARK);
if (writerResourceMarkAndTaskIdMap.get(writerResourceMark) == null) {
writerResourceMarkAndTaskIdMap.put(writerResourceMark, new LinkedList());
}
// 把 writerResourceMark 加到 writerResourceMarkAndTaskIdMap 中
writerResourceMarkAndTaskIdMap.get(writerResourceMark).add(taskId);
}
// 选出reader和writer其中最大的
if (readerResourceMarkAndTaskIdMap.size() >= writerResourceMarkAndTaskIdMap.size()) {
// 采用 reader 对资源做的标记进行 shuffle
return readerResourceMarkAndTaskIdMap;
} else {
// 采用 writer 对资源做的标记进行 shuffle
return writerResourceMarkAndTaskIdMap;
}
}
分配任务
将上一部任务的分组，划分到每个taskGroup里
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56private static List doAssign(LinkedHashMap> resourceMarkAndTaskIdMap, Configuration jobConfiguration, int taskGroupNumber){
List contentConfig = jobConfiguration.getListConfiguration(CoreConstant.DATAX_JOB_CONTENT);
Configuration taskGroupTemplate = jobConfiguration.clone();
taskGroupTemplate.remove(CoreConstant.DATAX_JOB_CONTENT);
List result = new LinkedList();
// 初始化taskGroupConfigList
List> taskGroupConfigList = new ArrayList>(taskGroupNumber);
for (int i = 0; i < taskGroupNumber; i++) {
taskGroupConfigList.add(new LinkedList());
}
// 取得resourceMarkAndTaskIdMap的值的最大个数
int mapValueMaxLength = -1;
List resourceMarks = new ArrayList();
for (Map.Entry> entry : resourceMarkAndTaskIdMap.entrySet()) {
resourceMarks.add(entry.getKey());
if (entry.getValue().size() > mapValueMaxLength) {
mapValueMaxLength = entry.getValue().size();
}
}
int taskGroupIndex = 0;
// 执行mapValueMaxLength次数，每一次轮询一遍resourceMarkAndTaskIdMap
for (int i = 0; i < mapValueMaxLength; i++) {
// 轮询resourceMarkAndTaskIdMap
for (String resourceMark : resourceMarks) {
if (resourceMarkAndTaskIdMap.get(resourceMark).size() > 0) {
// 取出第一个
int taskId = resourceMarkAndTaskIdMap.get(resourceMark).get(0);
// 轮询的向taskGroupConfigList插入值
taskGroupConfigList.get(taskGroupIndex % taskGroupNumber).add(contentConfig.get(taskId));
// taskGroupIndex自增
taskGroupIndex++;
// 删除第一个
resourceMarkAndTaskIdMap.get(resourceMark).remove(0);
}
}
}
Configuration tempTaskGroupConfig;
for (int i = 0; i < taskGroupNumber; i++) {
tempTaskGroupConfig = taskGroupTemplate.clone();
// 设置TaskGroup的配置
tempTaskGroupConfig.set(CoreConstant.DATAX_JOB_CONTENT, taskGroupConfigList.get(i));
tempTaskGroupConfig.set(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_ID, i);
result.add(tempTaskGroupConfig);
}
// 返回结果
return result;
}
为组分配channel
上面已经把任务划分成多个组，为了每个组能够均匀的分配channel，还需要调整。算法原理是，当channel总的数目，不能整除TaskGroup的数目时。多的余数个channel，从中挑选出余数个TaskGroup，每个多分配一个。
比如现在有13个channel，然后taskgroup确有5个。那么首先每个组先分 13 / 5 = 2 个。那么还剩下多的3个chanel，分配给前面个taskgroup。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16private static void adjustChannelNumPerTaskGroup(List taskGroupConfig, int channelNumber){
int taskGroupNumber = taskGroupConfig.size();
int avgChannelsPerTaskGroup = channelNumber / taskGroupNumber;
int remainderChannelCount = channelNumber % taskGroupNumber;
// 表示有 remainderChannelCount 个 taskGroup,其对应 Channel 个数应该为：avgChannelsPerTaskGroup + 1；
// (taskGroupNumber - remainderChannelCount)个 taskGroup,其对应 Channel 个数应该为：avgChannelsPerTaskGroup
int i = 0;
for (; i < remainderChannelCount; i++) {
taskGroupConfig.get(i).set(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL, avgChannelsPerTaskGroup + 1);
}
for (int j = 0; j < taskGroupNumber - remainderChannelCount; j++) {
taskGroupConfig.get(i + j).set(CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_CHANNEL, avgChannelsPerTaskGroup);
}
}
datax底层原理_Datax 任务分配原理相关推荐
- php底层运行机制与原理
php底层运行机制与原理 1 PHP的设计理念及特点 多进程模型:由于PHP是多进程模型,不同请求间互不干涉,这样保证了一个请求挂掉不会对全盘服务造成影响,当然,时代发展,PHP也早已支持多线程模型. ...
- ASP.NET页面与IIS底层交互和工作原理详解 （二）
第三回: 引言 Http 请求处理流程 和 Http Handler 介绍 这两篇文章里,我们首先了解了Http请求在服务器端的处理流程,随后我们知道Http请求最终会由实现了IHttpHandler ...
- ASP.NET页面与IIS底层交互和工作原理详解（一）
第一回: 引言 我查阅过不少Asp.Net的书籍,发现大多数作者都是站在一个比较高的层次上讲解Asp.Net.他们耐心.细致地告诉你如何一步步拖放控件.设置控件属性.编写CodeBehind代码,以实 ...
- 国仁网络资讯：微信视频号怎么变现赚钱；首先要了解平台的底层逻辑与算法原理。
互联网,犹如浩瀚星空,里面的生态正如星际爆炸一样,在不断地肆虐扩张,但是如果单从内容这一条线的发展来看,其实脉络十分清晰. 每伴随着一次内容推荐算法的迭代与优化,就但是想玩转一个平台前,一定要先明白该 ...
- 越来越多的岗位需要DPDK，那从DPDK该如何提升网络底层效率丨网络原理丨Linux服务器开发丨后端开发丨网络底层原理
越来越多的岗位需要dpdk,那从dpdk该如何提升网络底层效率 1. dpdk线程模型 2. kni与数据接收处理流程 3. 手把手代码实现 视频讲解如下,点击观看: 越来越多的岗位需要DPDK,那从 ...
- 高性能分布式缓存Redis--- Redis底层结构和缓存原理 --- 持续更新
高性能分布式缓存Redis全系列文章主目录(进不去说明还没写完)https://blog.csdn.net/grd_java/article/details/124192973 本文只是整个系列笔记的 ...
- 只会用框架，看不懂源码，不了解其底层机制与实现原理，成了一名只会搬运源码库的开发。说白了，就是真正牛逼的技术不属于你。
对于开发来说,我们在工作中普遍都会用到各个开源框架,比如最基础的 Spring,使开发网络编程变得特别简单的 Netty 框架,还有成为目前微服务框架首选的 Spring Cloud 等.在多个框架之 ...
- grpc通信原理_容器原理架构详解（全）
目录 1 容器原理架构 1.1 容器与虚拟化 1.2 容器应用架构 1.3 容器引擎架构 1.4 Namespace与Cgroups 1.5 容器镜像原理 2 K8S原理架构 2.1 K8S主要功能 ...
- 【重难点】【JUC 04】synchronized 原理、ReentrantLock 原理、synchronized 和 Lock 的对比、CAS 无锁原理
[重难点][JUC 04]synchronized 原理.ReentrantLock 原理.synchronized 和 Lock 的对比.CAS 无锁原理 文章目录 [重难点][JUC 04]syn ...
- serverlet 原理_容器原理架构详解（全）
目录 1 容器原理架构:容器与虚拟化.容器应用/引擎架构.Namespace与Cgroups.镜像原理 2 K8S原理架构:K8S主要功能.K8S 系统架构.Pod原理与调度 3 K8S存储方案:容器 ...
最新文章
- 你们这行我懂，不给点好处都不接！
- Hangfire入门（任务调度）
- 诗与远方：无题（九十三）
- php yii框架连接数据库,Yii 框架使用数据库(databases)的方法示例
- 动手写procedure以及注意的细节
- 评价类模型：1.层次分析法
- zoj 3703(背包)
- 产品小姐姐收到这个黑科技后，开心了一整天...
- 关于 CSS will-change 属性你需要知道的事
- Linux常用的基础组件
- 【Spring笔记】Spring介绍IOC理论推导
- 字典树(Trie树)的实现及应用
- mysql inner 连接多表_MySQL数据库之多表查询inner join内连接
- 如何改善在线游戏的体验？有哪些实用技巧？
- H3CIE（WLAN）学习笔记（4）——PHY层协议
- 机器人电焊电流电压怎么调_电焊机电流如何调整，气保焊机电压电流怎么调
- 微信/支付宝扫码支付流程
- 【语音识别入门】Python音频处理示例（含完整代码）
- 在Windows上将桌面图标做成贪吃蛇游戏
- 计算机455端口,455端口怎么关闭-455端口关闭的方法 - 河东软件园