首批通过分布式安全可靠测评，为关键业务系统打造
OceanBase 数据库 DataX 使用示例
更新时间：2023-07-18 17:25:59
分享
OceanBase 数据库使用插件 oceanbasev10reader 和 oceanbasev10writer 来读写。
oceanbasev10reader 配置示例
"reader":{
"name":"oceanbasev10reader",
"parameter":{
"where":"",
"timeout":10000,
"readBatchSize":100000,
"readByPartition":"true",
"column": [
"列名1","列名2"
],
"connection":[
{
"jdbcUrl":[
"||_dsc_ob10_dsc_||集群名:租户名||_dsc_ob10_dsc_||jdbc:oceanbase://连接IP:连接端口/模式名或数据库名"
],
"table":[
"表名"
]
}
],
"username":"租户内用户名",
"password":"******"
}
}
示例：
从 OceanBase 数据库中将表 ware 导出到 CSV 文件。
[admin@*** /home/admin/datax3]
$cat job/ob_tpcc_ware_2_csv.json
{
"job":{
"setting":{
"speed":{
"channel":10
},
"errorLimit":{
"record":0, "percentage": 0.02
}
},
"content":[
{
"reader":{
"name":"oceanbasev10reader",
"parameter":{
"where":"",
"timeout":10000,
"readBatchSize":100000,
"readByPartition":"true",
"column": [
"W_ID","W_YTD","W_TAX","W_NAME","W_STREET_1","W_STREET_2","W_CITY","W_STATE","W_ZIP"
],
"connection":[
{
"jdbcUrl":["||_dsc_ob10_dsc_||obdemo:obbmsql||_dsc_ob10_dsc_||jdbc:oceanbase://127.1:2883/tpcc"],
"table":["ware"]
}
],
"username":"tpcc",
"password":"******"
}
},
"writer":{
"name":"txtfilewriter",
"parameter":{
"path":"/home/admin/csvdata/",
"fileName":"ware",
"writeMode":"truncate",
"dateFormat":"yyyy-MM-dd",
"charset":"UTF-8",
"nullFormat":"",
"fileDelimiter":"||"
}
}
}
]
}
}
[admin@*** /home/admin/datax3]
$bin/datax.py job/ob_tpcc_ware_2_csv.json
返回结果：
oceanbasev10writer 配置示例
使用 DataX 向 OceanBase 数据库里写入时，要避免写入速度过快导致 OceanBase 数据库的增量内存耗尽。
说明
建议 DataX 配置文件里针对写入做一个写入限速设置，关键字是 memstoreThreshold。
"writer": {
"name": "oceanbasev10writer",
"parameter": {
"username": "租户内的用户名",
"password": "******",
"writeMode": "insert",
"column": [
"列名1","列名2"
],
"preSql": [
""
],
"connection": [
{
"jdbcUrl": "||_dsc_ob10_dsc_||集群名:租户名||_dsc_ob10_dsc_||jdbc:oceanbase://连接IP:连接端口(默认2883)/模式名或数据库名",
"table": [
"表名"
]
}
],
"batchSize": 1024,
"memstoreThreshold":"0.9"
}
}
示例：
导入 CSV 文件到 OceanBase 数据库中的 ware 表。
[admin@*** /home/admin/datax3]
$cat job/csv_2_ob_tpcc_ware2.json
{
"job":{
"setting":{
"speed":{
"channel":32
},
"errorLimit":{
"record":0, "percentage": 0.02
}
},
"content":[
{
"reader":{
"name":"txtfilereader",
"parameter":{
"path":["/home/admin/csvdata/ware*"],
"encoding":"UTF-8",
"column":[
{ "index":0, "type":"long" }
,{ "index":1, "type":"long" }
,{ "index":2, "type":"long" }
,{ "index":3, "type":"string" }
,{ "index":4, "type":"string" }
,{ "index":5, "type":"string" }
,{ "index":6, "type":"string" }
,{ "index":7, "type":"string" }
,{ "index":8, "type":"string" }
],
"fieldDelimiter":",",
"fileFormat":"text"
}
},
"writer":{
"name":"oceanbasev10writer",
"parameter":{
"writeMode":"insert",
"column":[
"W_ID","W_YTD","W_TAX","W_NAME","W_STREET_1","W_STREET_2","W_CITY","W_STATE","W_ZIP"
],
"connection":[
{
"jdbcUrl":"||_dsc_ob10_dsc_||obdemo:obbmsql||_dsc_ob10_dsc_||jdbc:oceanbase://127.1:2883/tpcc",
"table":["WARE2"]
}
],
"username":"tpcc",
"password":"******",
"batchSize":256,
"memstoreThreshold":"0.9"
}
}
}
]
}
}
[admin@*** /home/admin/datax3]
$bin/datax.py job/csv_2_ob_tpcc_ware2.json
返回结果：
参数说明
| 参数 | 是否必填 | 描述 |
|---|---|---|
| jdbcUrl | 是 | 对端数据库的 JDBC 连接信息，使用 JSON 的数组描述，并支持一个库填写多个连接地址。您在 JSON 数组中填写一个 JDBC 连接即可。jdbcUrl 的格式，请参考各数据库官方文档。 注意 jdbcUrl 必须包含在 connection 配置单元中。 |
| username | 是 | 数据源的用户名。 |
| password | 是 | 数据源指定用户名的密码。 |
| table | 是 | 所选取的需要同步的表。使用 JSON 的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需确保多张表是同一 schema 结构。 注意 table 必须包含在 connection 配置单元中。 |
| column | 是 | 所配置的表中需要同步的列名集合，使用 JSON 的数组描述字段信息。您可以使用 '*' 代表默认使用所有列配置。 警告 不建议将 column 配置为 '*'，在 schema 发生变更时可能会导致任务失败。 支持列裁剪，即列可以挑选部分列进行导出。 支持列换序，即列可以不按照表 schema 信息进行导出。 支持常量配置，用户需要按照Mysql SQL语法格式： ["id", "`table`", "1", "'abc.xyz'", "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 数据库
将 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": [
"列名1","列名2"
],
"connection": [
{
"table": [
"bmsql_oorder"
],
"jdbcUrl": ["jdbc:mysql://127.0.0.1:3306/tpccdb?useUnicode=true&characterEncoding=utf8"]
}
]
}
},
"writer": {
"name": "oceanbasev10writer",
"parameter": {
"obWriteMode": "insert",
"column": [
"列名1","列名2"
],
"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": [
"列名1","列名2"
],
"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": [
"列名1","列名2"
],
"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"
]
}
]
}
}
}
]
}
}
OceanBase 数据同步到 Oracle
配置文件如下：
{
"job": {
"setting": {
"speed": {
"channel": 16
},
"errorLimit": {
"record": 0,
"percentage": 0.1
}
},
"content": [
{
"reader": {
"name": "oceanbasev10reader",
"parameter": {
"where": "",
"readBatchSize": 10000,
"column": [
"列名1","列名2"
],
"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": "oraclewriter",
"parameter": {
"username": "tpcc",
"password": "********",
"column": [
"列名1","列名2"
],
"preSql": [
"truncate table bmsql_oorder"
],
"batchSize": 512,
"connection": [
{
"jdbcUrl": "jdbc:oracle:thin:@127.0.0.1:1521:helowin",
"table": [
"bmsql_oorder"
]
}
]
}
}
}
]
}
}
