基于 DataX 的 DolphinDB 数据导入工具

DataX 是在阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL, Oracle, SqlServer, Postgre, HDFS, Hive, ADS, HBase, TableStore(OTS), MaxCompute(ODPS), DRDS 等各种异构数据源之间高效的数据同步功能，详情可查看 DataX 已支持的数据源。

DataX 是可扩展的数据同步框架，将不同数据源的同步抽象为从源头数据源读取数据的 Reader 插件，以及向目标端写入数据的 Writer 插件。理论上 DataX 框架可以支持任意数据源类型的数据同步工作。每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。

详细信息请参阅 DataX 指南，以下仅列出必要步骤。

• DolphinDB 写插件配置项

  配置项	是否必须	数据类型	默认值	描述
  host	是	STRING	无	Server Host
  port	是	INT	无	Server Port
  userId	是	STRING	无	DolphinDB 用户名导入分布式库时，必须是要有权限的用户才能操作
  pwd	是	STRING	无	DolphinDB 用户密码
  dbPath	是	STRING	无	需要写入的目标分布式库名称，比如"dfs://MYDB"。
  preSql	否	STRING	无	写入数据到目的表前，会先执行这里的 DolphinDB 脚本。
  postSql	否	STRING	无	写入数据到目的表后，会执行这里的 DolphinDB 脚本。
  column	否	STRING	无	目的表需要写入数据的字段。注意：不能与 table 字段同时使用。使用时请阅读本表下方的注意事项。
  tableName	是	STRING	无	目标数据表名称
  batchSize	否	INT	10000000	datax 每次写入 DolphinDB 的批次记录数。
  table	是			写入表的字段集合，具体参考后续table 项配置详解。
  saveFunctionName	否	STRING	无	自定义数据处理函数。若未指定此配置，插件在接收到 reader 的数据后，会将数据提交到 DolphinDB 并通过 tableInsert 函数写入指定库表；如果定义此参数，则会用指定函数替换 tableInsert 函数。
  saveFunctionDef	否	STRING	无	数据入库自定义函数。此函数指用DolphinDB 脚本来实现的数据入库过程。 此函数必须接受三个参数：dbName（分布式库路径）, tbName（数据表名）, data（从 datax 导入的数据，table 格式）。

  在使用 column 时，请注意：

  • 字段之间用英文逗号分隔，例如: "column": 。
  • 如果要依次写入全部列，使用表示，例如: "column": 。
  • 不可以为空 “column": ，或者”column": 。
  • 导入数据是按照原表的字段顺序，无法通过识别新表的字段名称而改变数据顺序。

• table 配置详解

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

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

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

  name：字段名称。

  isKeyField：是否唯一键值，可以允许组合唯一键。本属性用于数据更新场景，用于确认更新数据的主键，若无更新数据的场景，无需设置。

  type：枚举值以及对应 DolphinDB 数据类型如下。

  DolphinDB 类型	配置值
  DOUBLE	DT_DOUBLE
  FLOAT	DT_FLOAT
  BOOL	DT_BOOL
  DATE	DT_DATE
  MONTH	DT_MONTH
  DATETIME	DT_DATETIME
  TIME	DT_TIME
  SECOND	DT_SECOND
  TIMESTAMP	DT_TIMESTAMP
  NANOTIME	DT_NANOTIME
  NANOTIMETAMP	DT_NANOTIMETAMP
  INT	DT_INT
  LONG	DT_LONG
  UUID	DT_UUID
  SHORT	DT_SHORT
  STRING	DT_STRING
  SYMBOL	DT_SYMBOL

• 更新分区表和维度表脚本模板代码（供参考）

  • savePartitionedData

    def rowUpdate(dbName, tbName, data, t){
    	updateRowCount = exec count(*) from ej(t,data,['OBJECT_ID'])
    	if(updateRowCount<=0) return
    	dfsPath = "dfs://" + dbName
    	temp = select * from t
    	cp = t.schema().chunkPath.substr(strlen("/" + dbName))
    	update temp set EVENT_ID = data.EVENT_ID,S_INFO_WINDCODE = data.S_INFO_WINDCODE, S_ACTIVITIESTYPE=data.S_ACTIVITIESTYPE, S_SURVEYDATE=data.S_SURVEYDATE, S_SURVEYTIME=data.S_SURVEYTIME, ANN_DT = data.ANN_DT,OPDATE=data.OPDATE,OPMODE=data.OPMODE where OBJECT_ID in data.OBJECT_ID
    	dropPartition(database(dfsPath), cp, tbName)
    	loadTable(dfsPath, tbName).append!(temp)
    }
    
    def savePartitionedData(dbName, tbName, data){
    	dfsPath = "dfs://" + dbName
    	login("admin","123456")
    	t = loadTable(dfsPath, tbName)
    	ds1 = sqlDS(<select * from t>)
    	mr(ds1, rowUpdate{dbName, tbName, data})
    }

  • saveDimensionData

    def saveDimensionData(dbName, tbName, data){
            login('admin','123456')
            dfsPath = 'dfs://' + dbName
            temp = select * from loadTable(dbPath, tbName)
            update temp set EVENT_ID = data.EVENT_ID,S_INFO_WINDCODE = data.S_INFO_WINDCODE, S_ACTIVITIESTYPE=data.S_ACTIVITIESTYPE, S_SURVEYDATE=data.S_SURVEYDATE, S_SURVEYTIME=data.S_SURVEYTIME, ANN_DT = data.ANN_DT,OPDATE=data.OPDATE,OPMODE=data.OPMODE where OBJECT_ID in data.OBJECT_ID
            db = database(dbPath)
            db.dropTable(tbName)
            dt = db.createTable(temp, tbName)
            dt.append!(temp)}

    ​

