从 Postgre/Greenplum 迁移到 DolphinDB

本例中部署 DolphinDB 的服务器操作系统为 Centos。

step1：在安装 PostgreSQL 的 ODBC 驱动前，需先安装 unixODBC 库，使用以下命令安装

Centos 系统：

# 安装 unixODBC 库
yum install unixODBC unixODBC-devel

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

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

注意：安装完成后可用odbcinst -j命令查看安装配置文件所在的位置，有两个比较常用的配置，一个是 ODBC 驱动配置，默认在/etc/odbcinst.ini，另一个是系统数据源配置，默认在/etc/odbc.ini。

step2：下载 PostgreSQL 的 ODBC 驱动并安装

以 Centos 系统为例，可以通过以下方式直接在 Linux 系统中直接安装 Postgresql-ODBC

yum install -y postgresql-odbc.x86_64

step3：配置 ODBC 配置文件

1）odbcinst.ini 文件用于设置 ODBC 的某个 Driver 要用到的 ODBC 驱动库的文件路径。 配置 odbcinst.ini 文件的内容（如果配置文件不存在，需手工创建） :

[PostgreSQL]
Description     = ODBC for PostgreSQL
Driver          = /usr/lib/psqlodbcw.so
Setup           = /usr/lib/libodbcpsqlS.so
Driver64        = /usr/lib64/psqlodbcw.so
Setup64         = /usr/lib64/libodbcpsqlS.so
FileUsage       = 1

2）/etc/odbc.ini 文件用于设置 ODBC 中所用到的 Driver 、所要使用的数据库等配置，更多配置项可参考 ODBC 连接字符串配置。其中的 Driver 为 /etc/odbcinst.ini 文件中配置的第一行方括号中的内容。添加内容如下（如果配置文件不存在，需手工创建）：

[postgresql] 					# ODBC 数据源名称
Description = PostgresSQLODBC	# ODBC 的描述
Driver = PostgreSQL				# 驱动名称
Database = postgres				# 数据库名称
Servername = 127.0.0.1			# 布置了 Postgresql 数据库的服务器 IP 地址
UserName = postgres				# 数据库相关的用户名
Password = postgres				# 数据库密码
Port = 5432					    # 布置了 Postgresql 数据库的服务器的端口号
ReadOnly = 0					# 关闭只读特性
ConnSettings = set client_encoding to UTF8	# 客户端编码

step4：测试 ODBC 连接

使用 ODBC 登录 Postgresql 数据库，测试连接及预期结果如下：

isql postgresql postgres postgres  # 后两位分别是用户名和密码

/**********output********/
+---------------------------------------+
| Connected!                            |
|                                       |
| sql-statement                         |
| help [tablename]                      |
| quit                                  |
|                                       |
+---------------------------------------+

登录成功如下所示，可输入 SQL 操作数据库

SQL>

注意：如有其他常见问题，可参考 ODBC插件常见问题 中第5章注意事项和常见问题。

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

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

step2：运行以下命令建立与 Oracle 的连接 （第一个参数为 ODBC 连接字符串 connStr）

connstr 请参阅连接字符串参考自行修改。

conn = odbc::connect("Driver={PostgreSQL};Server=*;Port=5432;Database=postgres;Uid=postgres;Pwd=postgres;", `PostgreSQL)

step3：运行以下命令开始同步数据

def transForm(mutable msg){
	msg.replaceColumn!(`TradeQty, int(msg[`TradeQty]))
	msg.replaceColumn!(`BuyNo, int(msg[`BuyNo]))
	msg.replaceColumn!(`SellNo, int(msg[`SellNo]))
	msg.replaceColumn!(`ChannelNo, int(msg[`ChannelNo]))
	msg.replaceColumn!(`TradeIndex, int(msg[`TradeIndex]))
	msg.replaceColumn!(`BizIndex, int(msg[`BizIndex]))
	return msg
}

def syncData(conn, dbName, tbName, dt){
	sql = "select SecurityID, TradeTime, TradePrice, TradeQty, TradeAmount, BuyNo, SellNo, ChannelNo, TradeIndex, TradeBSFlag, BizIndex from ticksh"
	if(!isNull(dt)) {
		sql = sql + " where to_date(TradeTime,'yyyy-MM-dd hh24:mi:ss') = dt"
	}
    odbc::query(conn,sql, loadTable(dbName,tbName), 100000, transForm)
}

dbName="dfs://TSDB_tick"
tbName="tick"
syncData(conn, dbName, tbName, NULL)

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

step4：后台多任务同步数据

本教程中 DolphinDB 的数据库按天分区，如果需要同步多天的数据，可以提交多个任务到后台：

for(dt in 2021.01.04..2021.01.05){
	submitJob(`syncPostgreTick, `syncPostgreTick, syncData, conn, dbName, tbName, dt)
}
// 查看后台任务
select * from getRecentJobs() where jobDesc = `syncPostgreTick

从 DataX 下载地址 下载 DataX 压缩包后，解压至自定义目录。执行 datax 自检如下：

cd datax/bin
python datax.py /opt/datax/job/job.json

可能出现报错，报错为 [您提供的配置文件存在错误信息... plugin.json ] 不存在：

该问题是因为在 reader 和 writer 目录下存在临时文件，影响了 datax 的运行，解决方法如下：

# 这里的/datax/plugin/writer/应修改为实际目录
find /datax/plugin/reader/ -type f -name "._*er" | xargs rm -rf
find /datax/plugin/writer/ -type f -name "._*er" | xargs rm -rf

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

step1：配置 json 文件

配置文件 pgddb.json 的具体内容如下，并将 json 文件置于自定义目录下，本教程中放置于 datax/job 目录下。

{
    "job": {
            "content": [{
                    "writer": {
                            "name": "dolphindbwriter",
                            "parameter": {
                                    "userId": "admin",
                                    "pwd": "123456",
                                    "host": "10.0.0.80",
                                    "port": 8848,
                                    "dbPath": "dfs://TSDB_tick",
                                    "tableName": "Tick",
                                    "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"
                                        }
                                    ]                            
                            }
                    },
                    "reader": {
                            "name": "postgresqlreader",
                            "parameter": {
                                    "username": "postgres",
                                    "column": ["SecurityID", "TradeTime", "TradePrice", "TradeQty", "TradeAmount", "BuyNo", "SellNo", "ChannelNo", "TradeIndex", "TradeBSFlag", "BizIndex"],
                                    "connection": [{
                                            "table": ["ticksh"],
                                            "jdbcUrl": ["jdbc:postgresql:postgres"]
                                    }],
                                    "password": "postgres",
                                    "where": ""
                            }                            
                    }
            }],
            "setting": {
                    "speed": {
                            "channel": 1
                    }
            }
    }
}

step2：Linux 终端中执行以下命令以执行 DataX 任务

cd ./datax
python bin/datax.py --jvm=-Xmx8g job/ddbtopg.json

step3：查看 DataX 同步结果

任务启动时刻                    : 2023-08-29 14:19:53
任务结束时刻                    : 2023-08-29 14:26:33
任务总计耗时                    :                400s
任务平均流量                    :            4.08MB/s
记录写入速度                    :          68029rec/s
读出记录总数                    :            27211975
读写失败总数                    :                   0