一、DataX工具簡介
1、設計理念
DataX是一個異構資料源離線同步工具，緻力于實作包括關系型資料庫(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各種異構資料源之間穩定高效的資料同步功能。解決異構資料源同步問題，DataX将複雜的網狀的同步鍊路變成了星型資料鍊路，DataX作為中間傳輸載體負責連接配接各種資料源。當需要接入一個新的資料源的時候，隻需要将此資料源對接到DataX，便能跟已有的資料源做到無縫資料同步。
絮叨一句：異構資料源指，為了處理不同種類的業務，使用不同的資料庫系統存儲資料。
2、元件結構
DataX本身作為離線資料同步架構，采用Framework+plugin架構建構。将資料源讀取和寫入抽象成為Reader和Writer插件，納入到整個同步架構中。
- Reader
Reader為資料采集子產品，負責讀取采集資料源的資料，将資料發送給Framework。
- Writer
Writer為資料寫入子產品，負責不斷向Framework取資料，并将資料寫入到目的端。
- Framework
Framework用于連接配接reader和writer，作為兩者的資料傳輸通道，并處理緩沖，流控，并發，資料轉換等核心技術問題。
3、架構設計
- Job
DataX完成單個資料同步的作業，稱為Job，DataX接受到一個Job之後，将啟動一個程序來完成整個作業同步過程。Job子產品是單個作業的中樞管理節點，承擔了資料清理、子任務切分(将單一作業計算轉化為多個子Task)、TaskGroup管理等功能。
- Split
DataXJob啟動後，會根據不同的源端切分政策，将Job切分成多個小的Task(子任務)，以便于并發執行。Task便是DataX作業的最小單元，每一個Task都會負責一部分資料的同步工作。
- Scheduler
切分多個Task之後，Job會調用Scheduler子產品，根據配置的并發資料量，将拆分成的Task重新組合，組裝成TaskGroup(任務組)。
- TaskGroup
每一個TaskGroup負責以一定的并發運作完畢配置設定好的所有Task，預設單個任務組的并發數量為5。每一個Task都由TaskGroup負責啟動，Task啟動後，會固定啟動Reader—>Channel—>Writer的線程來完成任務同步工作。DataX作業運作起來之後，Job監控并等待多個TaskGroup子產品任務完成，等待所有TaskGroup任務完成後Job成功退出。否則，異常退出，程序退出值非0。
二、環境安裝
推薦Python2.6+，Jdk1.8+(腦補安裝流程)。
1、Python包下載下傳
# yum -y install wget# wget https://www.python.org/ftp/python/2.7.15/Python-2.7.15.tgz# tar -zxvf Python-2.7.15.tgz 2、安裝Python
# yum install gcc openssl-devel bzip2-devel[[email protected] Python-2.7.15]# ./configure --enable-optimizations# make altinstall# python -V 3、DataX安裝
# pwd/opt/module# lldatax# cd /opt/module/datax/bin-- 測試環境是否正确# python datax.py /opt/module/datax/job/job.json 三、同步任務
1、同步表建立
-- PostgreSQLCREATE TABLE sync_user (id INT NOT NULL,user_name VARCHAR (32) NOT NULL,user_age int4 NOT NULL,CONSTRAINT "sync_user_pkey" PRIMARY KEY ("id"));CREATE TABLE data_user (id INT NOT NULL,user_name VARCHAR (32) NOT NULL,user_age int4 NOT NULL,CONSTRAINT "sync_user_pkey" PRIMARY KEY ("id")); 2、編寫任務腳本
[[email protected] job]# pwd/opt/module/datax/job[[email protected] job]# vim postgresql_job.json 3、腳本内容
{ "job": { "setting": { "speed": { "channel": "3" } }, "content": [ { "reader": { "name": "postgresqlreader", "parameter": { "username": "root01", "password": "123456", "column": ["id","user_name","user_age"], "connection": [ { "jdbcUrl": ["jdbc:postgresql://192.168.72.131:5432/db_01"], "table": ["data_user"] } ] } }, "writer": { "name": "postgresqlwriter", "parameter": { "username": "root01", "password": "123456", "column": ["id","user_name","user_age"], "connection": [ { "jdbcUrl": "jdbc:postgresql://192.168.72.131:5432/db_01", "table": ["sync_user"] } ], "postSql": [], "preSql": [] } } } ] }} 4、執行腳本
# /opt/module/datax/bin/datax.py /opt/module/datax/job/postgresql_job.json 5、執行日志
2020-04-23 18:25:33.404 [job-0] INFO JobContainer - 任務啟動時刻 : 2020-04-23 18:25:22任務結束時刻 : 2020-04-23 18:25:33任務總計耗時 : 10s任務平均流量 : 1B/s記錄寫入速度 : 0rec/s讀出記錄總數 : 2讀寫失敗總數 : 0 四、源碼流程分析
注意：這裡源碼隻貼出核心流程，如果要看完整源碼，可以自行從Git上下載下傳。
1、讀取資料
核心入口：PostgresqlReader
啟動讀任務
public static class Task extends Reader.Task { @Override public void startRead(RecordSender recordSender) { int fetchSize = this.readerSliceConfig.getInt(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE); this.commonRdbmsReaderSlave.startRead(this.readerSliceConfig, recordSender, super.getTaskPluginCollector(), fetchSize); }} 讀取任務啟動之後，執行讀取資料操作。
核心類：CommonRdbmsReader
public void startRead(Configuration readerSliceConfig, RecordSender recordSender, TaskPluginCollector taskPluginCollector, int fetchSize) { ResultSet rs = null; try { // 資料讀取 rs = DBUtil.query(conn, querySql, fetchSize); queryPerfRecord.end(); ResultSetMetaData metaData = rs.getMetaData(); columnNumber = metaData.getColumnCount(); PerfRecord allResultPerfRecord = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.RESULT_NEXT_ALL); allResultPerfRecord.start(); long rsNextUsedTime = 0; long lastTime = System.nanoTime(); // 資料傳輸至交換區 while (rs.next()) { rsNextUsedTime += (System.nanoTime() - lastTime); this.transportOneRecord(recordSender, rs,metaData, columnNumber, mandatoryEncoding, taskPluginCollector); lastTime = System.nanoTime(); } allResultPerfRecord.end(rsNextUsedTime); }catch (Exception e) { throw RdbmsException.asQueryException(this.dataBaseType, e, querySql, table, username); } finally { DBUtil.closeDBResources(null, conn); }} 2、資料傳輸
核心接口：RecordSender(發送)
public interface RecordSender {public Record createRecord();public void sendToWriter(Record record);public void flush();public void terminate();public void shutdown();} 核心接口：RecordReceiver(接收)
public interface RecordReceiver {public Record getFromReader();public void shutdown();} 核心類：BufferedRecordExchanger
class BufferedRecordExchanger implements RecordSender, RecordReceiver 3、寫入資料
核心入口：PostgresqlWriter
啟動寫任務
public static class Task extends Writer.Task {public void startWrite(RecordReceiver recordReceiver) {this.commonRdbmsWriterSlave.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector());}} 寫資料任務啟動之後，執行資料寫入操作。
核心類：CommonRdbmsWriter
public void startWriteWithConnection(RecordReceiver recordReceiver, Connection connection) { // 寫資料庫的SQL語句 calcWriteRecordSql(); List writeBuffer = new ArrayList<>(this.batchSize); int bufferBytes = 0; try { Record record; while ((record = recordReceiver.getFromReader()) != null) { writeBuffer.add(record); bufferBytes += record.getMemorySize(); if (writeBuffer.size() >= batchSize || bufferBytes >= batchByteSize) { doBatchInsert(connection, writeBuffer); writeBuffer.clear(); bufferBytes = 0; } } if (!writeBuffer.isEmpty()) { doBatchInsert(connection, writeBuffer); writeBuffer.clear(); bufferBytes = 0; } } catch (Exception e) { throw DataXException.asDataXException( DBUtilErrorCode.WRITE_DATA_ERROR, e); } finally { writeBuffer.clear(); bufferBytes = 0; DBUtil.closeDBResources(null, null, connection); }} 推薦閱讀：源碼 -> GitHub || GitEE
資料源管理 | 基于JDBC模式，适配和管理動态資料源
資料源管理 | 動态權限校驗，表結構和資料遷移流程
資料源管理 | 主從庫動态路由，AOP模式讀寫分離
資料源管理 | 關系型分庫分表，列式庫分布式計算
資料源管理 | PostgreSQL環境整合，JSON類型應用