異構數據庫遷移——DATAX

時間 2019-11-16

標籤異構數據庫遷移 datax 欄目 SQL 简体版

原文原文鏈接

背景

在最近接觸到的一個case裏面，須要把db2的數據遷移至oracle，客戶可接收的停機時間爲3小時。mysql

同步方式的比較

一說到停機時間，你們第一時間想到Oracle公司的GoldenGate實時同步工具。但在測試過程當中發現，因爲沒法提早檢查，並且初始化時間好久等問題，致使咱們最後不得不放棄使用這一神器。git

既然OGG不能使用，那能傳統導出文本再用sql load導入，那是否可行呢？根據以往的經驗，只要數據一落地就存在亂碼，數據錯位等問題，因爲沒法進行hash對帳，數據質量根本沒法保證。github

我司的某平臺軟件，直接從內存中進行數據轉換，讓咱們看到一大但願。因爲列的順序不一，沒法知足部分需求，只能又放棄這一神器。sql

就在此時，提出了一個開源軟件——DATAX。數據庫

什麼是DATAX

DataX 是阿里巴巴集團內被普遍使用的離線數據同步工具/平臺，實現包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS 等各類異構數據源之間高效的數據同步功能。數組

注意，這裏說的是離線數據同步工具，就是必須得停機以後再進行數據同步。那DATAX究竟支持哪些數據庫進行數據同步呢？如下是GitHub提供的列表：服務器

類型網絡	數據源多線程	Reader(讀)架構	Writer(寫)	文檔
RDBMS 關係型數據庫	MySQL	√	√	讀、寫
	Oracle	√	√	讀、寫
	SQLServer	√	√	讀、寫
	PostgreSQL	√	√	讀、寫
	DRDS	√	√	讀、寫
	通用RDBMS(支持全部關係型數據庫)	√	√	讀、寫
阿里雲數倉數據存儲	ODPS	√	√	讀、寫
	ADS		√	寫
	OSS	√	√	讀、寫
	OCS	√	√	讀、寫
NoSQL數據存儲	OTS	√	√	讀、寫
	Hbase0.94	√	√	讀、寫
	Hbase1.1	√	√	讀、寫
	MongoDB	√	√	讀、寫
	Hive	√	√	讀、寫
無結構化數據存儲	TxtFile	√	√	讀、寫
	FTP	√	√	讀、寫
	HDFS	√	√	讀、寫
	Elasticsearch		√	寫

DataX3.0核心架構

在這個case裏面，咱們使用datax3.0的核心架構。DataX 3.0 開源版本支持單機多線程模式完成同步做業運行，按一個DataX做業生命週期的時序圖，從總體架構設計很是簡要說明DataX各個模塊相互關係。

核心模塊介紹：

DataX完成單個數據同步的做業，咱們稱之爲Job，DataX接受到一個Job以後，將啓動一個進程來完成整個做業同步過程。DataX Job模塊是單個做業的中樞管理節點，承擔了數據清理、子任務切分(將單一做業計算轉化爲多個子Task)、TaskGroup管理等功能。

DataXJob啓動後，會根據不一樣的源端切分策略，將Job切分紅多個小的Task(子任務)，以便於併發執行。Task即是DataX做業的最小單元，每個Task都會負責一部分數據的同步工做。

切分多個Task以後，DataX Job會調用Scheduler模塊，根據配置的併發數據量，將拆分紅的Task從新組合，組裝成TaskGroup(任務組)。每個TaskGroup負責以必定的併發運行完畢分配好的全部Task，默認單個任務組的併發數量爲5。

每個Task都由TaskGroup負責啓動，Task啓動後，會固定啓動Reader—>Channel—>Writer的線程來完成任務同步工做。

DataX做業運行起來以後， Job監控並等待多個TaskGroup模塊任務完成，等待全部TaskGroup任務完成後Job成功退出。不然，異常退出，進程退出值非0

JSON配置

JOB全局配置：

JOB經過配置對同步整個過程進行管控。

