DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，支持包括 MySQL、SQL Server、Oracle、PostgreSQL、HDFS、Hive、HBase、OTS、ODPS 等各种异构数据源之间高效的数据同步功能。DataX 采用了框架 + 插件 的模式。目前已开源，代码托管在 GitHub。

    DataX 数据同步效率较高，满足大多数场景下的异构数据库间的数据同步需求。同时其配置灵活，支持字段级别的配置，可以轻松应对因异构数据库迁移到 TiDB 而产生的一些改动。

    方案设计如图所示

    5.4 SqlServer 到 TiDB（DATAX） - 图1

    第一阶段：切换支持 TiDB 的应用上线之前，把 SQL Server 数据库中的全量数据用 DataX 同步到 TiDB 库中。为避免对线上业务产生影响，可以选择备份库，或者在业务低峰期操作。

    第二阶段：把全量同步改为增量同步，利用 UpdateTime 字段（或其它条件，根据实际业务灵活调整）作为同步 Where 条件，进行增量覆盖式同步。 t_sync_record 表中记录每张表上次增量任务的执行时间。

    第三阶段：支持 TiDB 的应用上线以后，增量同步切换读写数据源改为逆向增量同步，将新数据近实时地写回 SQL Server 数据库。一旦上线之后出现需要回退的情况，可随时切回 SQL Server，待修复之后再次上线。

    具体操作步骤：

    第一步：部署 DataX

    下载

    1. wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz

    解压

    1. tar -zxvf datax.tar.gz

    自检

    1. python {YOUR_DATAX_HOME}/bin/datax.py {YOUR_DATAX_HOME}/job/job.json

    第二步：编写 DataX 数据同步 Job（Json格式）

    （1）全量同步Job

    vi full.json

    1. {
    3.     "job": {
    5.         "setting": {
    7.             "speed": {
    9.                 #数据分片，分片数据可同时进行同步
    11.                 "channel": 128 
    13.             }
    15.         },
    17.         "content": [{
    19.              #SQL Server配置
    21.             "reader": {
    23.                 "name": "sqlserverreader",
    25.                 "parameter": {
    27.                     #数据库用户名
    29.                     "username": "${srcUserName}",
    31.                     #数据库密码
    33.                     "password": "${srcPassword}",
    35.                      #需要迁移的列名，* 表示全部列
    37.                     "column": ["*"]
    39.                     "connection": [{
    41.                         #需要迁移的表名
    43.                         "table": ["${tableName}"],
    45.                          数据库 jdbc 连接
    47.                         "jdbcUrl": ["${srcUrl}"]
    49.                     }]
    51.                 }
    53.         },
    55.         #TiDB配置
    57.         "writer": {
    59.                 "name": "tidbwriter",
    61.                 "parameter": {
    63.                     #数据库用户名
    65.                     "username": "${desUserName}",
    67.                      #数据库密码
    69.                     "password": "${desPassword}",
    71.                      #使用 Replace 语法
    73.                     "writeMode": "replace",
    75.                     #目标表列名，* 表示全部列
    77.                     "column": ["*"],
    79.                     "connection": [{
    81.                         #数据库 jdbc 连接
    83.                         "jdbcUrl": "${desUrl}",
    85.                         #目标表名
    87.                         "table": ["${tableName}"]
    89.                     }],
    91.                      #本次迁移开始前执行的sql-作数据迁移日志使用
    93.                     "preSql": [
    95.                        "replace into t_sync_record(table_name,start_date,end_date) values('@table',now(),null)"],
    97.                     #本次迁移完成后执行的 sql- 作数据迁移日志使用
    99.                     "postSql": [
    101.                        "update t_sync_record set end_date=now() where table_name='@table' " ]
    103.                 }
    105.             }
    107.         }]
    109.     }
    111. }

    （2）增量同步 Job

    vi increase.json

    1. {
    3.     "job": {
    5.         "setting": {
    7.             "speed": {
    9.                 #数据分片，分片数据可同时进行同步
    11.                 "channel": 128 
    13.             }
    15.         },
    17.         "content": [{
    19.              #SQL Server配置
    21.             "reader": {
    23.                 "name": "sqlserverreader",
    25.                 "parameter": {
    27.                     #数据库用户名
    29.                     "username": "${srcUserName}",
    31.                     #数据库密码
    33.                     "password": "${srcPassword}",
    35.                      #需要迁移的列名，* 表示全部列
    37.                     "column": ["*"],
    39.                     "connection": [{
    41.                         #需要迁移的表名
    43.                         "table": ["${tableName}"],
    45.                          数据库 jdbc 连接
    47.                         "jdbcUrl": ["${srcUrl}"]
    49.                     }],
    51.                     #抓取一个时间窗口的增量数据
    53.                     "where": "updateTime >= '${syncTime}'"
    55.                 }
    57.         },
    59.         #TiDB配置
    61.         "writer": {
    63.                 "name": "tidbwriter",
    65.                 "parameter": {
    67.                     #数据库用户名
    69.                     "username": "${desUserName}",
    71.                      #数据库密码
    73.                     "password": "${desPassword}",
    75.                      #使用 Replace 语法
    77.                     "writeMode": "replace",
    79.                     #目标表列名，* 表示全部列
    81.                     "column": ["*"],
    83.                     "connection": [{
    85.                         #数据库 jdbc 连接
    87.                         "jdbcUrl": "${desUrl}",
    89.                         #目标表名
    91.                         "table": ["${tableName}"]
    93.                     }],
    95.                      #本次迁移开始前执行的sql-作数据迁移日志使用
    97.                     "preSql": [
    99.                        "replace into t_sync_record(table_name,start_date,end_date) values('@table',now(),null)"],
    101.                     #本次迁移完成后执行的 sql- 作数据迁移日志使用
    103.                     "postSql": [
    105.                        "update t_sync_record set end_date=now() where table_name='@table' " ]
    107.                 }
    109.             }
    111.         }]
    113.     }
    115. }

    （3）编写运行DataX Job的Shell执行脚本

    vi datax_excute_job.sh

    1. #!/bin/bash
    3. source /etc/profile
    5. srcUrl="Reader Sql Server 地址"
    7. srcUserName="Sql Server 账号"
    9. srcPassword="Sql Server 密码"
    11. desUrl="Writer TiDB 地址"
    13. desUserName="TiDB 账号"
    15. desPassword="TiDB 密码"
    17. # 同步开始时间
    19. defaultsyncUpdateTime="2020-03-03 18:00:00.000"
    21. # 同步周期(秒)
    23. sleepTime="N"
    25. tableName="Table1,Table2,..."
    27. # 循环次数标识，-1为一直循环，其他按输入次数循环
    29. flg=-1
    31. while [ "$flg" -gt 0 -o "$flg" -eq -1 ]
    33. do
    35.         #更新时间设置为上次循序执行的时间
    37.         if [ "" = "$preRunTime" ]; then
    39.                 syncTime=$defaultsyncUpdateTime
    41.         else
    43.                 syncTime=$preRunTime
    45.         fi
    47.         #记录下本次循环执行时间，供下次循环使用
    49.         preRunTime=$(date -d +"%Y-%m-%d %T")".000";
    51.         echo $syncTime
    53.         echo $preRunTime
    55.         echo $flg
    57.         python {YOUR_DATAX_HOME}/bin/datax.py -p "-DsyncTime='${syncTime}' -DtableName='${tableName}' -DsrcUrl='${srcUrl}' -DsrcUserName='${srcUserName}' -DsrcPassword='${srcPassword}' -DdesUrl='${desUrl}' -DdesUserName='${desUserName}' -DdesPassword='${desPassword}'" {YOUR_DATAX_HOME}/job/increase.json
    59.         if [ -1  -lt  "$flg" ]; then
    61.                let "flg-=1"
    63.         fi
    65.         sleep $sleepTime
    67. done

    （4）执行 Shell 脚本

    1. chmod +x datax_excute_job.sh
    3. nohup ./datax_excute_job.sh > info.file 2>&1 &

    至此，核心脚本和操作都已完成，可通过修改参数配置，达到自己不同的需求，还可以配合数据对比服务，以期达到将应用程序从 SQL Server 顺利迁移到 TiDB 的目的。