使用dataX批量從ob遷移表到oracle

IT那活兒發(fā)布于2023-01-11 13:19 / 2939人閱讀

點擊上方“IT那活兒”公眾號，關(guān)注后了解更多內(nèi)容，不管IT什么活兒，干就完了?。?！

某核心系統(tǒng)業(yè)務(wù)部分業(yè)務(wù)遷移到國產(chǎn)數(shù)據(jù)庫OceanBase，但是歷史數(shù)據(jù)仍然需要遷移到Oracle數(shù)據(jù)庫,對數(shù)據(jù)實時性要求不高，只需要清理前完成1個月的數(shù)據(jù)遷移即可。

開始一段時間使用的是OMS遷移，但是OMS遷移不支持單個分區(qū)的遷移，只能全表遷移，數(shù)據(jù)量較大，遷移時間較長。并且后續(xù)還有業(yè)務(wù)遷移到國產(chǎn)數(shù)據(jù)庫OceanBase需要備份，所以考慮使用dataX遷移數(shù)據(jù)。

一

DataX

1.1 DataX簡介

DataX 是阿里巴巴集團內(nèi)被廣泛使用的離線數(shù)據(jù)同步工具，致力于實現(xiàn)包括：關(guān)系型數(shù)據(jù)庫(MySQL、Oracle等)、HDFS、Hive、HBase、ODPS、FTP等各種異構(gòu)數(shù)據(jù)源之間穩(wěn)定高效的數(shù)據(jù)同步功能。

為了解決異構(gòu)數(shù)據(jù)源同步問題，DataX將復(fù)雜的網(wǎng)狀的同步鏈路變成了星型數(shù)據(jù)鏈路，DataX作為中間傳輸載體負(fù)責(zé)連接各種數(shù)據(jù)源。當(dāng)需要接入一個新的數(shù)據(jù)源的時候，只需要將此數(shù)據(jù)源對接到DataX，便能跟已有的數(shù)據(jù)源做到無縫數(shù)據(jù)同步。

1.2 DataX 3.0 框架

DataX本身作為離線數(shù)據(jù)同步框架，采用Framework + plugin架構(gòu)構(gòu)建。

將數(shù)據(jù)源讀取和寫入抽象成為Reader/Writer插件，納入到整個同步框架中，比較簡潔。

Reader：Reader為數(shù)據(jù)采集模塊，負(fù)責(zé)采集數(shù)據(jù)源的數(shù)據(jù)，將數(shù)據(jù)發(fā)送給Framework。
Writer：Writer為數(shù)據(jù)寫入模塊，負(fù)責(zé)不斷向Framework取數(shù)據(jù)，并將數(shù)據(jù)寫入到目的端。
Framework：Framework用于連接reader和writer，作為兩者的數(shù)據(jù)傳輸通道，并處理緩沖，流控，并發(fā)，數(shù)據(jù)轉(zhuǎn)換等核心技術(shù)問題。

二

datax的使用

2.1 在Oracle創(chuàng)建目標(biāo)表的表結(jié)構(gòu)

datax同步數(shù)據(jù)需要先在目標(biāo)端創(chuàng)建相應(yīng)的表結(jié)構(gòu)。

在這里是使用的OMS直接從OB遷移表結(jié)構(gòu)到Oracle。

2.2 創(chuàng)建配置文件（json格式）

Datax的基本執(zhí)行語句為:

python  {DATAX_HOME}/bin/datax.py  {JSON_FILE_NAME}.json

每個任務(wù)的參數(shù)文件是一個 json 格式，主要由setting和一個 reader 和一個 writer 組成。

批量配置json為：

{
  "job": {
    "setting": { 
      "speed": {
        "channel": 100,
    "bytes"：0
      },
      "errorLimit": {
     "record":"",
       "percentage":
      }
    },
    "content": [
      {
        "reader": {
          "name": "oceanbasev10reader",
          "parameter": {
            "username": "",
            "password": "", 
            "column": [
              "*"
            ],
            "connection": [
              { 
                           "jdbcUrl":[ ""],
                            "querySql": [
                                           "select * from ${readTb} PARTITION(${readpartition}) "
                 ] 
              }
            ],
            "batchSize": 1024
          }
        },
        "writer": {
          "name": "oraclewriter",
          "parameter": {
            "where": "",
            "column": ["*"],
            "preSql": [],
            "connection": [
              {
                                "jdbcUrl": "",
                                "table": ["${writeTb}"]
              }
            ],
            "username": "",
            "password": ""
          }
        }
      }
    ]
  }
}