"job": {
        "setting": {
            "speed": {
                "byte": 1048576
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.02
            }
        },

speed爲同步速度限制參數，這裏有三個參數channel、record和byte。

channel：管道數，能夠理解爲並行數，需與splitPk一同使用，不然無效果。

record：每次同步多少條數據，取record和byte中的最小值

byte：每次同步多少字節數據，取record和byte中的最小值

errorLimit爲錯誤數據限制，這裏有兩個參數record和percentage，指當異常數據達到多少時同步取消，取record和percentage的最小值。

Reader線程配置：

RDBMSReader經過JDBC鏈接器鏈接到遠程的RDBMS數據庫，並根據用戶配置的信息生成查詢SELECT SQL語句併發送到遠程RDBMS數據庫，並將該SQL執行返回結果使用DataX自定義的數據類型拼裝爲抽象的數據集，並傳遞給下游Writer處理。

對於用戶配置Table、Column、Where的信息，RDBMSReader將其拼接爲SQL語句發送到RDBMS數據庫；對於用戶配置querySql信息，RDBMS直接將其發送到RDBMS數據庫。

DB2被阿里分爲通用RDBMS的範疇。配置一個從RDBMS數據庫同步抽取數據做業樣例：

"content": [
            {
                "reader": {
                    "name": "rdbmsreader",
                    "parameter": {
                        "username": "xxx",
                        "password": "xxx",
                        "column": [
                            "id",
                            "name"
                        ],
                        "splitPk": "pk",
                        "connection": [
                            {
                                "table": [
                                    "table"
                                ],
                                "jdbcUrl": [
                                    "jdbc:dm://ip:port/database"
                                ]
                            }
                        ],
                        "fetchSize": 1024,
                        "where": "1 = 1"
                    }
                },

username

源數據庫的用戶名

password

源數據庫的密碼

encoding

數據庫字符集,GBK,UTF8等

column

所配置的表中須要同步的列名集合，使用JSON的數組描述字段信息。用戶使用表明默認使用全部列配置，例如['']。

支持列裁剪，即列能夠挑選部分列進行導出。

支持列換序，即列能夠不按照表schema信息進行導出。

支持常量配置，用戶須要按照JSON格式: ["id", "1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"] id爲普通列名，1爲整形數字常量，'bazhen.csy'爲字符串常量，null爲空指針，to_char(a + 1)爲表達式，2.3爲浮點數，true爲布爾值。

Column必須顯示填寫，不容許爲空！

splitPk

RDBMSReader進行數據抽取時，若是指定splitPk，表示用戶但願使用splitPk表明的字段進行數據分片，DataX所以會啓動併發任務進行數據同步，這樣能夠大大提供數據同步的效能。

推薦splitPk用戶使用表主鍵，由於表主鍵一般狀況下比較均勻，所以切分出來的分片也不容易出現數據熱點。

目前splitPk僅支持整形數據切分，不支持浮點、字符串型、日期等其餘類型。若是用戶指定其餘非支持類型，RDBMSReader將報錯！

注意：這裏並不是只能是主鍵，擁有惟一約束的列也可。

table

所選取的須要同步的表名

jdbcUrl

描述的是到對端數據庫的JDBC鏈接信息，jdbcUrl按照RDBMS官方規範，並能夠填寫鏈接附件控制信息。請注意不一樣的數據庫jdbc的格式是不一樣的，DataX會根據具體jdbc的格式選擇合適的數據庫驅動完成數據讀取。

db2格式 jdbc:db2://ip:port/database

fetchSize

該配置項定義了插件和數據庫服務器端每次批量數據獲取條數，該值決定了DataX和服務器端的網絡交互次數，可以較大的提高數據抽取性能。

注意，該值最大建議值爲2048。

where

篩選條件，RDBMSReader根據指定的column、table、where條件拼接SQL，並根據這個SQL進行數據抽取。例如在作測試時，能夠將where條件指定爲limit 10；在實際業務場景中，每每會選擇當天的數據進行同步，能夠將where條件指定爲gmt_create > $bizdate 。

小結：

我司某產品不被咱們選用主要是配置中不存在column和where兩個條件的參數，沒法對部分數據進行過濾。

Writer線程配置：

OracleWriter 插件實現了寫入數據到 Oracle 主庫的目的表的功能。在底層實現上， OracleWriter 經過 JDBC 鏈接遠程 Oracle 數據庫，並執行相應的 insert into ... sql 語句將數據寫入 Oracle，內部會分批次提交入庫。

OracleWriter 面向ETL開發工程師，他們使用 OracleWriter 從數倉導入數據到 Oracle。同時 OracleWriter 亦能夠做爲數據遷移工具爲DBA等用戶提供服務。

OracleWriter 經過 DataX 框架獲取 Reader 生成的協議數據，根據你配置生成相應的SQL語句

insert into...(當主鍵/惟一性索引衝突時會寫不進去衝突的行)

對於使用datax同步到oracle的表，建議刪除主鍵和惟一索引，數據校驗完成後再進行重建。

                "writer": {
                    "name": "oraclewriter",
                    "parameter": {
                        "username": "root",
                        "password": "root",
                        "column": [
                            "id",
                            "name"
                        ],
                        "preSql": [
                            "delete from test"
                        ],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:oracle:thin:@[HOST_NAME]:PORT:[DATABASE_NAME]",
                                "table": [
                                    "test"
                                ]
                            }
                        ]
                    }

username

目的數據庫的用戶名

password

目的數據庫的密碼

encoding

數據庫字符集

batchSize

一次性批量提交的記錄數大小，該值能夠極大減小DataX與Oracle的網絡交互次數，並提高總體吞吐量。最大隻能設置爲1024

column

的表須要寫入數據的字段,字段之間用英文逗號分隔。例如: "column": ["id","name","age"]。若是要依次寫入所有列，使用表示, 例如: "column": [""]

preSql

執行語句時會先檢查是否存在若存在，根據條件刪除