从 ClickHouse 迁移到 DolphinDB

Centos 系统：

# 安装 unixODBC 库
yum install unixODBC unixODBC-devel

如果是 Ubuntu 系统，使用以下命令安装：

# 安装 unixODBC 库
apt-get install unixodbc unixodbc-dev

第一步，下载 clickhouse-odbc 并解压至自定义目录中，假设安装目录为 <savedir>。

mkdir <savedir>
cd <savedir>
wget https://github.com/ClickHouse/clickhouse-odbc/releases/download/v1.2.1.20220905/clickhouse-odbc-linux.zip
unzip clickhouse-odbc-linux.zip

第二步，在 /etc/odbcinst.ini 文件中，添加以下内容：

注意：需要替换 <savedir> 为实际的安装目录。

[ODBC Drivers] 
ClickHouse ODBC Driver (ANSI)    = Installed 
ClickHouse ODBC Driver (Unicode) = Installed 

[ClickHouse ODBC Driver (ANSI)] 
Description = ODBC Driver (ANSI) for ClickHouse 
Driver      = <savedir>/clickhouse-odbc-1.2.1-Linux/lib64/libclickhouseodbc.so 
Setup       = <savedir>/clickhouse-odbc-1.2.1-Linux/lib64/libclickhouseodbc.so 
UsageCount  = 1 

[ClickHouse ODBC Driver (Unicode)] 
Description = ODBC Driver (Unicode) for ClickHouse 
Driver      =<savedir>/clickhouse-odbc-1.2.1-Linux/lib64/libclickhouseodbcw.so 
Setup       =<savedir>/clickhouse-odbc-1.2.1-Linux/lib64/libclickhouseodbcw.so 
UsageCount  = 1 

第三步，在 /etc/odbc.ini 文件中，添加以下内容：

注意：若不存在该文件可自行创建。

[ODBC Data Sources] 
ClickHouse DSN (ANSI)=ClickHouse ODBC Driver (ANSI) 
ClickHouse DSN (Unicode)=ClickHouse ODBC Driver (Unicode) 

[ClickHouseAnsi] 
Driver=ClickHouse ODBC Driver (ANSI) 
Description=DSN (localhost) for ClickHouse ODBC Driver (ANSI) 
Server = localhost
Database = default
UID = default
PWD = 123456
Port = 8123
Proto = http

[ClickHouseUni] 
Driver=ClickHouse ODBC Driver (Unicode) 
Description=DSN (localhost) for ClickHouse ODBC Driver (Unicode) 
Server = localhost
Database = default
UID = default
PWD = 123456
Port = 8123
Proto = http

1. 运行以下命令加载 ODBC 插件（ ServerPath 请自行修改）

   loadPlugin("ServerPath/plugins/odbc/PluginODBC.txt")

2. 运行以下命令建立与 ClickHouse 的连接（ Dsn 的名称请自行修改）

   conn = odbc::connect("Dsn=ClickHouseAnsi", `ClickHouse)

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

   def syncData(conn, dbName, tbName, dt){
       sql = "select SecurityID, TradeTime, TradePrice, TradeQty, TradeAmount, BuyNo, SellNo, ChannelNo, TradeIndex, TradeBSFlag, BizIndex from migrate.ticksh"
       if(!isNull(dt)) {
           sql = sql + " WHERE toDate(TradeTime) = '"+temporalFormat(dt,'yyyy-MM-dd')+"'"
       }
       odbc::query(conn,sql, loadTable(dbName,tbName), 100000)
   }
   dbName="dfs://TSDB_tick"
   tbName="tick"
   syncData(conn, dbName, tbName, NULL)

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

4. 同步增量数据

   def scheduleLoad(conn,dbName,tbName){
       sqlQuery = "select SecurityID, TradeTime, TradePrice, TradeQty, TradeAmount, BuyNo, SellNo, ChannelNo, TradeIndex, TradeBSFlag, BizIndex from migrate.ticksh"
       sql = sqlQuery + " WHERE toDate(TradeTime) = '"+temporalFormat(today()-1,'yyyy-MM-dd')+"'"
       odbc::query(conn,sql, loadTable(dbName,tbName), 100000)
   }
   scheduleJob(jobId=`test, jobDesc="test",jobFunc=scheduleLoad{conn,dbName,tbName},scheduleTime=00:05m,startDate=2023.11.03, endDate=2024.11.03, frequency='D')

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

注意：2023 年 9 月后发布的 datax 中才有 clickhousereader 插件，如已安装老版本 datax，则只需下载安装包中的 clickhousereader 复制到 DataX/plugin/reader 目录下，即可使用。

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

1. 配置 json 文件

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

   {
           "job": {
                   "content": [{
                           "writer": {
                                   "parameter": {
                                           "dbPath": "dfs://TSDB_tick",
                                           "userId": "admin",
                                           "tableName": "tick",
                                           "host": "127.0.0.1",
                                           "pwd": "123456",
                                           "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": 10001
                                   },
                                   "name": "dolphindbwriter"
                           },
                           "reader": {
                                   "parameter": {
                                           "username": "default",
                                           "column": ["SecurityID", "toString(TradeTime)", "TradePrice", "TradeQty", "TradeAmount", "BuyNo", "SellNo", "ChannelNo", "TradeIndex", "TradeBSFlag", "BizIndex"],
                                           "connection": [{
                                                   "table": ["ticksh"],
                                                   "jdbcUrl": ["jdbc:clickhouse://127.0.0.1:8123/migrate"]
                                           }],
                                           "password": "123456",
                                           "where": ""
                                   },
                                   "name": "clickhousereader"
                           }
                   }],
                   "setting": {
                           "speed": {
                                   "channel": 1
                           }
                   }
           }
   }

   注：当前 clickhousereader 无法识别 DateTime64 类型，故需转为字符串（"toString(TradeTime)"）取数。

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

   python datax.py ../../datax-writer-master/ddb_script/ClickHouseToDDB.json

3. 查看 DataX 同步结果

   任务启动时刻                    : 2023-11-03 17:11:26
   任务结束时刻                    : 2023-11-03 17:14:57
   任务总计耗时                    :                210s
   任务平均流量                    :            8.93MB/s
   记录写入速度                    :         129580rec/s
   读出记录总数                    :            27211975
   读写失败总数                    :                   0

4. 同步增量数据

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

   "reader": {
       "parameter": {
           "username": "default",
           "column": ["SecurityID", "toString(TradeTime)", "TradePrice", "TradeQty", "TradeAmount", "BuyNo", "SellNo", "ChannelNo", "TradeIndex", "TradeBSFlag", "BizIndex"],
           "connection": [{
               "table": ["ticksh"],
               "jdbcUrl": ["jdbc:clickhouse://127.0.0.1:8123/migrate"]
           }],
           "password": "123456",
           "where": "toDate(TradeTime) = date_sub(day,1,today())"
       },
       "name": "clickhousereader",
   }