概述

DataX是阿里巴巴開源的一個異構數據源離線同步工具，致力于實現包括關系型數據庫（MySQL、Oracle等）、HDFS、Hive、ODPS、HBase、FTP等各種異構數據源之間穩定高效的數據同步功能。您可以通過DataX服務讀取上游數據，然后由DataX Doris Writer將數據寫入到云數據庫 SelectDB 版。

前提條件

• 已安裝Maven環境。

• 已安裝Python3.6及以上版本。

使用示例

如下以MySQL數據源為例，介紹在Linux環境下如何通過DataX將MySQL數據導入至云數據庫 SelectDB 版。

步驟一：配置DataX環境

1. 下載DataX程序包代碼，插件代碼下載請訪問Doris社區。

2. 運行DataX程序包中的init-env.sh腳本，構建DataX開發環境。

   sh init-env.sh

3. 編譯mysqlreader和doriswriter。

   1. 編譯整個DataX項目。

      cd DataX/
      mvn package assembly:assembly -Dmaven.test.skip=true

      編譯產出結果在target/datax/datax/.目錄下。

      說明

      hdfsreader，hdfswriter，ossreader和osswriter這四個插件需要額外的jar包，如果不需要這些插件，可以在DataX/pom.xml中刪除這些插件的模塊。

   2. 單獨編譯mysqlreader和selectdbwriter插件。

      mvn clean install -pl plugin-rdbms-util,mysqlreader,doriswriter -DskipTests

步驟二：構造需要導入的數據

