使用 DataX 迁移 CSV 文件到 OceanBase 数据库

更新时间：2024-06-07 10:26:33

DataX 是阿里云 DataWorks 数据集成的开源版本，是阿里巴巴集团内被广泛使用的离线数据同步工具/平台。DataX 实现了包括 MySQL、Oracle、SQLserver、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、Hologres、DRDS 、OceanBase 等各种异构数据源之间高效的数据同步功能。

OceanBase 数据库社区版客户，可以在 DataX 开源网站内下载源码，自行编译。编译时，可根据需要剔除在 pom.xml 中不用的数据库插件，否则编译出来的包会非常大。

框架设计

datax

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

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

DataX 以任务的形式迁移数据。每个任务只处理一个表，每个任务有一个 json 格式的配置文件，配置文件里包含 reader 和 writer 两部分。reader 和 writer 分别对 DataX 支持的数据库读写插件，例如，将 MySQL 表数据迁移到 OceanBase 数据库操作时，需要从 MySQL 读取数据写入 OceanBase 数据库，因此使用的插件为 MySQL 数据库的 txtfilereader 插件和 OceanBase 数据库的 oceanbasev10writer 插件来搭配完成。这里介绍下 txtfilereader 和 oceanbasev10writer 插件。

`txtfilereader` 插件

txtfilereader 插件实现了从本地文件系统数据存储上读取数据的功能。在底层实现上，txtfilereader 获取本地文件数据，并转换为 DataX 传输协议传递给 Writer。

详细功能和参数说明请参考官方说明：txtfilereader 插件。

`oceanbasev10writer` 插件

oceanbasev10writer 插件实现了将数据写入到 OceanBase 数据库的目标表中的功能。在底层实现上，oceanbasev10writer 通过 Java 客户端（底层 MySQL JDBC 或 OceanBase Client）以 obproxy 远程连接 OceanBase 数据库，并执行相应的 insert sql 语句来将数据写入到 OceanBase 数据库， OceanBase 数据库内部会分批次提交入库。

实现原理方面，oceanbasev10writer 通过 DataX 框架获取 Reader 生成的协议数据，生成 insert 语句。在写入数据时，如果出现主键或唯一键冲突，OceanBase 数据库的 MySQL 租户可以通过 replace 模式来更新表中的所有字段；OceanBase 数据库的 Oracle 租户当前只能使用 Insert 方式。出于性能考虑，写入采用 batch 方式批量写，当行数累计到预定阈值时，才发起写入请求。

DataX 配置文件

配置文件示例：

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "streamreader",
          "parameter": {
            "sliceRecordCount": 10,
            "column": [
              {
                "type": "long",
                "value": "10"
              },
              {
                "type": "string",
                "value": "hello，你好，世界-DataX"
              }
            ]
          }
        },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "encoding": "UTF-8",
            "print": true
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 2
       }
    }
  }
}

注意

datax 仅迁移表数据，需要提前在目标端创建好对应的表对象结构。

将 json 配置文件放到 DataX 的目录 job 下，或者自定义路径。执行方法如下：

$bin/datax.py job/stream2stream.json

输出信息：

<.....>

2021-08-26 11:06:09.217 [job-0] INFO  JobContainer - PerfTrace not enable!
2021-08-26 11:06:09.218 [job-0] INFO  StandAloneJobContainerCommunicator - Total 20 records, 380 bytes | Speed 38B/s, 2 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.000s | Percentage 100.00%
2021-08-26 11:06:09.223 [job-0] INFO  JobContainer -
任务启动时刻                    : 2021-08-26 11:05:59
任务结束时刻                    : 2021-08-26 11:06:09
任务总计耗时                    :                 10s
任务平均流量                    :               38B/s
记录写入速度                    :              2rec/s
读出记录总数                    :                  20
读写失败总数                    :                   0

DataX 任务执行结束会有个简单的任务报告，包含上述输出的平均流量、写入速度和读写失败总数等。

DataX 的 job 的参数 settings 可以指定速度参数和错误记录容忍度等。

"setting": {
            "speed": {
                "channel": 10 
            },
            "errorLimit": {
                "record": 10,
                "percentage": 0.1
            }
        }

参数说明：

