datax里所有的关系型数据库都走通用的处理Reader,com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader,当进行split的时候会进行切分获取channel的个数,
public List<Configuration> split(Configuration originalConfig, int adviceNumber) { return ReaderSplitUtil.doSplit(originalConfig, adviceNumber); }
继续进入doSplit方法com.alibaba.datax.plugin.rdbms.reader.util.ReaderSplitUtil 此处会去判断是否是table配置的模式
if (isTableMode) { // adviceNumber这里是channel数量大小, 即datax并发task数量 // eachTableShouldSplittedNumber是单表应该切分的份数, 向上取整可能和adviceNumber没有比例关系了已经 eachTableShouldSplittedNumber = calculateEachTableShouldSplittedNumber( adviceNumber, originalSliceConfig.getInt(Constant.TABLE_NUMBER_MARK)); }
如果前面计算的adviceNumber=3 配置了一个table 则每个表分到的channel是3/1=3,
// 说明是配置的 table 方式 if (isTableMode) { // 已在之前进行了扩展和`处理,可以直接使用 List<String> tables = connConf.getList(Key.TABLE, String.class); Validate.isTrue(null != tables && !tables.isEmpty(), "您读取数据库表配置错误."); String splitPk = originalSliceConfig.getString(Key.SPLIT_PK, null); /** * 1-判断是否配置了splitPk,如果没有配置则每个table都会当成一个任务,生成一个配置文件给任务运行使用 * 2-如果配置了splitPk,如果只配了一个table,则重新计算eachTableShouldSplittedNumber=eachTableShouldSplittedNumber * 5; * 3-如果配置了多个table,eachTableShouldSplittedNumber不变,然后循环对每个表进行切分splitSingleTable */ //最终切分份数不一定等于 eachTableShouldSplittedNumber boolean needSplitTable = eachTableShouldSplittedNumber > 1 && StringUtils.isNotBlank(splitPk); if (needSplitTable) { if (tables.size() == 1) { //原来:如果是单表的,主键切分num=num*2+1 // splitPk is null这类的情况的数据量本身就比真实数据量少很多, 和channel大小比率关系时,不建议考虑 //eachTableShouldSplittedNumber = eachTableShouldSplittedNumber * 2 + 1;// 不应该加1导致长尾 //考虑其他比率数字?(splitPk is null, 忽略此长尾) eachTableShouldSplittedNumber = eachTableShouldSplittedNumber * 5; } // 尝试对每个表,切分为eachTableShouldSplittedNumber 份 for (String table : tables) { tempSlice = sliceConfig.clone(); tempSlice.set(Key.TABLE, table); List<Configuration> splittedSlices = SingleTableSplitUtil .splitSingleTable(tempSlice, eachTableShouldSplittedNumber); splittedConfigs.addAll(splittedSlices); } } else { for (String table : tables) { tempSlice = sliceConfig.clone(); tempSlice.set(Key.TABLE, table); String queryColumn = HintUtil.buildQueryColumn(jdbcUrl, table, column); tempSlice.set(Key.QUERY_SQL, SingleTableSplitUtil.buildQuerySql(queryColumn, table, where)); splittedConfigs.add(tempSlice); } } } else { // 说明是配置的 querySql 方式 List<String> sqls = connConf.getList(Key.QUERY_SQL, String.class); // TODO 是否check 配置为多条语句?? for (String querySql : sqls) { tempSlice = sliceConfig.clone(); tempSlice.set(Key.QUERY_SQL, querySql); splittedConfigs.add(tempSlice); } }
此处主要分为两种,一种是table模式配置,一种是querysql模式配置, querySql模式相对简单,配置了几个sql,就会生成几个任务的配置文件。
我们主要关注table模式,主要的流程:
/** * 1-判断是否配置了splitPk,如果没有配置则每个table都会当成一个任务,生成一个配置文件给任务运行使用 * 2-如果配置了splitPk,如果只配了一个table,则重新计算eachTableShouldSplittedNumber=eachTableShouldSplittedNumber * 5; * 3-如果配置了多个table,eachTableShouldSplittedNumber不变,然后循环对每个表进行切分splitSingleTable */
接下来关注splitSingleTable方法
大体流程是:
首先会根据 Configuration configuration, int adviceNum 配置文件信息和需要切分的个数进行切分会计算出splitPk的最大最小值,然后按照adviceNum进行分割,然后生成具体的sql
如果配置了15个channel,单表, 拆分,则此时通过上面的流程最后可以计算出task数目为75,在我的mysql表中有3089条数据,此时他会返回下面的配置总共75个有效配置,主要是看querySql,每个配置的条件都会不同。这只是其中一个
{ "column": "id,username,telephone", "columnList": ["id", "username", "telephone"], "fetchSize": -2147483648, "isTableMode": true, "jdbcUrl": "jdbc:mysql://localhost:3306/datax?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true", "loadBalanceResourceMark": "localhost", "password": "root123", "pkType": "pkTypeLong", "querySql": "select id,username,telephone from user where (547 <= id AND id < 588) ", "splitPk": "id", "table": "user", "tableNumber": 1, "username": "root" } 第一个sql: "querySql": "select id,username,telephone from user where (1 <= id AND id < 43) " 最后一个sql是: "querySql": "select id,username,telephone from user where id IS NULL" 最后一个有效sql是: "querySql": "select id,username,telephone from user where (3048 <= id AND id <= 3089) " 可以看到和表中的数据一致,这里主要是把数据量过大的表,按照配置计算出的channel进行切分, 然后生成分段的sql,一个sql对应一个任务,进行执行。
后面会根据taskGroup的个数,来对着75个任务进行分组,然后提交到线程池中并发执行任务.
如果没有SPLIT_PK(目前splitPk仅支持整形、字符串数据切分),则为每个表生成一个任务
在DataX中,mysqlreader配置有两种模式,一种是table模式,另外一种是querySql模式,两种模式使用起来略有差别。
在table模式下, channel个数决定了reader和writer的个数上限,假设为m个:如果指定了splitPk字段,DataX会将mysql表中数据按照splitPk切分成n段,n大致为5倍的channel个数。
splitPk的字段限制了必需是整型或者字符串类型。由于DataX的实现方式是按照spliPk字段分段查询数据库表,那么spliPk字段的选取应该尽可能的选择分布均匀且有索引的字段,比如主键id、唯一键等字段。DataX会启动m个reader线程,消费DataX切分好的n个查询sql语句(task), 对应的会有m个writer线程将查询出来的数据写入目标数据源中,并行度为m(也就是配置的channel个数),如果不指定splitPk字段,DataX将不会进行数据的切分,并行度直接退化成1。
需要指出的是,table模式下,如果用户指定了spliPk将数据切分成了n段,由于这些task不是在同一个事务下进行select,那么最终取出的全量数据很有可能是不一致的。为了拿到一致性数据,要么不要配置spliPk使用单线程,要么确保mysql中要导出的数据不会再发生变化。
querySql模式一般用于有条件的数据导出,
"connection": [ { "querySql": [ #指定执行的SQL语句 "select bucket_name, delta , timestamp ,cdn_in, cdn_out ,total_request from vip_quota where bucket_name='xxx' " ], "jdbcUrl": ["jdbc:mysql://10.10.0.8:3306/db1?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true" #jdbc连接串 ] } ]
在此模式下,DataX不会再按照指定的column、table参数进行sql的拼接,而是会直接略过这些配置(如果有),直接执行querySql语句,task数量总是1,因此在此模式下channel的配置不再有多线程的效果。
有人肯定会有疑问,有什么办法可以尽可能加速数据的导出呢?
一般来说,大家首先想到的是提高并发度。在DataX中channel的个数决定了并发数量,但是要使channel参数生效,并不是简单配一下channel就完事了。在MySQL导入Tablestore表的场景下,channel生效仅在能够split出多个SQL语句的场景下,也就是table模式+spliPk下有用。
DataX的数据同步涉及三部分:
1.数据读取 2.数据交换 3.数据写入
对于以上三个环节,都有不同的优化方式,分析如下。
1.数据读取
对于数据源读取,导出的两种模式:table模式和sqlQuery模式前面做了阐述,这里不再重复。
2. 数据交换
对于数据交换,前面提到,发送给MySQL数据库SQL语句后会得到查询的数据集,缓存在DataX的buffer中;除此之外,每个channel也维护了自己的record队列,如果存在并发,channel的个数越多,也会需要更多的内存。因此首先需要考虑的是jvm的内存大小参数, 这个在启动jvm进程的时候配置。
除此之外,有几个控制channel的关键参数
"transport": { "channel": { "class": "com.alibaba.datax.core.transport.channel.memory.MemoryChannel", "speed": { "byte": -1, "record": -1 }, "flowControlInterval": 20, "capacity": 512, "byteCapacity": 67108864 }, "exchanger": { "class": "com.alibaba.datax.core.plugin.BufferedRecordExchanger", "bufferSize": 32 } },
以上配置位于conf/core.json中:capacity限制了channel中队列的大小(也就是最多缓存record的个数)byteCapacity限制了record占用的内存大小,core.json中的默认配置是64MB,若不指定将会被配置为8MB
这两个参数决定了每个channel能buffer的记录数量和内存占用情况,如果有需要调整,用户应该按照DataX实际的运行环境予以配置。例如MySQL中每个record都比较大,那么可以考虑适当调高byteCapacity,当然调整这个参数还要考虑机器的内存情况。
bufferSize指定了BufferedRecordExchanger的缓存,reader读了多少个往channel放。
一般情况下,channel队列本身配置的调整并不会很常见,但是对于另外几个流控参数,在使用DataX的时候应该注意。有两个常用的流控参数:
a. byte 限制通道的默认传输速率, -1表示不限制
b. record 限制通道的传输记录数,-1表示不限制
这两个参数都是在flowControlInterval间隔里采样后根据采样值来决定是否流控的。
{ "core": { #定义了全局的系统参数,不指定会使用默认值 "transport": { "channel": { "speed": { "record": 5000, "byte": 102400 } } } }, "job": { "setting": { "speed": { #定义了单个channel的控制参数 "record": 10000, }, "errorLimit": { "record": 0, "percentage": 0.02 } }, "content": [ { "reader": { .....#省略 }, "writer": { .....#省略 } } ] } }
3.数据写入
适当的提高批量写入的批次大小(batchWriteCount),也可以有效地提高吞吐率。相关关键配置如下:
{ "job": { "setting": { ....#省略 }, "content": [ { "reader": { .....#省略 }, "writer": { "name": "otswriter", "parameter": { ....... "writeMode":"UpdateRow", "batchWriteCount":100 } } } ] } }