DataX插件二次開發指南

一、 DataX為什么要使用插件機制？

從設計之初，DataX就把異構資料源同步作為自身的使命，為了應對不同資料源的差異、同時提供一致的同步原語和擴展能力，DataX自然而然地采用了框架 + 插件 的模式：

• 插件只需關心資料的讀取或者寫入本身，
• 而同步的共性問題，比如：型別轉換、性能、統計，則交由框架來處理，

作為插件開發人員，則需要關注兩個問題：

1. 資料源本身的讀寫資料正確性，
2. 如何與框架溝通、合理正確地使用框架，

二、插件視角看框架

邏輯執行模型

插件開發者基本只需要關注特定資料源系統的讀和寫，以及自己的代碼在邏輯上是怎樣被執行的，哪一個方法是在什么時候被呼叫的，開發之前需要明確以下概念：

• Job: Job是DataX用以描述從一個源頭到一個目的端的同步作業，是DataX資料同步的最小業務單元，比如：從一張mysql的表同步到odps的一個表的特定磁區，
• Task: Task是為最大化而把Job拆分得到的最小執行單元，比如：讀一張有1024個分表的mysql分庫分表的Job，拆分成1024個讀Task，用若干個并發執行，
• TaskGroup: 描述的是一組Task集合，在同一個TaskGroupContainer執行下的Task集合稱之為TaskGroup
• JobContainer: Job執行器，負責Job全域拆分、調度、前置陳述句和后置陳述句等作業的作業單元，類似Yarn中的JobTracker
• TaskGroupContainer: TaskGroup執行器，負責執行一組Task的作業單元，類似Yarn中的TaskTracker，

簡而言之， Job拆分成Task，在分別在框架提供的容器中執行，插件只需要實作Job和Task兩部分邏輯，

物理執行模型

框架為插件提供物理上的執行能力（執行緒），DataX框架有三種運行模式：

• Standalone: 單行程運行，沒有外部依賴，
• Local: 單行程運行，統計資訊、錯誤資訊匯報到集中存盤，
• Distrubuted: 分布式多行程運行，依賴DataX Service服務，

當然，上述三種模式對插件的撰寫而言沒有什么區別，你只需要避開一些小錯誤，插件就能夠在單機/分布式之間無縫切換了，
當JobContainer和TaskGroupContainer運行在同一個行程內時，就是單機模式（Standalone和Local）；當它們分布在不同的行程中執行時，就是分布式（Distributed）模式，

編程介面

Job和Task的邏輯是怎么對應到具體的代碼中的？

首先，插件的入口類必須擴展Reader或Writer抽象類，并且實作分別實作Job和Task兩個內部抽象類，Job和Task的實作必須是 內部類 的形式，原因見 加載原理 一節，以Reader為例：

public class SomeReader extends Reader {
    public static class Job extends Reader.Job {

        @Override
        public void init() {
        }
		
		@Override
		public void prepare() {
        }

        @Override
        public List<Configuration> split(int adviceNumber) {
            return null;
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

    }

    public static class Task extends Reader.Task {

        @Override
        public void init() {
        }
		
		@Override
		public void prepare() {
        }

