diff --git a/dorisreader/doc/dorisreader.md b/dorisreader/doc/dorisreader.md new file mode 100644 index 0000000000..c249c17834 --- /dev/null +++ b/dorisreader/doc/dorisreader.md @@ -0,0 +1,224 @@ +# DorisReader 插件文档 + +___ + +## 1 快速介绍 + +DorisReader插件实现了从Doris读取数据。在底层实现上，DorisReader通过JDBC连接远程Doris数据库，并执行相应的sql语句将数据从doris库中SELECT出来。 + +## 2 实现原理 + +简而言之，DorisReader通过JDBC连接器连接到远程的Doris数据库，并根据用户配置的信息生成查询SELECT +SQL语句，然后发送到远程Doris数据库，并将该SQL执行返回结果使用DataX自定义的数据类型拼装为抽象的数据集，并传递给下游Writer处理。 + +对于用户配置Table、Column、Where的信息，DorisReader将其拼接为SQL语句发送到Doris数据库；对于用户配置querySql信息，DorisReader直接将其发送到Doris数据库。 + +## 3 功能说明 + +### 3.1 配置样例 + +* 配置一个从Doris数据库同步抽取数据到本地的作业: + +``` +{ + "job": { + "setting": { + "speed": { + "channel": 3 + }, + "errorLimit": { + "record": 0, + "percentage": 0.02 + } + }, + "content": [ + { + "reader": { + "name": "dorisreader", + "parameter": { + "username": "root", + "password": "root", + "column": [ + "id", + "name" + ], + "splitPk": "db_id", + "connection": [ + { + "table": [ + "table" + ], + "jdbcUrl": [ + "jdbc:Doris://127.0.0.1:9030/database" + ] + } + ] + } + }, + "writer": { + "name": "streamwriter", + "parameter": { + "print":true + } + } + } + ] + } +} + +``` + +* 配置一个自定义SQL的数据库同步任务到本地内容的作业： + +``` +{ + "job": { + "setting": { + "speed": { + "channel":1 + } + }, + "content": [ + { + "reader": { + "name": "dorisreader", + "parameter": { + "username": "root", + "password": "root", + "connection": [ + { + "querySql": [ + "select db_id,on_line_flag from db_info where db_id < 10;", + "select db_id,on_line_flag from db_info where db_id >= 10;" + + ], + "jdbcUrl": [ + "jdbc:Doris://127.0.0.1:9030/database" + ] + } + ] + } + }, + "writer": { + "name": "streamwriter", + "parameter": { + "print": false, + "encoding": "UTF-8" + } + } + } + ] + } +} +``` + +### 3.2 参数说明 + +* **jdbcUrl** + + * + 描述：描述的是到对端数据库的JDBC连接信息，使用JSON的数组描述，并支持一个库填写多个连接地址。之所以使用JSON数组描述连接信息，是因为阿里集团内部支持多个IP探测，如果配置了多个，DorisReader可以依次探测ip的可连接性，直到选择一个合法的IP。如果全部连接失败，DorisReader报错。 + 注意，jdbcUrl必须包含在connection配置单元中。对于阿里集团外部使用情况，JSON数组填写一个JDBC连接即可。 + + * 必选：是
+ + * 默认值：无
+ +* **username** + + * 描述：数据源的用户名
+ + * 必选：是
+ + * 默认值：无
+ +* **password** + + * 描述：数据源指定用户名的密码
+ + * 必选：是
+ + * 默认值：无
+ +* **table** + + * + 描述：所选取的需要同步的表。使用JSON的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需保证多张表是同一schema结构，DorisReader不予检查表是否同一逻辑表。注意，table必须包含在connection配置单元中。
+ + * 必选：是
+ + * 默认值：无
+ +* **column** + + * 描述：所配置的表中需要同步的列名集合，使用JSON的数组描述字段信息。用户使用\*代表默认使用所有列配置，例如['\*']。 + + 支持列裁剪，即列可以挑选部分列进行导出。 + + 支持列换序，即列可以不按照表schema信息进行导出。 + + 支持常量配置，用户需要按照Doris SQL语法格式: + ["id", "\`table\`", "1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"] + id为普通列名，\`table\`为包含保留字的列名，1为整形数字常量，'bazhen.csy'为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。 + + * 必选：是
+ + * 默认值：无
+ +* **splitPk** + + * 描述：DorisReader进行数据抽取时，如果指定splitPk，表示用户希望使用splitPk代表的字段进行数据分片，DataX因此会启动并发任务进行数据同步，这样可以大大提供数据同步的效能。 + + 推荐splitPk用户使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。 + + 目前splitPk仅支持整形数据切分，`不支持浮点、字符串、日期等其他类型`。如果用户指定其他非支持类型，DorisReader将报错！ + + 如果splitPk不填写，包括不提供splitPk或者splitPk值为空，DataX视作使用单通道同步该表数据。 + + * 必选：否
+ + * 默认值：空
+ +* **where** + + * 描述：筛选条件，DorisReader根据指定的column、table、where条件拼接SQL，并根据这个SQL进行数据抽取。在实际业务场景中，往往会选择当天的数据进行同步，可以将where条件指定为gmt_create > + $bizdate 。注意：不可以将where条件指定为limit 10，limit不是SQL的合法where子句。
+ + where条件可以有效地进行业务增量同步。如果不填写where语句，包括不提供where的key或者value，DataX均视作同步全量数据。 + + * 必选：否
+ + * 默认值：无
+ +* **querySql** + + * + 描述：在有些业务场景下，where这一配置项不足以描述所筛选的条件，用户可以通过该配置型来自定义筛选SQL。当用户配置了这一项之后，DataX系统就会忽略table，column这些配置型，直接使用这个配置项的内容对数据进行筛选，例如需要进行多表join后同步数据，使用select + a,b from table_a join table_b on table_a.id = table_b.id
+ + `当用户配置querySql时，DorisReader直接忽略table、column、where条件的配置`，querySql优先级大于table、column、where选项。 + + * 必选：否
+ + * 默认值：无
+ +### 3.3 类型转换 + +目前DorisReader支持大部分Doris类型，但也存在部分个别类型没有支持的情况，请注意检查你的类型。 + +下面列出DorisReaderr针对Doris类型转换列表: + +| DataX 内部类型| doris 数据类型 | +| -------- |-------------------------------------------------------| +| Long | int, tinyint, smallint, int, bigint,Largint | +| Double | float, double, decimal | +| String | varchar, char, text, string, map, json, array, struct | +| Date | date, datetime | +| Boolean | Boolean | + +请注意: + +* `tinyint(1) DataX视作为整形`。 + + + diff --git a/dorisreader/pom.xml b/dorisreader/pom.xml new file mode 100755 index 0000000000..15a025b6c4 --- /dev/null +++ b/dorisreader/pom.xml @@ -0,0 +1,81 @@ + + + 4.0.0 + + com.alibaba.datax + datax-all + 0.0.1-SNAPSHOT + + dorisreader + dorisreader + 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} + + + mysql + mysql-connector-java + ${mysql.driver.version} + + + + + + + + + + maven-compiler-plugin + + ${jdk-version} + ${jdk-version} + ${project-sourceEncoding} + + + + + maven-assembly-plugin + + + src/main/assembly/package.xml + + datax + + + + dwzip + package + + single + + + + + + + diff --git a/dorisreader/src/main/assembly/package.xml b/dorisreader/src/main/assembly/package.xml new file mode 100755 index 0000000000..724613f9b9 --- /dev/null +++ b/dorisreader/src/main/assembly/package.xml @@ -0,0 +1,35 @@ + + + + dir + + false + + + src/main/resources + + plugin.json + plugin_job_template.json + + plugin/reader/dorisreader + + + target/ + + dorisreader-0.0.1-SNAPSHOT.jar + + plugin/reader/dorisreader + + + + + + false + plugin/reader/dorisreader/libs + runtime + + + diff --git a/dorisreader/src/main/java/com/alibaba/datax/plugin/reader/dorisreader/DorisReader.java b/dorisreader/src/main/java/com/alibaba/datax/plugin/reader/dorisreader/DorisReader.java new file mode 100755 index 0000000000..56a44316c9 --- /dev/null +++ b/dorisreader/src/main/java/com/alibaba/datax/plugin/reader/dorisreader/DorisReader.java @@ -0,0 +1,94 @@ +package com.alibaba.datax.plugin.reader.dorisreader; + +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.reader.Constant; +import com.alibaba.datax.plugin.rdbms.util.DataBaseType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +public class DorisReader extends Reader { + + private static final DataBaseType DATABASE_TYPE = DataBaseType.Doris; + + public static class Job extends Reader.Job { + private static final Logger LOG = LoggerFactory + .getLogger(Job.class); + + private Configuration originalConfig = null; + private CommonRdbmsReader.Job commonRdbmsReaderJob; + + @Override + public void init() { + this.originalConfig = super.getPluginJobConf(); + + Integer fetchSize = this.originalConfig.getInt(Constant.FETCH_SIZE,Integer.MIN_VALUE); + this.originalConfig.set(Constant.FETCH_SIZE, fetchSize); + + this.commonRdbmsReaderJob = new CommonRdbmsReader.Job(DATABASE_TYPE); + this.commonRdbmsReaderJob.init(this.originalConfig); + } + + @Override + public void preCheck(){ + init(); + this.commonRdbmsReaderJob.preCheck(this.originalConfig,DATABASE_TYPE); + + } + + @Override + public List split(int adviceNumber) { + return this.commonRdbmsReaderJob.split(this.originalConfig, adviceNumber); + + } + + @Override + public void post() { + this.commonRdbmsReaderJob.post(this.originalConfig); + } + + @Override + public void destroy() { + this.commonRdbmsReaderJob.destroy(this.originalConfig); + } + + } + + public static class Task extends Reader.Task { + + private Configuration readerSliceConfig; + private CommonRdbmsReader.Task commonRdbmsReaderTask; + + @Override + public void init() { + this.readerSliceConfig = super.getPluginJobConf(); + this.commonRdbmsReaderTask = new CommonRdbmsReader.Task(DATABASE_TYPE,super.getTaskGroupId(), super.getTaskId()); + this.commonRdbmsReaderTask.init(this.readerSliceConfig); + + } + + @Override + public void startRead(RecordSender recordSender) { + int fetchSize = this.readerSliceConfig.getInt(Constant.FETCH_SIZE); + + this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender, + super.getTaskPluginCollector(), fetchSize); + } + + @Override + public void post() { + this.commonRdbmsReaderTask.post(this.readerSliceConfig); + } + + @Override + public void destroy() { + this.commonRdbmsReaderTask.destroy(this.readerSliceConfig); + } + + } + +} diff --git a/dorisreader/src/main/java/com/alibaba/datax/plugin/reader/dorisreader/DorisReaderErrorCode.java b/dorisreader/src/main/java/com/alibaba/datax/plugin/reader/dorisreader/DorisReaderErrorCode.java new file mode 100755 index 0000000000..f9a8c4495b --- /dev/null +++ b/dorisreader/src/main/java/com/alibaba/datax/plugin/reader/dorisreader/DorisReaderErrorCode.java @@ -0,0 +1,31 @@ +package com.alibaba.datax.plugin.reader.dorisreader; + +import com.alibaba.datax.common.spi.ErrorCode; + +public enum DorisReaderErrorCode implements ErrorCode { + ; + + private final String code; + private final String description; + + private DorisReaderErrorCode(String code, String description) { + this.code = code; + this.description = description; + } + + @Override + public String getCode() { + return this.code; + } + + @Override + public String getDescription() { + return this.description; + } + + @Override + public String toString() { + return String.format("Code:[%s], Description:[%s]. ", this.code, + this.description); + } +} diff --git a/dorisreader/src/main/resources/plugin.json b/dorisreader/src/main/resources/plugin.json new file mode 100755 index 0000000000..981d1af897 --- /dev/null +++ b/dorisreader/src/main/resources/plugin.json @@ -0,0 +1,6 @@ +{ + "name": "dorisreader", + "class": "com.alibaba.datax.plugin.reader.dorisreader.DorisReader", + "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/dorisreader/src/main/resources/plugin_job_template.json b/dorisreader/src/main/resources/plugin_job_template.json new file mode 100644 index 0000000000..2e3d9fa831 --- /dev/null +++ b/dorisreader/src/main/resources/plugin_job_template.json @@ -0,0 +1,15 @@ +{ + "name": "dorisreader", + "parameter": { + "username": "", + "password": "", + "column": [], + "connection": [ + { + "jdbcUrl": [], + "table": [] + } + ], + "where": "" + } +} \ No newline at end of file diff --git a/package.xml b/package.xml index 41cd9c5515..5ea8bd22ff 100644 --- a/package.xml +++ b/package.xml @@ -236,6 +236,13 @@ datax + + dorisreader/target/datax/ + + **/*.* + + 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 dfb39d8550..cc61ee6f1d 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 @@ -26,7 +26,8 @@ public enum DataBaseType { OceanBase("oceanbase", "com.alipay.oceanbase.jdbc.Driver"), StarRocks("starrocks", "com.mysql.jdbc.Driver"), GaussDB("gaussdb", "org.opengauss.Driver"), - Databend("databend", "com.databend.jdbc.DatabendDriver"); + Databend("databend", "com.databend.jdbc.DatabendDriver"), + Doris("doris","com.mysql.jdbc.Driver"); private String typeName; private String driverClassName; @@ -74,6 +75,8 @@ public String appendJDBCSuffixForReader(String jdbc) { break; case GaussDB: break; + case Doris: + break; default: throw DataXException.asDataXException(DBUtilErrorCode.UNSUPPORTED_TYPE, "unsupported database type."); } diff --git a/pom.xml b/pom.xml index 7847ee5bca..03020dd898 100644 --- a/pom.xml +++ b/pom.xml @@ -80,7 +80,7 @@ loghubreader datahubreader starrocksreader - + dorisreader mysqlwriter starrockswriter