diff --git a/core/src/main/bin/perftrace.py b/core/src/main/bin/perftrace.py index 41a1ecb3..b9c79a43 100755 --- a/core/src/main/bin/perftrace.py +++ b/core/src/main/bin/perftrace.py @@ -174,6 +174,9 @@ def parsePluginName(jdbcUrl, pluginType): db2Regex = re.compile('jdbc:(db2)://.*') if (db2Regex.match(jdbcUrl)): name = 'db2' + kingbaseesRegex = re.compile('jdbc:(kingbase8)://.*') + if (kingbaseesRegex.match(jdbcUrl)): + name = 'kingbasees' return "%s%s" % (name, pluginType) def renderDataXJson(paramsDict, readerOrWriter = 'reader', channel = 1): diff --git a/kingbaseesreader/doc/kingbaseesreader.md b/kingbaseesreader/doc/kingbaseesreader.md new file mode 100644 index 00000000..ec5495a6 --- /dev/null +++ b/kingbaseesreader/doc/kingbaseesreader.md @@ -0,0 +1,241 @@ + +# KingbaseesReader 插件文档 + + +___ + + +## 1 快速介绍 + +KingbaseesReader插件实现了从KingbaseES读取数据。在底层实现上，KingbaseesReader通过JDBC连接远程KingbaseES数据库，并执行相应的sql语句将数据从KingbaseES库中SELECT出来。 + +## 2 实现原理 + +简而言之，KingbaseesReader通过JDBC连接器连接到远程的KingbaseES数据库，并根据用户配置的信息生成查询SELECT SQL语句并发送到远程KingbaseES数据库，并将该SQL执行返回结果使用DataX自定义的数据类型拼装为抽象的数据集，并传递给下游Writer处理。 + +对于用户配置Table、Column、Where的信息，KingbaseesReader将其拼接为SQL语句发送到KingbaseES数据库；对于用户配置querySql信息，KingbaseesReader直接将其发送到KingbaseES数据库。 + + +## 3 功能说明 + +### 3.1 配置样例 + +* 配置一个从KingbaseES数据库同步抽取数据到本地的作业: + +``` +{ + "job": { + "setting": { + "speed": { + //设置传输速度，单位为byte/s，DataX运行会尽可能达到该速度但是不超过它. + "byte": 1048576 + }, + //出错限制 + "errorLimit": { + //出错的record条数上限，当大于该值即报错。 + "record": 0, + //出错的record百分比上限 1.0表示100%，0.02表示2% + "percentage": 0.02 + } + }, + "content": [ + { + "reader": { + "name": "kingbaseesreader", + "parameter": { + // 数据库连接用户名 + "username": "xx", + // 数据库连接密码 + "password": "xx", + "column": [ + "id"，"name" + ], + //切分主键 + "splitPk": "id", + "connection": [ + { + "table": [ + "table" + ], + "jdbcUrl": [ + "jdbc:kingbase8://host:port/database" + ] + } + ] + } + }, + "writer": { + //writer类型 + "name": "streamwriter", + //是否打印内容 + "parameter": { + "print":true, + } + } + } + ] + } +} + +``` + +* 配置一个自定义SQL的数据库同步任务到本地内容的作业： + +``` +{ + "job": { + "setting": { + "speed": 1048576 + }, + "content": [ + { + "reader": { + "name": "kingbaseesreader", + "parameter": { + "username": "xx", + "password": "xx", + "where": "", + "connection": [ + { + "querySql": [ + "select db_id,on_line_flag from db_info where db_id < 10;" + ], + "jdbcUrl": [ + "jdbc:kingbase8://host:port/database", "jdbc:kingbase8://host:port/database" + ] + } + ] + } + }, + "writer": { + "name": "streamwriter", + "parameter": { + "print": false, + "encoding": "UTF-8" + } + } + } + ] + } +} +``` + + +### 3.2 参数说明 + +* **jdbcUrl** + + * 描述：描述的是到对端数据库的JDBC连接信息，使用JSON的数组描述，并支持一个库填写多个连接地址。之所以使用JSON数组描述连接信息，是因为阿里集团内部支持多个IP探测，如果配置了多个，KingbaseesReader可以依次探测ip的可连接性，直到选择一个合法的IP。如果全部连接失败，KingbaseesReader报错。 注意，jdbcUrl必须包含在connection配置单元中。对于阿里集团外部使用情况，JSON数组填写一个JDBC连接即可。 + + jdbcUrl按照KingbaseES官方规范，并可以填写连接附件控制信息。具体请参看[KingbaseES官方文档](https://help.kingbase.com.cn/doc-view-5683.html)。 + + * 必选：是
+ + * 默认值：无
+ +* **username** + + * 描述：数据源的用户名
+ + * 必选：是
+ + * 默认值：无
+ +* **password** + + * 描述：数据源指定用户名的密码
+ + * 必选：是
+ + * 默认值：无
+ +* **table** + + * 描述：所选取的需要同步的表。使用JSON的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需保证多张表是同一schema结构，KingbaseesReader不予检查表是否同一逻辑表。注意，table必须包含在connection配置单元中。
+ + * 必选：是
+ + * 默认值：无
+ +* **column** + + * 描述：所配置的表中需要同步的列名集合，使用JSON的数组描述字段信息。用户使用\*代表默认使用所有列配置，例如['\*']。 + + 支持列裁剪，即列可以挑选部分列进行导出。 + + 支持列换序，即列可以不按照表schema信息进行导出。 + + 支持常量配置，用户需要按照KingbaseES语法格式: + ["id", "'hello'::varchar", "true", "2.5::real", "power(2,3)"] + id为普通列名，'hello'::varchar为字符串常量，true为布尔值，2.5为浮点数, power(2,3)为函数。 + + **column必须用户显示指定同步的列集合，不允许为空！** + + * 必选：是
+ + * 默认值：无
+ +* **splitPk** + + * 描述：KingbaseesReader进行数据抽取时，如果指定splitPk，表示用户希望使用splitPk代表的字段进行数据分片，DataX因此会启动并发任务进行数据同步，这样可以大大提供数据同步的效能。 + + 推荐splitPk用户使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。 + + 目前splitPk仅支持整形数据切分，`不支持浮点、字符串型、日期等其他类型`。如果用户指定其他非支持类型，KingbaseesReader将报错！ + + splitPk设置为空，底层将视作用户不允许对单表进行切分，因此使用单通道进行抽取。 + + * 必选：否
+ + * 默认值：空
+ +* **where** + + * 描述：筛选条件，KingbaseesReader根据指定的column、table、where条件拼接SQL，并根据这个SQL进行数据抽取。在实际业务场景中，往往会选择当天的数据进行同步，可以将where条件指定为gmt_create > $bizdate 。注意：不可以将where条件指定为limit 10，limit不是SQL的合法where子句。
+ + where条件可以有效地进行业务增量同步。 where条件不配置或者为空，视作全表同步数据。 + + * 必选：否
+ + * 默认值：无
+ +* **querySql** + + * 描述：在有些业务场景下，where这一配置项不足以描述所筛选的条件，用户可以通过该配置型来自定义筛选SQL。当用户配置了这一项之后，DataX系统就会忽略table，column这些配置型，直接使用这个配置项的内容对数据进行筛选，例如需要进行多表join后同步数据，使用select a,b from table_a join table_b on table_a.id = table_b.id
+ + `当用户配置querySql时，KingbaseesReader直接忽略table、column、where条件的配置`。 + + * 必选：否
+ + * 默认值：无
+ +* **fetchSize** + + * 描述：该配置项定义了插件和数据库服务器端每次批量数据获取条数，该值决定了DataX和服务器端的网络交互次数，能够较大的提升数据抽取性能。
+ + `注意，该值过大(>2048)可能造成DataX进程OOM。`。 + + * 必选：否
+ + * 默认值：1024
+ + +### 3.3 类型转换 + +目前KingbaseesReader支持大部分KingbaseES类型，但也存在部分个别类型没有支持的情况，请注意检查你的类型。 + +下面列出KingbaseesReader针对KingbaseES类型转换列表: + + +| DataX 内部类型| KingbaseES 数据类型 | +| -------- | ----- | +| Long |bigint, bigserial, integer, smallint, serial | +| Double |double precision, money, numeric, real | +| String |varchar, char, text, bit, inet| +| Date |date, time, timestamp | +| Boolean |bool| +| Bytes |bytea| + +请注意: + +* `除上述罗列字段类型外，其他类型均不支持; money,inet,bit需用户使用a_inet::varchar类似的语法转换`。 \ No newline at end of file diff --git a/kingbaseesreader/pom.xml b/kingbaseesreader/pom.xml new file mode 100644 index 00000000..6e844c10 --- /dev/null +++ b/kingbaseesreader/pom.xml @@ -0,0 +1,88 @@ + + + + datax-all + com.alibaba.datax + 0.0.1-SNAPSHOT + + 4.0.0 + + kingbaseesreader + kingbaseesreader + jar + + + + com.alibaba.datax + datax-common + ${datax-project-version} + + + slf4j-log4j12 + org.slf4j + + + + + + org.slf4j + slf4j-api + + + + ch.qos.logback + logback-classic + + + + com.alibaba.datax + plugin-rdbms-util + ${datax-project-version} + + + + com.kingbase8 + kingbase8 + 8.2.0 + system + ${basedir}/src/main/libs/kingbase8-8.2.0.jar + + + + + + + + + maven-compiler-plugin + + ${jdk-version} + ${jdk-version} + ${project-sourceEncoding} + + + + + maven-assembly-plugin + + + src/main/assembly/package.xml + + datax + + + + dwzip + package + + single + + + + + + + + diff --git a/kingbaseesreader/src/main/assembly/package.xml b/kingbaseesreader/src/main/assembly/package.xml new file mode 100644 index 00000000..e369c5f0 --- /dev/null +++ b/kingbaseesreader/src/main/assembly/package.xml @@ -0,0 +1,42 @@ + + + + dir + + false + + + src/main/resources + + plugin.json + plugin_job_template.json + + plugin/reader/kingbaseesreader + + + target/ + + kingbaseesreader-0.0.1-SNAPSHOT.jar + + plugin/reader/kingbaseesreader + + + src/main/libs + + *.* + + plugin/reader/kingbaseesreader/libs + + + + + + false + plugin/reader/kingbaseesreader/libs + runtime + + + diff --git a/kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/Constant.java b/kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/Constant.java new file mode 100644 index 00000000..bed4c6e6 --- /dev/null +++ b/kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/Constant.java @@ -0,0 +1,7 @@ +package com.alibaba.datax.plugin.reader.kingbaseesreader; + +public class Constant { + + public static final int DEFAULT_FETCH_SIZE = 1000; + +} diff --git a/kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/KingbaseesReader.java b/kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/KingbaseesReader.java new file mode 100644 index 00000000..9246655f --- /dev/null +++ b/kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/KingbaseesReader.java @@ -0,0 +1,86 @@ +package com.alibaba.datax.plugin.reader.kingbaseesreader; + +import com.alibaba.datax.common.exception.DataXException; +import com.alibaba.datax.common.plugin.RecordSender; +import com.alibaba.datax.common.spi.Reader; +import com.alibaba.datax.common.util.Configuration; +import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader; +import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode; +import com.alibaba.datax.plugin.rdbms.util.DataBaseType; + +import java.util.List; + +public class KingbaseesReader extends Reader { + + private static final DataBaseType DATABASE_TYPE = DataBaseType.KingbaseES; + + public static class Job extends Reader.Job { + + private Configuration originalConfig; + private CommonRdbmsReader.Job commonRdbmsReaderMaster; + + @Override + public void init() { + this.originalConfig = super.getPluginJobConf(); + int fetchSize = this.originalConfig.getInt(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE, + Constant.DEFAULT_FETCH_SIZE); + if (fetchSize < 1) { + throw DataXException.asDataXException(DBUtilErrorCode.REQUIRED_VALUE, + String.format("您配置的fetchSize有误，根据DataX的设计，fetchSize : [%d] 设置值不能小于 1.", fetchSize)); + } + this.originalConfig.set(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE, fetchSize); + + this.commonRdbmsReaderMaster = new CommonRdbmsReader.Job(DATABASE_TYPE); + this.commonRdbmsReaderMaster.init(this.originalConfig); + } + + @Override + public List split(int adviceNumber) { + return this.commonRdbmsReaderMaster.split(this.originalConfig, adviceNumber); + } + + @Override + public void post() { + this.commonRdbmsReaderMaster.post(this.originalConfig); + } + + @Override + public void destroy() { + this.commonRdbmsReaderMaster.destroy(this.originalConfig); + } + + } + + public static class Task extends Reader.Task { + + private Configuration readerSliceConfig; + private CommonRdbmsReader.Task commonRdbmsReaderSlave; + + @Override + public void init() { + this.readerSliceConfig = super.getPluginJobConf(); + this.commonRdbmsReaderSlave = new CommonRdbmsReader.Task(DATABASE_TYPE, super.getTaskGroupId(), super.getTaskId()); + this.commonRdbmsReaderSlave.init(this.readerSliceConfig); + } + + @Override + public void startRead(RecordSender recordSender) { + int fetchSize = this.readerSliceConfig.getInt(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE); + + this.commonRdbmsReaderSlave.startRead(this.readerSliceConfig, recordSender, + super.getTaskPluginCollector(), fetchSize); + } + + @Override + public void post() { + this.commonRdbmsReaderSlave.post(this.readerSliceConfig); + } + + @Override + public void destroy() { + this.commonRdbmsReaderSlave.destroy(this.readerSliceConfig); + } + + } + +} diff --git a/kingbaseesreader/src/main/libs/kingbase8-8.2.0.jar b/kingbaseesreader/src/main/libs/kingbase8-8.2.0.jar new file mode 100644 index 00000000..0b5ac964 Binary files /dev/null and b/kingbaseesreader/src/main/libs/kingbase8-8.2.0.jar differ diff --git a/kingbaseesreader/src/main/resources/plugin.json b/kingbaseesreader/src/main/resources/plugin.json new file mode 100644 index 00000000..9bc0684b --- /dev/null +++ b/kingbaseesreader/src/main/resources/plugin.json @@ -0,0 +1,6 @@ +{ + "name": "kingbaseesreader", + "class": "com.alibaba.datax.plugin.reader.kingbaseesreader.KingbaseesReader", + "description": "useScene: prod. mechanism: Jdbc connection using the database, execute select sql, retrieve data from the ResultSet. warn: The more you know about the database, the less problems you encounter.", + "developer": "alibaba" +} \ No newline at end of file diff --git a/kingbaseesreader/src/main/resources/plugin_job_template.json b/kingbaseesreader/src/main/resources/plugin_job_template.json new file mode 100644 index 00000000..49c07098 --- /dev/null +++ b/kingbaseesreader/src/main/resources/plugin_job_template.json @@ -0,0 +1,13 @@ +{ + "name": "kingbaseesreader", + "parameter": { + "username": "", + "password": "", + "connection": [ + { + "table": [], + "jdbcUrl": [] + } + ] + } +} \ No newline at end of file diff --git a/kingbaseeswriter/doc/kingbaseeswriter.md b/kingbaseeswriter/doc/kingbaseeswriter.md new file mode 100644 index 00000000..96e1a3ac --- /dev/null +++ b/kingbaseeswriter/doc/kingbaseeswriter.md @@ -0,0 +1,208 @@ +# DataX KingbaseesWriter + + +--- + + +## 1 快速介绍 + +KingbaseesWriter插件实现了写入数据到 KingbaseES主库目的表的功能。在底层实现上，KingbaseesWriter通过JDBC连接远程 KingbaseES 数据库，并执行相应的 insert into ... sql 语句将数据写入 KingbaseES，内部会分批次提交入库。 + +KingbaseesWriter面向ETL开发工程师，他们使用KingbaseesWriter从数仓导入数据到KingbaseES。同时 KingbaseesWriter亦可以作为数据迁移工具为DBA等用户提供服务。 + + +## 2 实现原理 + +KingbaseesWriter通过 DataX 框架获取 Reader 生成的协议数据，根据你配置生成相应的SQL插入语句 + + +* `insert into...`(当主键/唯一性索引冲突时会写不进去冲突的行) + +
+ + 注意： + 1. 目的表所在数据库必须是主库才能写入数据；整个任务至少需具备 insert into...的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。 + 2. KingbaseesWriter和MysqlWriter不同，不支持配置writeMode参数。 + + +## 3 功能说明 + +### 3.1 配置样例 + +* 这里使用一份从内存产生到 KingbaseesWriter导入的数据。 + +```json +{ + "job": { + "setting": { + "speed": { + "channel": 1 + } + }, + "content": [ + { + "reader": { + "name": "streamreader", + "parameter": { + "column" : [ + { + "value": "DataX", + "type": "string" + }, + { + "value": 19880808, + "type": "long" + }, + { + "value": "1988-08-08 08:08:08", + "type": "date" + }, + { + "value": true, + "type": "bool" + }, + { + "value": "test", + "type": "bytes" + } + ], + "sliceRecordCount": 1000 + } + }, + "writer": { + "name": "kingbaseeswriter", + "parameter": { + "username": "xx", + "password": "xx", + "column": [ + "id", + "name" + ], + "preSql": [ + "delete from test" + ], + "connection": [ + { + "jdbcUrl": "jdbc:kingbase8://127.0.0.1:3002/datax", + "table": [ + "test" + ] + } + ] + } + } + } + ] + } +} + +``` + + +### 3.2 参数说明 + +* **jdbcUrl** + + * 描述：目的数据库的 JDBC 连接信息 ,jdbcUrl必须包含在connection配置单元中。 + + 注意：1、在一个数据库上只能配置一个值。 + 2、jdbcUrl按照KingbaseES官方规范，并可以填写连接附加参数信息。具体请参看KingbaseES官方文档或者咨询对应 DBA。 + + + * 必选：是
+ + * 默认值：无
+ +* **username** + + * 描述：目的数据库的用户名
+ + * 必选：是
+ + * 默认值：无
+ +* **password** + + * 描述：目的数据库的密码
+ + * 必选：是
+ + * 默认值：无
+ +* **table** + + * 描述：目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致。 + + 注意：table 和 jdbcUrl 必须包含在 connection 配置单元中 + + * 必选：是
+ + * 默认值：无
+ +* **column** + + * 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。如果要依次写入全部列，使用\*表示, 例如: "column": ["\*"] + + 注意：1、我们强烈不推荐你这样配置，因为当你目的表字段个数、类型等有改动时，你的任务可能运行不正确或者失败 + 2、此处 column 不能配置任何常量值 + + * 必选：是
+ + * 默认值：否
+ +* **preSql** + + * 描述：写入数据到目的表前，会先执行这里的标准语句。如果 Sql 中有你需要操作到的表名称，请使用 `@table` 表示，这样在实际执行 Sql 语句时，会对变量按照实际表名称进行替换。比如你的任务是要写入到目的端的100个同构分表(表名称为:datax_00,datax01, ... datax_98,datax_99)，并且你希望导入数据前，先对表中数据进行删除操作，那么你可以这样配置：`"preSql":["delete from @table"]`，效果是：在执行到每个表写入数据前，会先执行对应的 delete from 对应表名称
+ + * 必选：否
+ + * 默认值：无
+ +* **postSql** + + * 描述：写入数据到目的表后，会执行这里的标准语句。（原理同 preSql ）
+ + * 必选：否
+ + * 默认值：无
+ +* **batchSize** + + * 描述：一次性批量提交的记录数大小，该值可以极大减少DataX与KingbaseES的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。
+ + * 必选：否
+ + * 默认值：1024
+ +### 3.3 类型转换 + +目前 KingbaseesWriter支持大部分 KingbaseES类型，但也存在部分没有支持的情况，请注意检查你的类型。 + +下面列出 KingbaseesWriter针对 KingbaseES类型转换列表: + +| DataX 内部类型| KingbaseES 数据类型 | +| -------- | ----- | +| Long |bigint, bigserial, integer, smallint, serial | +| Double |double precision, money, numeric, real | +| String |varchar, char, text, bit| +| Date |date, time, timestamp | +| Boolean |bool| +| Bytes |bytea| + + +## FAQ + +*** + +**Q: KingbaseesWriter 执行 postSql 语句报错，那么数据导入到目标数据库了吗?** + +A: DataX 导入过程存在三块逻辑，pre 操作、导入操作、post 操作，其中任意一环报错，DataX 作业报错。由于 DataX 不能保证在同一个事务完成上述几个操作，因此有可能数据已经落入到目标端。 + +*** + +**Q: 按照上述说法，那么有部分脏数据导入数据库，如果影响到线上数据库怎么办?** + +A: 目前有两种解法，第一种配置 pre 语句，该 sql 可以清理当天导入数据， DataX 每次导入时候可以把上次清理干净并导入完整数据。 +第二种，向临时表导入数据，完成后再 rename 到线上表。 + +*** diff --git a/kingbaseeswriter/pom.xml b/kingbaseeswriter/pom.xml new file mode 100644 index 00000000..284c8c5e --- /dev/null +++ b/kingbaseeswriter/pom.xml @@ -0,0 +1,84 @@ + + 4.0.0 + + com.alibaba.datax + datax-all + 0.0.1-SNAPSHOT + + kingbaseeswriter + kingbaseeswriter + jar + writer data into kingbasees database + + + + com.alibaba.datax + datax-common + ${datax-project-version} + + + slf4j-log4j12 + org.slf4j + + + + + + org.slf4j + slf4j-api + + + + ch.qos.logback + logback-classic + + + + com.alibaba.datax + plugin-rdbms-util + ${datax-project-version} + + + + com.kingbase8 + kingbase8 + 8.2.0 + system + ${basedir}/src/main/libs/kingbase8-8.2.0.jar + + + + + + + + maven-compiler-plugin + + ${jdk-version} + ${jdk-version} + ${project-sourceEncoding} + + + + + maven-assembly-plugin + + + src/main/assembly/package.xml + + datax + + + + dwzip + package + + single + + + + + + + diff --git a/kingbaseeswriter/src/main/assembly/package.xml b/kingbaseeswriter/src/main/assembly/package.xml new file mode 100644 index 00000000..aa78a6ec --- /dev/null +++ b/kingbaseeswriter/src/main/assembly/package.xml @@ -0,0 +1,42 @@ + + + + dir + + false + + + src/main/resources + + plugin.json + plugin_job_template.json + + plugin/writer/kingbaseeswriter + + + target/ + + kingbaseeswriter-0.0.1-SNAPSHOT.jar + + plugin/writer/kingbaseeswriter + + + src/main/libs + + *.* + + plugin/writer/kingbaseeswriter/libs + + + + + + false + plugin/writer/kingbaseeswriter/libs + runtime + + + diff --git a/kingbaseeswriter/src/main/java/com/alibaba/datax/plugin/writer/kingbaseeswriter/KingbaseesWriter.java b/kingbaseeswriter/src/main/java/com/alibaba/datax/plugin/writer/kingbaseeswriter/KingbaseesWriter.java new file mode 100644 index 00000000..dec5ff95 --- /dev/null +++ b/kingbaseeswriter/src/main/java/com/alibaba/datax/plugin/writer/kingbaseeswriter/KingbaseesWriter.java @@ -0,0 +1,100 @@ +package com.alibaba.datax.plugin.writer.kingbaseeswriter; + +import com.alibaba.datax.common.exception.DataXException; +import com.alibaba.datax.common.plugin.RecordReceiver; +import com.alibaba.datax.common.spi.Writer; +import com.alibaba.datax.common.util.Configuration; +import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode; +import com.alibaba.datax.plugin.rdbms.util.DataBaseType; +import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter; +import com.alibaba.datax.plugin.rdbms.writer.Key; + +import java.util.List; + +public class KingbaseesWriter extends Writer { + private static final DataBaseType DATABASE_TYPE = DataBaseType.KingbaseES; + + public static class Job extends Writer.Job { + private Configuration originalConfig = null; + private CommonRdbmsWriter.Job commonRdbmsWriterMaster; + + @Override + public void init() { + this.originalConfig = super.getPluginJobConf(); + + // warn：not like mysql, KingbaseES only support insert mode, don't use + String writeMode = this.originalConfig.getString(Key.WRITE_MODE); + if (null != writeMode) { + throw DataXException.asDataXException(DBUtilErrorCode.CONF_ERROR, + String.format("写入模式(writeMode)配置有误. 因为KingbaseES不支持配置参数项 writeMode: %s, KingbaseES仅使用insert sql 插入数据. 请检查您的配置并作出修改.", writeMode)); + } + + this.commonRdbmsWriterMaster = new CommonRdbmsWriter.Job(DATABASE_TYPE); + this.commonRdbmsWriterMaster.init(this.originalConfig); + } + + @Override + public void prepare() { + this.commonRdbmsWriterMaster.prepare(this.originalConfig); + } + + @Override + public List split(int mandatoryNumber) { + return this.commonRdbmsWriterMaster.split(this.originalConfig, mandatoryNumber); + } + + @Override + public void post() { + this.commonRdbmsWriterMaster.post(this.originalConfig); + } + + @Override + public void destroy() { + this.commonRdbmsWriterMaster.destroy(this.originalConfig); + } + + } + + public static class Task extends Writer.Task { + private Configuration writerSliceConfig; + private CommonRdbmsWriter.Task commonRdbmsWriterSlave; + + @Override + public void init() { + this.writerSliceConfig = super.getPluginJobConf(); + this.commonRdbmsWriterSlave = new CommonRdbmsWriter.Task(DATABASE_TYPE){ + @Override + public String calcValueHolder(String columnType){ + if("serial".equalsIgnoreCase(columnType)){ + return "?::int"; + }else if("bit".equalsIgnoreCase(columnType)){ + return "?::bit varying"; + } + return "?::" + columnType; + } + }; + this.commonRdbmsWriterSlave.init(this.writerSliceConfig); + } + + @Override + public void prepare() { + this.commonRdbmsWriterSlave.prepare(this.writerSliceConfig); + } + + public void startWrite(RecordReceiver recordReceiver) { + this.commonRdbmsWriterSlave.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector()); + } + + @Override + public void post() { + this.commonRdbmsWriterSlave.post(this.writerSliceConfig); + } + + @Override + public void destroy() { + this.commonRdbmsWriterSlave.destroy(this.writerSliceConfig); + } + + } + +} diff --git a/kingbaseeswriter/src/main/libs/kingbase8-8.2.0.jar b/kingbaseeswriter/src/main/libs/kingbase8-8.2.0.jar new file mode 100644 index 00000000..0b5ac964 Binary files /dev/null and b/kingbaseeswriter/src/main/libs/kingbase8-8.2.0.jar differ diff --git a/kingbaseeswriter/src/main/resources/plugin.json b/kingbaseeswriter/src/main/resources/plugin.json new file mode 100644 index 00000000..83517760 --- /dev/null +++ b/kingbaseeswriter/src/main/resources/plugin.json @@ -0,0 +1,6 @@ +{ + "name": "kingbaseeswriter", + "class": "com.alibaba.datax.plugin.writer.kingbaseeswriter.KingbaseesWriter", + "description": "useScene: prod. mechanism: Jdbc connection using the database, execute insert sql. warn: The more you know about the database, the less problems you encounter.", + "developer": "alibaba" +} \ No newline at end of file diff --git a/kingbaseeswriter/src/main/resources/plugin_job_template.json b/kingbaseeswriter/src/main/resources/plugin_job_template.json new file mode 100644 index 00000000..94b66168 --- /dev/null +++ b/kingbaseeswriter/src/main/resources/plugin_job_template.json @@ -0,0 +1,17 @@ +{ + "name": "kingbaseeswriter", + "parameter": { + "username": "", + "password": "", + "column": [], + "preSql": [], + "connection": [ + { + "jdbcUrl": "", + "table": [] + } + ], + "preSql": [], + "postSql": [] + } +} \ No newline at end of file diff --git a/package.xml b/package.xml index 6d97b372..2323ce18 100755 --- a/package.xml +++ b/package.xml @@ -74,6 +74,13 @@ datax + + kingbaseesreader/target/datax/ + + **/*.* + + datax + rdbmsreader/target/datax/ @@ -266,6 +273,13 @@ datax + + kingbaseeswriter/target/datax/ + + **/*.* + + datax + rdbmswriter/target/datax/ diff --git a/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DataBaseType.java b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DataBaseType.java index ea11d99b..eb82aa44 100755 --- a/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DataBaseType.java +++ b/plugin-rdbms-util/src/main/java/com/alibaba/datax/plugin/rdbms/util/DataBaseType.java @@ -19,7 +19,8 @@ public enum DataBaseType { RDBMS("rdbms", "com.alibaba.datax.plugin.rdbms.util.DataBaseType"), DB2("db2", "com.ibm.db2.jcc.DB2Driver"), ADS("ads","com.mysql.jdbc.Driver"), - ClickHouse("clickhouse", "ru.yandex.clickhouse.ClickHouseDriver"); + ClickHouse("clickhouse", "ru.yandex.clickhouse.ClickHouseDriver"), + KingbaseES("kingbasees", "com.kingbase8.Driver"); private String typeName; @@ -59,6 +60,8 @@ public enum DataBaseType { break; case RDBMS: break; + case KingbaseES: + break; default: throw DataXException.asDataXException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type."); } @@ -98,6 +101,8 @@ public enum DataBaseType { break; case RDBMS: break; + case KingbaseES: + break; default: throw DataXException.asDataXException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type."); } @@ -122,6 +127,7 @@ public enum DataBaseType { break; case DB2: case PostgreSQL: + case KingbaseES: break; default: throw DataXException.asDataXException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type."); @@ -145,6 +151,7 @@ public enum DataBaseType { break; case DB2: case PostgreSQL: + case KingbaseES: break; default: throw DataXException.asDataXException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type"); @@ -168,6 +175,8 @@ public enum DataBaseType { break; case PostgreSQL: break; + case KingbaseES: + break; default: throw DataXException.asDataXException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type"); } diff --git a/pom.xml b/pom.xml index bd270cc5..b3bf7b9d 100755 --- a/pom.xml +++ b/pom.xml @@ -50,6 +50,7 @@ drdsreader sqlserverreader postgresqlreader + kingbaseesreader oraclereader odpsreader otsreader @@ -80,6 +81,7 @@ oraclewriter sqlserverwriter postgresqlwriter + kingbaseeswriter osswriter mongodbwriter adswriter