从 OceanBase 迁移到 DolphinDB

参考： DolphinDB MySQL Plugin。

1. 运行以下命令加载 MySQL 插件

   loadPlugin("ServerPath/plugins/mysql/PluginMySQL.txt")

2. 运行以下命令建立与 OceanBase 的连接

   conn = mysql::connect(`127.0.0.1,2881,`root,`123456,`db1)

3. 运行以下命令开始同步数据

   mysql::loadEx(conn, database('dfs://TSDB_tick'), `tick, `TradeTime`SecurityID,"tick")

   数据共 27211975 条，同步数据耗时约52秒。

4. 同步增量数据

   如需实现增量同步，只需将 mysql::loadEX 中的源数据表名替换为查询语句, 并借助 DolphinDB 提供的 scheduleJob 函数设置定时任务，即可实现增量同步，示例如下，每天 00:05 同步前一天数据：

   def scheduleLoad(){
     sqlQuery = "select * from tick where date(TradeTime) =  '" +temporalFormat(today()-1, 'y-MM-dd') +"' ;"
     mysql::loadEx(conn, database('dfs://TSDB_tick'), `tick, `TradeTime`SecurityID,sqlQuery)
   }
   scheduleJob(jobId=`test, jobDesc="test",jobFunc=scheduleLoad,scheduleTime=00:05m,startDate=2023.04.04, endDate=2024.01.01, frequency='D')

   注意：为防止节点重启时定时任务解析失败，预先在配置文件里添加 preloadModules=plugins::mysql。

从 DataX 下载地址下载 DataX 压缩包后，解压至自定义目录。

将 DataX-DolphinDBWriter中源码的 ./dist/dolphindbwriter 目录下所有内容拷贝到 DataX/plugin/writer 目录下。

1. 根据实际环境配置json文件。

   配置文件 OceanBase_tick.json 的具体内容如下，并将 json 文件置于自定义目录下，本教程中方放置于 datax-writer-master/ddb_script/ 目录下。

   {
       "job": {
           "setting": {
               "speed": {
                   "channel":1
               }
           },
           "content": [
               {
                   "writer": {
                       "parameter": {
                           "dbPath": "dfs://TSDB_tick",
                           "tableName": "tick",
                           "userId": "admin",
                           "pwd": "123456",
                           "host": "127.0.0.1",
                           "batchSize": 200000,
                           "table": [
                               {
                                   "type": "DT_SYMBOL",
                                   "name": "SecurityID"
                               },
                               {
                                   "type": "DT_TIMESTAMP",
                                   "name": "TradeTime"
                               },
                               {
                                   "type": "DT_DOUBLE",
                                   "name": "TradePrice"
                               },
                               {
                                   "type": "DT_INT",
                                   "name": "TradeQty"
                               },
                               {
                                   "type": "DT_DOUBLE",
                                   "name": "TradeAmount"
                               },
                               {
                                   "type": "DT_INT",
                                   "name": "BuyNo"
                               },
                               {
                                   "type": "DT_INT",
                                   "name": "SellNo"
                               },
                               {
                                   "type": "DT_INT",
                                   "name": "TradeIndex"
                               },
                               {
                                   "type": "DT_INT",
                                   "name": "ChannelNo"
                               },
                               {
                                   "type": "DT_SYMBOL",
                                   "name": "TradeBSFlag"
                               },
                               {
                                   "type": "DT_INT",
                                   "name": "BizIndex"
                               }
                           ],
                           "port": 8800
                       },
                       "name": "dolphindbwriter"
                   },
                   "reader": {
                       "name": "oceanbasev10reader",
                       "parameter": {
                           "username": "root",
                           "password": "123456",
                           "batchSize":10000,
                           "column": [
                               "*"
                           ],
                           "connection": [
                               {
                                   "table": [
                                       "tick"
                                   ],
                                   "jdbcUrl": [
                                       "jdbc:oceanbase://127.0.0.1:2883/db1"
                                   ]
                               }
                           ]
                       }
                   }
               }
           ]
       }
   }
   

2. Linux 终端中执行以下命令以执行 DataX 任务。

   cd ./dataX/bin/
   python datax.py ../../datax-writer-master/ddb_script/ocean.json

3. 查看DataX同步结果。

   任务启动时刻                    : 2023-04-03 14:58:52
   任务结束时刻                    : 2023-04-03 15:00:52
   任务总计耗时                    :                120s
   任务平均流量                    :           12.32MB/s
   记录写入速度                    :         226766rec/s
   读出记录总数                    :            27211975
   读写失败总数                    :                   0

4. 同步增量数据。

   使用 DataX 同步增量数据，可在 ocean.json 的 ”reader“ 中增加 "where" 条件对数据日期进行筛选，如此每次执行同步任务时至同步 where 条件过滤后的数据，以同步前一天的数据为例，示例如下：

   "reader": {
       "name": "oceanbasev10reader",
       "parameter": {
           "username": "root",
           "password": "123456",
           "batchSize":10000,
           "column": [
               "*"
           ],
           "connection": [
               {
                   "table": [
                       "tick"
                   ],
                   "jdbcUrl": [
                       "jdbc:oceanbase://127.0.0.1:2883/db1"
                   ]
               }
           ],
           "where":"date(TradeTime) = (SELECT DATE_ADD(CURDATE(), INTERVAL -1 DAY))"
       }
   }

table 用于配置写入表的字段集合。内部结构为

 {"name": "columnName", "type": "DT_STRING", "isKeyField":true}

请注意此处列定义的顺序，需要与原表提取的列顺序完全一致。

• name：字段名称。
• isKeyField：是否唯一键值，可以允许组合唯一键。本属性用于数据更新场景，用于确认更新数据的主键，若无更新数据的场景，无需设置。
• type： 枚举值以及对应 DolphinDB 数据类型如下：