1. 創建MySQL測試表。

   CREATE TABLE `employees` (
     `emp_no` int NOT NULL,
     `birth_date` date NOT NULL,
     `first_name` varchar(14) NOT NULL,
     `last_name` varchar(16) NOT NULL,
     `gender` enum('M','F') NOT NULL,
     `hire_date` date NOT NULL,
     PRIMARY KEY (`emp_no`)
   ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3

2. 使用DMS構建測試數據，詳情請參見測試數據構建。

步驟三：配置云數據庫 SelectDB 版實例。

1. 通過MySQL協議連接云數據庫 SelectDB 版實例，詳情請參見連接實例。

2. 創建測試數據庫和測試表。

   1. 創建測試數據庫。

      CREATE DATABASE test_db;

   2. 創建測試表。

      USE test_db;
      CREATE TABLE employees (
          emp_no       int NOT NULL,
          birth_date   date,
          first_name   varchar(20),
          last_name    varchar(20),
          gender       char(2),
          hire_date    date
      )
      UNIQUE KEY(`emp_no`)
      DISTRIBUTED BY HASH(`emp_no`) BUCKETS 1;

步驟四：通過DataX服務同步MySQL數據到SelectDB

1. 開通云數據庫 SelectDB 版公網地址，詳情請參見申請和釋放公網地址。

2. 將DataX主機的公網IP添加到IP白名單中，詳情請參見設置白名單。

3. 創建配置文件mysqlToSelectDB.json，配置任務信息。

   {
     "job":{
       "content":[
         {
           "reader":{
               "name": "mysqlreader",
               "parameter": {
                   "column": [
                       "emp_no",
                       "birth_date",
                       "first_name",
                       "last_name",
                       "gender",
                       "hire_date"
                   ],
                   "where": "emp_no>0",
                   "connection": [
                       {
                           "jdbcUrl": [
                               "jdbc:mysql://host:port/test_db?useUnicode=true&allowPublicKeyRetrieval=true&characterEncoding=utf-8"
                           ],
                           "table": [
                               "employees"
                           ]
                       }
                   ],
                   "password": "123456",
                   "splitPk": "emp_no",
                   "username": "admin"
               }
           },
           "writer":{
             "name":"doriswriter",
             "parameter":{
               "loadUrl":[
                 "selectdb-cn-xxx-public.selectdbfe.rds.aliyuncs.com:8080"
               ],
               "loadProps":{
                 "format":"json",
                 "strip_outer_array":"true"
               },
               "column":[
                   "emp_no",
                   "birth_date",
                   "first_name",
                   "last_name",
                   "gender",
                   "hire_date"
               ],
               "username":"admin",
               "password":"123456",
               "postSql":[
   
               ],
               "preSql":[
   
               ],
               "connection":[
                 {
                   "jdbcUrl":"jdbc:mysql://selectdb-cn-xxx-public.selectdbfe.rds.aliyuncs.com:9030/test_db",
                   "table":[
                     "employees"
                   ],
                   "selectedDatabase":"test_db"
                 }
               ],
               "maxBatchRows":1000000,
               "batchSize":536870912000
             }
           }
         }
       ],
       "setting":{
         "errorLimit":{
           "percentage":0.02,
           "record":0
         },
         "speed":{
           "channel":5
         }
       }
     }
   }

4. 參數說明

   參數	是否必填	默認值	描述
   jdbcUrl	是	無	JDBC連接URLjdbc:mysql://<ip>:<port>。 您可以從云數據庫 SelectDB 版控制臺的實例詳情 > 網絡信息中獲取IP地址和MySQL協議端口。 IP地址：VPC地址或公網地址。 端口：MySQL協議端口。 示例：jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030 說明 如果DataX主機和SelectDB在同一VPC下，即可使用VPC地址。如果不在同一VPC下請使用公網地址。
   loadUrl	是	無	SelectDB的HTTP協議訪問地址<ip>:<port>。 您可以從云數據庫 SelectDB 版控制臺的實例詳情 > 網絡信息中獲取IP地址和HTTP協議端口。 IP地址：VPC地址或公網地址。 端口：HTTP協議端口。 示例：selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080 說明 如果DataX主機和SelectDB在同一VPC下，即可使用VPC地址。如果不在同一VPC下請使用公網地址。
   username	是	無	云數據庫 SelectDB 版實例的用戶名。
   password	是	無	云數據庫 SelectDB 版實例對應用戶的密碼。
   connection.selectedDatabase	是	無	需要寫入的云數據庫 SelectDB 版數據庫名稱。
   connection.table	是	無	需要寫入的云數據庫 SelectDB 版表名稱。
   column	是	無	目的表需要寫入數據的字段，這些字段將作為生成的JSON數據的字段名。字段之間用英文逗號分隔。示例："column": ["id","name","age"]。
   preSql	否	無	寫入數據到目的表前，會先執行這里的標準語句。
   postSql	否	無	寫入數據到目的表后，會執行這里的標準語句。
   maxBatchRows	否	500000	每批次導入數據的最大行數。和batchSize共同控制每批次的導入數量。每批次數據達到兩個閾值之一，即開始導入這一批次的數據。
   batchSize	否	104857600	每批次導入數據的最大數據量。和maxBatchRows共同控制每批次的導入數量。每批次數據達到兩個閾值之一，即開始導入這一批次的數據。默認值為100 M。
   maxRetries	否	3	每批次導入數據失敗后的重試次數。
   labelPrefix	否	datax_doris_writer_	每批次上傳文件的label前綴。最終的label將由 'labelPrefix + UUID'組成全局唯一的label，確保數據不會重復導入。
   loadProps	否	無	與Stream Load的請求參數相同。詳情請參見Stream Load參數說明。配置導入數據格式使用參數format，導入數據格式默認使用CSV，支持JSON，詳情請參考類型轉換。
   flushInterval	否	30000	數據寫入批次的時間間隔。默認為30000ms

5. 命令行提交任務。

   cd target/datax/datax/bin
   python datax.py ../mysqlToSelectDB.json

類型轉換

默認傳入的數據均會被轉為字符串，并以\t作為列分隔符，\n作為行分隔符，組成csv文件進行SelectDB導入操作。 默認是csv格式導入，如需更改列分隔符， 則正確配置loadProps即可，示例如下。

"loadProps": {
    "format": "csv",
    "column_separator": "\\x01",
    "line_delimiter": "\\x02"
}

如需更改導入格式為JSON，則正確配置loadProps下的format即可，示例如下。

"loadProps": {
    "format": "json",
    "strip_outer_array": true
}