channel表示任務(wù)并發(fā)數(shù)。
record: 出錯記錄數(shù)超過record設(shè)置的條數(shù)時，任務(wù)標(biāo)記為失敗.
percentage: 當(dāng)出錯記錄數(shù)超過percentage百分?jǐn)?shù)時，任務(wù)標(biāo)記為失敗.
bytes表示每秒字節(jié)數(shù)，默認(rèn)為0（不限速）。

2.3 ${readTb} 在json文件中表示變量，傳輸時使用-D參數(shù)進(jìn)行設(shè)置.

單個表同步：

python /home/admin/tools/datax3/bin/datax.py 
/home/admin/ob_ss/ob_to_ora_ss_tbcs5.json -p"-DreadTb=table1 
-Dreadpartition=partition1 -DwriteTb=table1 "

同步完成后，job相關(guān)信息：

2.4 批量執(zhí)行腳本

#!/bin/bash

v_table_list=/home/admin/ob_ss/source_table.lst
v_exec_command=/home/admin/tools/datax3/bin/datax.py
v_path_json=/home/admin/ob_ss/ob_to_ora_ss_tbcs5.json
v_path_log=/home/admin/ob_ss/log/

#從table_name.txt獲取表名、分區(qū)
for table_name in `cat $v_table_list`
Do
v_source_table_name1=`echo $table_name|awk -F ":" {print $1}`
v_source_table_partition=`echo $table_name|awk -F ":" {print $2}`
v_target_table_name=`echo $table_name|awk -F ":" {print $1}|awk -F "." {print $2}`
$v_exec_command --loglevel=info -p "
-DreadTb=${v_source_table_name1} 
-Dreadpartition=${v_source_table_partition} 
-DwriteTb=${v_target_table_name} 
" $v_path_json  >> "$v_path_log"$v_source_table_name1"_"$v_source_table_partition".log

三

DataX的并發(fā)參數(shù)

Json配置文件讀寫數(shù)據(jù)有兩種模式：

一種是table模式；
一種是querySql模式.

因為channel參數(shù)的實現(xiàn)是通過生成多個SQL語句實現(xiàn)的，所以 channel生效僅在能夠split出多個SQL語句的場景下，也就是table模式+spliPk下有用。

在table模式下， channel個數(shù)決定了reader和writer的個數(shù)上限，假設(shè)為m個：如果指定了splitPk字段，DataX會將mysql表中數(shù)據(jù)按照splitPk切分成n段，n大致為5倍的channel個數(shù)。

splitPk的字段限制了必須是整型或者字符串類型。由于DataX的實現(xiàn)方式是按照spliPk字段分段查詢數(shù)據(jù)庫表，那么spliPk字段的選取應(yīng)該盡可能的選擇分布均勻且有索引的字段，比如主鍵id、唯一鍵等字段。

DataX會啟動m個reader線程，消費DataX切分好的n個查詢sql語句（task），對應(yīng)的會有m個writer線程將查詢出來的數(shù)據(jù)寫入目標(biāo)數(shù)據(jù)源中，并行度為m（也就是配置的channel個數(shù)），如果不指定splitPk字段，DataX將不會進(jìn)行數(shù)據(jù)的切分，并行度直接退化成1。

需要指出的是，oceanbasev10reader，plitPk的字段限制了必須是整型。

四

遇到的問題

原因：TBCS.SYS_C0038005是全局索引，當(dāng)在json的preSql設(shè)置”alter table ** truncate partition **” 時導(dǎo)致索引失效。
解決辦法：rebuild重建索引，preSql設(shè)置” truncate table **”.

文章結(jié)語：dataX的總體使用相對比較簡單，作為數(shù)據(jù)同步框架，將不同數(shù)據(jù)源的同步抽象為從源頭數(shù)據(jù)源讀取數(shù)據(jù)的 Reader 插件，以及向目標(biāo)端寫入數(shù)據(jù)的 Writer 插件，只要有相應(yīng)的讀寫插件理論上 DataX 框架可以支持任意數(shù)據(jù)源類型的數(shù)據(jù)同步工作。

比較麻煩的是如何盡可能的提高同步的效率，雖然提供了 channel并發(fā)參數(shù)，但對于表本身的結(jié)構(gòu)和數(shù)據(jù)要求比較高，并不是所有的表都能滿足channel的條件。另外由于表結(jié)構(gòu)需要事先建立，對于批量的同步和表結(jié)構(gòu)變更的情況，支持性差。