基于 DataX 的 DolphinDB 数据导入工具

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

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

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

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

  • 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": 1000000,
                        "saveFunctionName":"savePartitionedData",
                        "saveFunctionDef":"def() {...}",
                        "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"
                            }
                        ]
                    }
                }
            }
        ]
    }
}

• 配置文件参数说明

  • host

    • 描述：Server Host。
    • 必选：是。
    • 默认值：无。

  • port

    • 描述：Server Port。
    • 必选：是。
    • 默认值：无。

  • username/userId

    • 描述：DolphinDB 用户名。导入分布式库时，必须要有权限的用户才能操作，否则会返回。
    • 必选：是。
    • 默认值：无。

  • password/pwd

    • 描述：DolphinDB 用户密码。
    • 必选：是。
    • 默认值：无。

  • dbPath

    • 描述：需要写入的目标分布式库名称，比如 "dfs://MYDB"。
    • 必选：是。
    • 默认值：无。

  • preSql

    • 描述：写入数据到目的表前，会先执行这里的 DolphinDB 脚本。
    • 必选：否。
    • 默认值：无。

  • postSql

    • 描述：写入数据到目的表后，会执行这里的 DolphinDB 脚本。
    • 必选：否。
    • 默认值：无。

  • column

    • 描述：目的表需要写入数据的字段。注意：不能与 table 字段同时使用。
    • 必选：否。
    • 默认值：无。
    • 使用说明：
      • 字段之间用英文逗号分隔，例如: "column": 。
      • 如果要依次写入全部列，使用表示，例如: "column": 。
      • 不可以为空 “column": ，或者”column": 。
      • 导入数据是按照原表的字段顺序，无法通过识别新表的字段名称而改变数据顺序。

  • tableName

    • 描述: 目标数据表名称。
    • 必选: 是。
    • 默认值: 无。

  • batchSize

    • 描述: datax 每次写入 dolphindb 的批次记录数。
    • 必须: 否。
    • 默认值: 10,000,000。

  • table

    • 描述：写入表的字段集合。内部结构为：{"name": "columnName", "type": "DT_STRING", "isKeyField":true}。请注意此处列定义的顺序，需要与原表提取的列顺序完全一致。
      • name：字段名称。
      • isKeyField：是否唯一键值，可以允许组合唯一键。本属性用于数据更新场景，用于确认更新数据的主键，若无更新数据的场景，无需设置。
      • type：枚举值以及对应 DataX 数据类型如下表。DolphinDB 的数据类型及精度请参考数据类型。

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

    • 必选：是。
    • 默认值：无。

• saveFunctionName

  • 描述：自定义数据处理函数。若未指定此配置，插件在接收到 reader 的数据后，会将数据提交到 DolphinDB 并通过 tableInsert 函数写入指定库表；如果定义此参数，则会用指定函数替换 tableInsert 函数。
  • 必选：否。
  • 默认值：无。 也可以指定自定义函数。 插件内置了 savePartitionedData(更新分布式表)/saveDimensionData(更新维度表) 两个函数，当 saveFunctionDef 未定义或为空时，saveFunctionName 可以取枚举值之一，对应用于更新分布式表和维度表的数据处理。

• saveFunctionDef

  • 描述：数据入库自定义函数。此函数指 用dolphindb 脚本来实现的数据入库过程。此函数必须接受三个参数：dfsPath(分布式库路径)，tbName(数据表名)，data(从 datax 导入的数据，table 格式)
  • 必选：当 saveFunctionName 参数不为空且非两个枚举值之一时，此参数必填。
  • 默认值：无。

• 引入 DolphinDB 的 upsert! 功能，修改配置文件 BASECODE.json 中的 writer 部分。

  "saveFunctionName":"upsertTable",
  "saveFunctionDef":"ignoreNull=true;keyColNames=`id;sortColumns=`value"