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

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

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

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

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

一

DataX

1.1 DataX簡(jiǎn)介

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

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

1.2 DataX 3.0 框架

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

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

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

二

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í)行語(yǔ)句為:

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

每個(gè)任務(wù)的參數(shù)文件是一個(gè) json 格式，主要由setting和一個(gè) reader 和一個(gè) 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: 出錯(cuò)記錄數(shù)超過(guò)record設(shè)置的條數(shù)時(shí)，任務(wù)標(biāo)記為失敗.
percentage: 當(dāng)出錯(cuò)記錄數(shù)超過(guò)percentage百分?jǐn)?shù)時(shí)，任務(wù)標(biāo)記為失敗.
bytes表示每秒字節(jié)數(shù)，默認(rèn)為0（不限速）。

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

單個(gè)表同步：

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配置文件讀寫(xiě)數(shù)據(jù)有兩種模式：

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

因?yàn)閏hannel參數(shù)的實(shí)現(xiàn)是通過(guò)生成多個(gè)SQL語(yǔ)句實(shí)現(xiàn)的，所以 channel生效僅在能夠split出多個(gè)SQL語(yǔ)句的場(chǎng)景下，也就是table模式+spliPk下有用。

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

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

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

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

四

遇到的問(wèn)題

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

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

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