        @Override
        public void startRead(RecordSender recordSender) {
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }
    }
}

Job介面功能如下：

• init: Job物件初始化作業，此時可以通過super.getPluginJobConf()獲取與本插件相關的配置，讀插件獲得配置中reader部分，寫插件獲得writer部分，
• prepare: 全域準備作業，比如mysqlwriter在寫入新資料之前執行一個truncate table的操作、讀取Hive資料之前，完成Kerberos認證等，
• split: 拆分Task，引數adviceNumber框架建議的拆分數，一般是運行時所配置的并發度，值回傳的是Task的配置串列，
• post: 全域的后置作業，比如mysqlwriter同步完影子表后的rename操作，
• destroy: Job物件自身的銷毀作業，

Task介面功能如下：

• init：Task物件的初始化，此時可以通過super.getPluginJobConf()獲取與本Task相關的配置，這里的配置是Job的split方法回傳的配置串列中的其中一個，
• prepare：區域的準備作業，
• startRead: 從資料源讀資料，寫入到RecordSender中，RecordSender會把資料寫入連接Reader和Writer的快取佇列，
• startWrite：從RecordReceiver中讀取資料，寫入目標資料源，RecordReceiver中的資料來自Reader和Writer之間的快取佇列，
• post: 區域的后置作業，
• destroy: Task象自身的銷毀作業，

需要注意的是：

• Job和Task之間一定不能有共享變數，因為分布式運行時不能保證共享變數會被正確初始化，兩者之間只能通過組態檔進行依賴，
• prepare和post在Job和Task中都存在，插件需要根據實際情況確定在什么地方執行操作，

框架按照如下的順序執行Job和Task的介面：

上圖中，黃色表示Job部分的執行階段，藍色表示Task部分的執行階段，綠色表示框架執行階段，

相關類關系如下：

插件定義

代碼寫好了，有沒有想過框架是怎么找到插件的入口類的？框架是如何加載插件的呢？

在每個插件的專案中，都有一個plugin.json檔案，這個檔案定義了插件的相關資訊，包括入口類，例如：

{
    "name": "mysqlwriter",
    "class": "com.alibaba.datax.plugin.writer.mysqlwriter.MysqlWriter",
    "description": "Use Jdbc connect to database, execute insert sql.",
    "developer": "alibaba"
}

• name: 插件名稱，大小寫敏感，框架根據用戶在組態檔中指定的名稱來搜尋插件， 十分重要 ，
• class: 入口類的全限定名稱，框架通過反射插件入口類的實體，十分重要 ，
• description: 描述資訊，
• developer: 開發人員，

打包發布

DataX使用assembly打包，assembly的使用方法請咨詢谷哥或者度娘，打包命令如下：

mvn clean package -DskipTests assembly:assembly

DataX插件需要遵循統一的目錄結構：

${DATAX_HOME}
|-- bin       
|   `-- datax.py
|-- conf
|   |-- core.json
|   `-- logback.xml
|-- lib
|   `-- datax-core-dependencies.jar
`-- plugin
    |-- reader
    |   `-- mysqlreader
    |       |-- libs
    |       |   `-- mysql-reader-plugin-dependencies.jar
    |       |-- mysqlreader-0.0.1-SNAPSHOT.jar
    |       `-- plugin.json
    `-- writer
        |-- mysqlwriter
        |   |-- libs
        |   |   `-- mysql-writer-plugin-dependencies.jar
        |   |-- mysqlwriter-0.0.1-SNAPSHOT.jar
        |   `-- plugin.json
        |-- oceanbasewriter
        `-- odpswriter

• ${DATAX_HOME}/bin: 可執行程式目錄，
• ${DATAX_HOME}/conf: 框架配置目錄，
• ${DATAX_HOME}/lib: 框架依賴庫目錄，
• ${DATAX_HOME}/plugin: 插件目錄，

插件目錄分為reader和writer子目錄，讀寫插件分別存放，插件目錄規范如下：

• ${PLUGIN_HOME}/libs: 插件的依賴庫，
• ${PLUGIN_HOME}/plugin-name-version.jar: 插件本身的jar，
• ${PLUGIN_HOME}/plugin.json: 插件描述檔案，

盡管框架加載插件時，會把${PLUGIN_HOME}下所有的jar放到classpath，但還是推薦依賴庫的jar和插件本身的jar分開存放，

注意：
插件的目錄名字必須和plugin.json中定義的插件名稱一致，

組態檔

DataX使用json作為組態檔的格式，一個典型的DataX任務配置如下：

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "odpsreader",
          "parameter": {
            "accessKey": "",
            "accessId": "",
            "column": [""],
            "isCompress": "",
            "odpsServer": "",
            "partition": [
              ""
            ],
            "project": "",
            "table": "",
            "tunnelServer": ""
          }
        },
        "writer": {
          "name": "oraclewriter",
          "parameter": {
            "username": "",
            "password": "",
            "column": ["*"],
            "connection": [
              {
                "jdbcUrl": "",
                "table": [
                  ""
                ]
              }
            ]
          }
        }
      }
    ]
  }
}

DataX框架有core.json組態檔，指定了框架的默認行為，任務的配置里頭可以指定框架中已經存在的配置項，而且具有更高的優先級，會覆寫core.json中的默認值，

配置中job.content.reader.parameter的value部分會傳給Reader.Job；job.content.writer.parameter的value部分會傳給Writer.Job ，Reader.Job和Writer.Job可以通過super.getPluginJobConf()來獲取，

DataX框架支持對特定的配置項進行RSA加密，例子中以*開頭的專案便是加密后的值， 配置項加密解密程序對插件是透明，插件仍然以不帶*的key來查詢配置和操作配置項 ，

如何設計配置引數

組態檔的設計是插件開發的第一步！

任務配置中reader和writer下parameter部分是插件的配置引數，插件的配置引數應當遵循以下原則：

• 駝峰命名：所有配置項采用駝峰命名法，首字母小寫，單詞首字母大寫，

• 正交原則：配置項必須正交，功能沒有重復，沒有潛規則，

• 富型別：合理使用json的型別，減少無謂的處理邏輯，減少出錯的可能，

  • 使用正確的資料型別，比如，bool型別的值使用true/false，而非"yes"/"true"/0等，
  • 合理使用集合型別，比如，用陣列替代有分隔符的字串，

• 類似通用：遵守同一型別的插件的習慣，比如關系型資料庫的connection引數都是如下結構：

  {
    "connection": [
      {
        "table": [
          "table_1",
          "table_2"
        ],
        "jdbcUrl": [
          "jdbc:mysql://127.0.0.1:3306/database_1",
          "jdbc:mysql://127.0.0.2:3306/database_1_slave"
        ]
      },
      {
        "table": [
          "table_3",
          "table_4"
        ],
        "jdbcUrl": [
          "jdbc:mysql://127.0.0.3:3306/database_2",
          "jdbc:mysql://127.0.0.4:3306/database_2_slave"
        ]
      }
    ]
  }
  

• ...

如何使用Configuration類

為了簡化對json的操作，DataX提供了簡單的DSL配合Configuration類使用，

Configuration提供了常見的get, 帶型別get，帶默認值get，set等讀寫配置項的操作，以及clone, toJSON等方法，配置項讀寫操作都需要傳入一個path做為引數，這個path就是DataX定義的DSL，語法有兩條：

1. 子map用.key表示，path的第一個點省略，
2. 陣列元素用[index]表示，

比如操作如下json：

{
  "a": {
    "b": {
      "c": 2
    },
    "f": [
      1,
      2,
      {
        "g": true,
        "h": false
      },
      4
    ]
  },
  "x": 4
}

比如呼叫configuration.get(path)方法，當path為如下值的時候得到的結果為：

• x：4
• a.b.c：2
• a.b.c.d：null
• a.b.f[0]：1
• a.b.f[2].g：true

注意，因為插件看到的配置只是整個配置的一部分，使用Configuration物件時，需要注意當前的根路徑是什么，

更多Configuration的操作請參考ConfigurationTest.java，

插件資料傳輸

跟一般的生產者-消費者模式一樣，Reader插件和Writer插件之間也是通過channel來實作資料的傳輸的，channel可以是記憶體的，也可能是持久化的，插件不必關心，插件通過RecordSender往channel寫入資料，通過RecordReceiver從channel讀取資料，

channel中的一條資料為一個Record的物件，Record中可以放多個Column物件，這可以簡單理解為資料庫中的記錄和列，

Record有如下方法：

public interface Record {
    // 加入一個列，放在最后的位置
    void addColumn(Column column);
    // 在指定下標處放置一個列
    void setColumn(int i, final Column column);
    // 獲取一個列
    Column getColumn(int i);
    // 轉換為json String
    String toString();
    // 獲取總列數
    int getColumnNumber();
    // 計算整條記錄在記憶體中占用的位元組數
    int getByteSize();
}

因為Record是一個介面，Reader插件首先呼叫RecordSender.createRecord()創建一個Record實體，然后把Column一個個添加到Record中，

Writer插件呼叫RecordReceiver.getFromReader()方法獲取Record，然后把Column遍歷出來，寫入目標存盤中，當Reader尚未退出，傳輸還在進行時，如果暫時沒有資料RecordReceiver.getFromReader()方法會阻塞直到有資料，如果傳輸已經結束，會回傳null，Writer插件可以據此判斷是否結束startWrite方法，

Column的構造和操作，我們在《型別轉換》一節介紹，

型別轉換

為了規范源端和目的端型別轉換操作，保證資料不失真，DataX支持六種內部資料型別：

• Long：定點數(Int、Short、Long、BigInteger等)，
• Double：浮點數(Float、Double、BigDecimal(無限精度)等)，
• String：字串型別，底層不限長，使用通用字符集(Unicode)，
• Date：日期型別，
• Bool：布林值，
• Bytes：二進制，可以存放諸如MP3等非結構化資料，

對應地，有DateColumn、LongColumn、DoubleColumn、BytesColumn、StringColumn和BoolColumn六種Column的實作，

Column除了提供資料相關的方法外，還提供一系列以as開頭的資料型別轉換轉換方法，

DataX的內部型別在實作上會選用不同的java型別：

型別之間相互轉換的關系如下：

臟資料處理

什么是臟資料？

目前主要有三類臟資料：

1. Reader讀到不支持的型別、不合法的值，
2. 不支持的型別轉換，比如：Bytes轉換為Date，
3. 寫入目標端失敗，比如：寫mysql整型長度超長，

如何處理臟資料

在Reader.Task和Writer.Task中，通過AbstractTaskPlugin.getTaskPluginCollector()可以拿到一個TaskPluginCollector，它提供了一系列collectDirtyRecord的方法，當臟資料出現時，只需要呼叫合適的collectDirtyRecord方法，把被認為是臟資料的Record傳入即可，

用戶可以在任務的配置中指定臟資料限制條數或者百分比限制，當臟資料超出限制時，框架會結束同步任務，退出，插件需要保證臟資料都被收集到，其他作業交給框架就好，

加載原理

1. 框架掃描plugin/reader和plugin/writer目錄，加載每個插件的plugin.json檔案，
2. 以plugin.json檔案中name為key，索引所有的插件配置，如果發現重名的插件，框架會例外退出，
3. 用戶在插件中在reader/writer配置的name欄位指定插件名字，框架根據插件的型別（reader/writer）和插件名稱去插件的路徑下掃描所有的jar，加入classpath，
4. 根據插件配置中定義的入口類，框架通過反射實體化對應的Job和Task物件，

三、插件介紹檔案

每個插件都必須在DataX官方wiki中有一篇檔案，檔案需要包括但不限于以下內容：

1. 快速介紹：介紹插件的使用場景，特點等，
2. 實作原理：介紹插件實作的底層原理，比如mysqlwriter通過insert into和replace into來實作插入，tair插件通過tair客戶端實作寫入，
3. 配置說明
   • 給出典型場景下的同步任務的json組態檔，
   • 介紹每個引數的含義、是否必選、默認值、取值范圍和其他約束，
4. 型別轉換
   • 插件是如何在實際的存盤型別和DataX的內部型別之間進行轉換的，
   • 以及是否存在特殊處理，
5. 性能報告
   • 軟硬體環境，系統版本，java版本，CPU、記憶體等，
   • 資料特征，記錄大小等，
   • 測驗引數集（多組），系統引數（比如并發數），插件引數（比如batchSize）
   • 不同引數下同步速度（Rec/s, MB/s），機器負載（load, cpu）等，對資料源壓力（load, cpu, mem等），
6. 約束限制：是否存在其他的使用限制條件，
7. FAQ：用戶經常會遇到的問題，

標籤：其他

上一篇：用戶行為分析模型實踐（三）——H5通用分析模型

下一篇：InnoDB在RR隔離級別下的幻讀問題的分析