使用 DataX 迁移

使用 DataX 迁移

更新时间：2023-08-03 17:20:32

DataX 是阿里巴巴集团内部广泛使用的离线数据同步工具/平台，它支持 MySQL、Oracle、HDFS、Hive、OceanBase、HBase、OTS、ODPS 等各种异构数据源之间高效的数据同步功能。基于 DataX 的同步机制，可以通过 OceanBase 数据库的 Reader 和 Writer 插件实现 OceanBase 数据库跨数据库、集群和异构数据库的数据迁移。

框架设计

datax

DataX 作为离线数据同步框架，采用"Framework + Plugin" 模式构建。将数据源读取和写入抽象为Reader/Writer 插件，纳入到整个同步框架中。

Reader 作为数据采集模块，负责采集数据源的数据，将数据发送给 Framework。
Writer 作为数据写入模块，负责不断向 Framework 获取数据，并将数据写入到目的端。
Framework 用于连接 Reader 和 Writer，作为两者的数据传输通道，并处理缓冲、流控、并发、数据转换等核心技术问题。

使用示例

使用 DataX 迁移 MySQL 数据到 OceanBase

将 MySQL 数据迁移到 OceanBase ，如果源端和目标端不能同时跟 DataX 服务器网络联通，那么可以通过 CSV 文件中转。如果源端数据库和目标端数据库能同时跟 DataX 所在服务器联通，则可以使用 DataX 直接将数据从源端迁移到目标端。配置文件如下：

{
    "job": {
        "setting": {
            "speed": {
                "channel": 4 
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.1
            }
        },
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "tpcc",
                        "password": "********",
                        "column": [
                            "*"
                        ],
                        "connection": [
                            {
                                "table": [
                                    "bmsql_oorder"
                                ],
                                "jdbcUrl": ["jdbc:mysql://127.0.0.1:3306/tpccdb?useUnicode=true&characterEncoding=utf8"]
                            }
                        ]
                    }
                },

                "writer": {
                    "name": "oceanbasev10writer",
                    "parameter": {
                        "obWriteMode": "insert",
                        "column": [
                            "*"
                        ],
                        "preSql": [
                            "truncate table bmsql_oorder"
                        ],
                        "connection": [
                            {
                                "jdbcUrl": "||_dsc_ob10_dsc_||obdemo:oboracle||_dsc_ob10_dsc_||jdbc:oceanbase://127.0.0.1:2883/tpcc?useLocalSessionState=true&allowBatch=true&allowMultiQueries=true&rewriteBatchedStatements=true",
                                "table": [
                                    "bmsql_oorder"
                                ]
                            }
                        ],
                        "username": "tpcc",
                        "password":"********",
                        "writerThreadCount":10,
                        "batchSize": 1000,
                        "memstoreThreshold": "0.9"
                    }
                }
            }
        ]
    }
}

使用 DataX 迁移 OceanBase 数据到 MySQL/Oracle

OceanBase 数据同步到 MySQL

配置文件如下：

{
    "job": {
        "setting": {
            "speed": {
                "channel": 16 
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.1
            }
        },
        "content": [
            {
                "reader": {
                    "name": "oceanbasev10reader",
                    "parameter": {
                        "where": "",
                        "readBatchSize": 10000,
                        "column": [
                            "*"
                        ],
                        "connection": [
                            {
                                "jdbcUrl": ["||_dsc_ob10_dsc_||obdemo:oboracle||_dsc_ob10_dsc_||jdbc:oceanbase://127.0.0.1:2883/tpcc"],
                                "table": [
                                    "bmsql_oorder"
                                ]
                            }
                        ],
                        "username": "tpcc",
                        "password":"********"
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                        "writeMode": "replace",
                        "username": "tpcc",
                        "password": "******",
                        "column": [
                            "*"
                        ],
                        "session": [
                         "set session sql_mode='ANSI'"
                        ],
                        "preSql": [
                            "truncate table bmsql_oorder"
                        ],
                        "batchSize": 512,
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:mysql://127.0.0.1:3306/tpccdb?useUnicode=true&characterEncoding=utf8",
                                "table": [
                                    "bmsql_oorder"
                                ]
                            }
                        ]
                    }
                }
            }    
        ]
    }
}

参数说明

参数	描述	是否必选	默认值
jdbcUrl	描述的是到对端数据库的 JDBC 连接信息，使用 JSON 的数组描述，并支持一个库填写多个连接地址。您在 JSON 数组中填写一个 JDBC 连接即可。jdbcUrl 按照 MySQL 官方规范，并可以填写连接附件控制信息。具体请参见 MySQL 官方文档。注意 jdbcUrl 必须包含在 connection 配置单元中。	是	无
username	数据源的用户名	是	无
password	数据源指定用户名的密码	是	无
table	所选取的需要同步的表。使用 JSON 的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需确保多张表是同一 schema 结构，MysqlReader 不予检查表是否同一逻辑表。注意 table 必须包含在 connection 配置单元中。	是	无
column	所配置的表中需要同步的列名集合，使用 JSON 的数组描述字段信息。您可以使用 `` 代表默认使用所有列配置，例如 `['']`。支持列裁剪，即列可以挑选部分列进行导出。支持列换序，即列可以不按照表 schema 信息进行导出。支持常量配置，用户需要按照 Mysql SQL语法格式： ["id", "`table`", "1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"] 。 `id` 为普通列名 `table` 为包含保留字的列名 `1` 为整形数字常量 `bazhen.csy` 为字符串常量 `null` 为空值 `to_char(a + 1)` 为表达式 `2.3` 为浮点数 `true` 为布尔值	是	无
where	筛选条件，MysqlReader 根据指定的column、table、where 条件拼接 SQL，并根据这个 SQL 进行数据抽取。在实际业务场景中，往往会选择当天的数据进行同步，可以将where 条件指定为 `gmt_create > $bizdate` 。注意不可以将 where 条件指定为 `limit 10`，`limit` 不是 SQL 的合法 where 子句。 where 条件可以有序地进行业务增量同步。如果不填写 where 语句，包括不提供 where 的 `key` 或者 `value`，DataX 均视作同步全量数据。	否	无

框架设计

使用示例

使用 DataX 迁移 MySQL 数据到 OceanBase

使用 DataX 迁移 OceanBase 数据到 MySQL/Oracle

参数说明

更多信息