解讀Datax mysql reader配置

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模式

在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模式

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
                    }
                }

            }
        ]
    }
}