• 配置文件示例

BASECODE.json

{
    "job": {
        "setting": {
            "speed": {
                "channel": 1
            }
        },
        "content": [
            {
                "reader": {
                    "name": "oraclereader",
                    "parameter": {
                        "username": "root",
                        "password": "password",
                        "column": [
                            "*"
                        ],
                        "connection": [
                            {
                                "table": [
                                    "BASECODE"
                                ],
                                "jdbcUrl": [
                                    "jdbc:oracle:thin:@127.0.0.1:1521:helowin"
                                ]
                            }
                        ],
                        "where":"OPDATE > to_date('2020-03-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS')"
                    }
                },
                "writer": {
                    "name": "dolphindbwriter",
                    "parameter": {
                        "userId": "user",
                        "pwd": "pass",
                        "host": "127.0.0.1",
                        "port": 8848,
                        "dbPath": "dfs://TESTDB",
                        "tableName": "BASECODE",
                        "batchSize": 100000,
                        "table": [
                            {
                                "type": "DT_DOUBLE",
                                "name": "S_INFO_MIN_PRICE_CHG_UNIT"
                            },
                            {
                                "type": "DT_DOUBLE",
                                "name": "S_INFO_LOT_SIZE"
                            },
                            {
                                "type": "DT_STRING",
                                "name": "S_INFO_ENAME"
                            },
                            {
                                "type": "DT_TIMESTAMP",
                                "name": "OPDATE"
                            },
                            {
                                "type": "DT_STRING",
                                "name": "OPMODE"
                            },
                            {
                                "type": "DT_STRING",
                                "name": "OBJECT_ID",
                                "isKeyField" :true
                            },
                            {
                                "type": "DT_STRING",
                                "name": "S_INFO_WINDCODE"
                            },
                            {
                                "type": "DT_STRING",
                                "name": "S_INFO_ASHARECODE"
                            },
                            {
                                "type": "DT_STRING",
                                "name": "S_INFO_COMPCODE"
                            },
                            {
                                "type": "DT_STRING",
                                "name": "S_INFO_SECURITIESTYPES"
                            },
                            {
                                "type": "DT_STRING",
                                "name": "S_INFO_SECTYPENAME"
                            }
                        ]
                    }
                }
            }
        ]
    }
}

• 特殊情况 - 时间精度丢失

reader 读取的上游数据库是用字符串的形式存储的时间字段，且该时间字段的需要保留毫秒及以上的精度，那么如果 dolphindbwriter 对应字段的 type 配置成 DT_TIME, DT_TIMESTAMP 等，会导致时间存入 DolphinDB 分布式表后丢失毫秒及以上的精度。

例如：从 Oracle 向 DolphinDB 导入数据， Oracle 用 varchar 类型存储时间：09:00:01.123 , DolphinDB 的分布式表用 TIME 类型存储该时间, 在 dolphindbwriter 的配置中设置对应字段的 type 为 DT_TIME,，入库之后的结果为 09:00:01.000。

解决方案：

将 dolphindbwriter 对应字段的 type 配置为 STRING， 再通过写自定义处理函数，将 STRING 转换成 DolphinDB 的时间类型。

"writer": {
                    "name": "dolphindbwriter",
                    "parameter": {
                        "userId": "user",
                        "pwd": "pass",
                        "host": "127.0.0.1",
                        "port": 8848,
                        "dbPath": "dfs://TESTDB",
                        "tableName": "BASECODE1",
                        "batchSize": 100000,
                        "saveFunctionName":"updateData",
                        "saveFunctionDef":"def updateData(dbName, tbName, mutable data){login('admin', '123456');data.replaceColumn!(`Dtime,  temporalParse(data.Dtime, 'HH:mm:ss.SSS'));loadTable(dbName, tbName).append!(data)} "
                        "table": [
                            {
                                "type": "DT_STRING",
                                "name": "Dtime"
                            }
                        ]
                    }
                }