errorLimit 表示报错记录数的容忍度，超出这个限制后任务就中断退出。
channel 是并发数，理论上并发越大，迁移性能越好。但实际操作中也要考虑源端的读压力、网络传输性能以及目标端写入性能。

环境准备

下载 tar 包地址：http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz

解压安装文件：

tar zxvf datax.tar.gz
cd datax

目录如下：

$tree -L 1 --filelimit 30
.
├── bin
├── conf
├── job
├── lib
├── log
├── log_perf
├── plugin
├── script
└── tmp

安装文件中有以下几个目录需要了解：

目录名	说明
bin	可执行文件目录。该目录下的 datax.py 为 DataX 任务的启动脚本
conf	日志文件配置目录。该目录下存放 datax 与任务无关的配置文件
lib	运行时期依赖的包。该目录存放 DataX 运行所需要的全局 jar 文件
job	该目录下有一个用于测试验证 datax 安装的任务配置文件
log	日志文件目录。该目录下存放 datax 任务运行的日志；datax 运行时，默认会将日志输出到标准输出，同时写入到 log 目录下
plugin	插件文件目录。该目录下保存 DataX 支持的各种数据源插件

使用 DataX 迁移 CSV 文件到 OceanBase 示例

示例：将源端导出的 CSV 文件复制到目标端的 DataX 服务器上，然后导入到目标端 OceanBase 数据库中。

myjob.json 配置文件如下：

{
    "job": {
        "setting": {
            "speed": {
                "channel": 4
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.1
            }
        },
        "content": [
            {
                "reader": {
                    "name": "txtfilereader",
                    "parameter": {
                        "path": ["/tmp/tpcc/bmsql_oorder"],
                        "fileName": "bmsql_oorder",
                        "encoding": "UTF-8",
                        "column": ["*"],
                        "dateFormat": "yyyy-MM-dd hh:mm:ss" ,
                        "nullFormat": "\\N" ,
                        "fieldDelimiter": ","
                    }
                },
                "writer": {
                    "name": "oceanbasev10writer",
                    "parameter": {
                        "obWriteMode": "insert",
                        "column": ["*"],
                        "preSql": [
                            "truncate table bmsql_oorder"
                        ],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:oceanbase://127.0.0.1:2883/tpcc?",
                                "table": [
                                    "bmsql_oorder"
                                ]
                            }
                        ],
                        "username": "tpcc",
                        "password":"********",
                        "writerThreadCount":10,
                        "batchSize": 1000,
                        "memstoreThreshold": "0.9"
                    }
                }
            }
        ]
    }
}

参数说明

参数	描述
name	描述的是连接数据库的 reader 或 writer 对应的数据库插件的名称。其中 MySQL 的 reader 插件为：mysqlreader，OceanBase 的 writer 插件为 oceanbasev10writer。具体 reader 和 writer 的插件可以参考 datax 的文档：DataX 数据源指南。
jdbcUrl	描述的是到连接的数据库的 JDBC 信息，使用 JSON 的数组描述，并支持一个库填写多个连接地址。您在 JSON 数组中填写一个 JDBC 连接即可。jdbcUrl 按照 MySQL 官方规范，并可以填写连接附件控制信息。具体请参见 MySQL 官方文档。注意 jdbcUrl 必须包含在 connection 配置单元中。 OceanBase 数据库需要通过 obproxy 进行连接，端口默认 2883。 writer 端的 jdbcUrl，无需在连接串两端加`[]`，reader 端的 jdbcUrl，必需在连接串两端加`[]` 必选：是默认值：无
username	数据源的用户名必选：是默认值：无
password	数据源指定用户名的密码必选：是默认值：无
table	所选取的需要同步的表。使用 JSON 的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需确保多张表是同一 schema 结构，MySQLReader 不予检查表是否同一逻辑表。注意 table 必须包含在 connection 配置单元中。必选：是默认值：无
column	所配置的表中需要同步的列名集合，使用 JSON 的数组描述字段信息。columns 不建议配置为 `['*']` ，因为表结构发生变化时该配置也会发生变化。建议的配置方式是指定具体的列名。支持列裁剪，即列可以挑选部分列进行导出。支持列换序，即列可以不按照表 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 均视作同步全量数据。必选：否默认值：无

配置 job 文件后，执行该 job。命令如下：

python datax.py ../job/myjob.json