github/DataX
5
0
Fork 0
mirror of https://github.com/alibaba/DataX.git synced 2025-05-03 03:40:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #19 from alibaba/master

Browse Source 
merge with upstream
This commit is contained in:
Shuduo Sang 2022-01-24 20:36:49 +08:00 committed by GitHub
commit 41c577da90
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
44 changed files with 1165 additions and 443 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -47,7 +47,7 @@ DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、N
| 阿里云数仓数据存储 | ODPS | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/odpsreader/doc/odpsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/odpswriter/doc/odpswriter.md)| | 阿里云数仓数据存储 | ODPS | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/odpsreader/doc/odpsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/odpswriter/doc/odpswriter.md)|
| | ADS | | √ |[写](https://github.com/alibaba/DataX/blob/master/adswriter/doc/adswriter.md)| | | ADS | | √ |[写](https://github.com/alibaba/DataX/blob/master/adswriter/doc/adswriter.md)|
| | OSS | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/ossreader/doc/ossreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/osswriter/doc/osswriter.md)| | | OSS | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/ossreader/doc/ossreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/osswriter/doc/osswriter.md)|
| | OCS | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/ocsreader/doc/ocsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/ocswriter/doc/ocswriter.md)| | | OCS | | √ |[写](https://github.com/alibaba/DataX/blob/master/ocswriter/doc/ocswriter.md)|
| NoSQL数据存储 | OTS | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/otsreader/doc/otsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/otswriter/doc/otswriter.md)| | NoSQL数据存储 | OTS | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/otsreader/doc/otsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/otswriter/doc/otswriter.md)|
| | Hbase0.94 | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/hbase094xreader/doc/hbase094xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase094xwriter/doc/hbase094xwriter.md)| | | Hbase0.94 | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/hbase094xreader/doc/hbase094xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase094xwriter/doc/hbase094xwriter.md)|
| | Hbase1.1 | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/hbase11xreader/doc/hbase11xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase11xwriter/doc/hbase11xwriter.md)| | | Hbase1.1 | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/hbase11xreader/doc/hbase11xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase11xwriter/doc/hbase11xwriter.md)|

2
mysqlreader/doc/mysqlreader.md
View File

@ -165,7 +165,7 @@ MysqlReader插件实现了从Mysql读取数据。在底层实现上，MysqlReade
支持常量配置，用户需要按照Mysql SQL语法格式: 支持常量配置，用户需要按照Mysql 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"]
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为布尔值。
* 必选：是 <br /> * 必选：是 <br />

17
oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/OceanBaseReader.java
View File

@ -32,19 +32,21 @@ public class OceanBaseReader extends Reader {
if (userConfigedFetchSize != null) { if (userConfigedFetchSize != null) {
LOG.warn("The [fetchSize] is not recognized, please use readBatchSize instead."); LOG.warn("The [fetchSize] is not recognized, please use readBatchSize instead.");
} }
this.originalConfig.set(Constant.FETCH_SIZE, Integer.MIN_VALUE); this.originalConfig.set(Constant.FETCH_SIZE, Integer.MIN_VALUE);
setDatabaseType(originalConfig); setDatabaseType(originalConfig);
this.readerJob = new ReaderJob(); this.readerJob = new ReaderJob();
this.readerJob.init(this.originalConfig); this.readerJob.init(this.originalConfig);
} }
@Override
public void prepare() {
//ObReaderUtils.DATABASE_TYPE获取当前数据库的语法模式
}
@Override @Override
public void preCheck() { public void preCheck() {
init(); init();
this.readerJob.preCheck(this.originalConfig, ObReaderUtils.DATABASE_TYPE); this.readerJob.preCheck(this.originalConfig, ObReaderUtils.databaseType);
} }
@ -70,7 +72,7 @@ public class OceanBaseReader extends Reader {
Configuration connConf = Configuration.from(conns.get(0).toString()); Configuration connConf = Configuration.from(conns.get(0).toString());
List<String> jdbcUrls = connConf.getList(Key.JDBC_URL, String.class); List<String> jdbcUrls = connConf.getList(Key.JDBC_URL, String.class);
String jdbcUrl = jdbcUrls.get(0); String jdbcUrl = jdbcUrls.get(0);
if(jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) { if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN); String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
if (ss.length != 3) { if (ss.length != 3) {
LOG.warn("unrecognized jdbc url: " + jdbcUrl); LOG.warn("unrecognized jdbc url: " + jdbcUrl);
@ -85,9 +87,10 @@ public class OceanBaseReader extends Reader {
Connection conn = DBUtil.getConnection(DataBaseType.OceanBase, obJdbcUrl, username, password); Connection conn = DBUtil.getConnection(DataBaseType.OceanBase, obJdbcUrl, username, password);
String compatibleMode = ObReaderUtils.getCompatibleMode(conn); String compatibleMode = ObReaderUtils.getCompatibleMode(conn);
if (ObReaderUtils.isOracleMode(compatibleMode)) { if (ObReaderUtils.isOracleMode(compatibleMode)) {
ObReaderUtils.DATABASE_TYPE = DataBaseType.OceanBase; ObReaderUtils.compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
} }
} catch (Exception e){
} catch (Exception e) {
LOG.warn("error in get compatible mode, using mysql as default: " + e.getMessage()); LOG.warn("error in get compatible mode, using mysql as default: " + e.getMessage());
} }
} }

27
oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderJob.java
View File

@ -6,13 +6,38 @@ import com.alibaba.datax.common.constant.CommonConstant;
import com.alibaba.datax.common.util.Configuration; import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader; import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
import com.alibaba.datax.plugin.rdbms.reader.Key; import com.alibaba.datax.plugin.rdbms.reader.Key;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import com.alibaba.datax.plugin.rdbms.writer.Constant; import com.alibaba.datax.plugin.rdbms.writer.Constant;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.OceanBaseReader;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils; import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReaderJob extends CommonRdbmsReader.Job { public class ReaderJob extends CommonRdbmsReader.Job {
private Logger LOG = LoggerFactory.getLogger(OceanBaseReader.Task.class);
public ReaderJob() { public ReaderJob() {
super(ObReaderUtils.DATABASE_TYPE); super(ObReaderUtils.databaseType);
}
@Override
public void init(Configuration originalConfig) {
//将config中的column和table中的关键字进行转义
List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
ObReaderUtils.escapeDatabaseKeywords(columns);
originalConfig.set(Key.COLUMN, columns);
List<JSONObject> conns = originalConfig.getList(com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, JSONObject.class);
for (int i = 0; i < conns.size(); i++) {
JSONObject conn = conns.get(i);
Configuration connConfig = Configuration.from(conn.toString());
List<String> tables = connConfig.getList(Key.TABLE, String.class);
ObReaderUtils.escapeDatabaseKeywords(tables);
originalConfig.set(String.format("%s[%d].%s", com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, i, Key.TABLE), tables);
}
super.init(originalConfig);
} }
@Override @Override

44
oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderTask.java
View File

@ -1,13 +1,5 @@
package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext; package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.datax.common.element.Column; import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record; import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.plugin.RecordSender; import com.alibaba.datax.common.plugin.RecordSender;
@ -19,11 +11,17 @@ import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
import com.alibaba.datax.plugin.rdbms.reader.Constant; import com.alibaba.datax.plugin.rdbms.reader.Constant;
import com.alibaba.datax.plugin.rdbms.reader.Key; import com.alibaba.datax.plugin.rdbms.reader.Key;
import com.alibaba.datax.plugin.rdbms.util.DBUtil; import com.alibaba.datax.plugin.rdbms.util.DBUtil;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import com.alibaba.datax.plugin.rdbms.util.RdbmsException; import com.alibaba.datax.plugin.rdbms.util.RdbmsException;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.Config; import com.alibaba.datax.plugin.reader.oceanbasev10reader.Config;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils; import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.TaskContext; import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.TaskContext;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
public class ReaderTask extends CommonRdbmsReader.Task { public class ReaderTask extends CommonRdbmsReader.Task {
private static final Logger LOG = LoggerFactory.getLogger(ReaderTask.class); private static final Logger LOG = LoggerFactory.getLogger(ReaderTask.class);
@ -41,11 +39,12 @@ public class ReaderTask extends CommonRdbmsReader.Task {
private boolean reuseConn = false; private boolean reuseConn = false;
public ReaderTask(int taskGroupId, int taskId) { public ReaderTask(int taskGroupId, int taskId) {
super(ObReaderUtils.DATABASE_TYPE, taskGroupId, taskId); super(ObReaderUtils.databaseType, taskGroupId, taskId);
this.taskGroupId = taskGroupId; this.taskGroupId = taskGroupId;
this.taskId = taskId; this.taskId = taskId;
} }
@Override
public void init(Configuration readerSliceConfig) { public void init(Configuration readerSliceConfig) {
/* for database connection */ /* for database connection */
username = readerSliceConfig.getString(Key.USERNAME); username = readerSliceConfig.getString(Key.USERNAME);
@ -54,7 +53,7 @@ public class ReaderTask extends CommonRdbmsReader.Task {
queryTimeoutSeconds = readerSliceConfig.getInt(Config.QUERY_TIMEOUT_SECOND, queryTimeoutSeconds = readerSliceConfig.getInt(Config.QUERY_TIMEOUT_SECOND,
Config.DEFAULT_QUERY_TIMEOUT_SECOND); Config.DEFAULT_QUERY_TIMEOUT_SECOND);
// ob10的处理 // ob10的处理
if(jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) { if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN); String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
if (ss.length == 3) { if (ss.length == 3) {
LOG.info("this is ob1_0 jdbc url."); LOG.info("this is ob1_0 jdbc url.");
@ -63,16 +62,14 @@ public class ReaderTask extends CommonRdbmsReader.Task {
} }
} }
if (ObReaderUtils.DATABASE_TYPE == DataBaseType.OceanBase) { jdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:") + "&socketTimeout=1800000&connectTimeout=60000"; //socketTimeout 半个小时
jdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:") + "&socketTimeout=1800000&connectTimeout=60000"; //socketTimeout 半个小时 if (ObReaderUtils.compatibleMode.equals(ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE)) {
compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE; compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
} else {
jdbcUrl = jdbcUrl + "&socketTimeout=1800000&connectTimeout=60000"; //socketTimeout 半个小时
} }
LOG.info("this is ob1_0 jdbc url. user=" + username + " :url=" + jdbcUrl); LOG.info("this is ob1_0 jdbc url. user=" + username + " :url=" + jdbcUrl);
mandatoryEncoding = readerSliceConfig.getString(Key.MANDATORY_ENCODING, ""); mandatoryEncoding = readerSliceConfig.getString(Key.MANDATORY_ENCODING, "");
retryLimit = readerSliceConfig.getInt(Config.RETRY_LIMIT, Config.DEFAULT_RETRY_LIMIT); retryLimit = readerSliceConfig.getInt(Config.RETRY_LIMIT, Config.DEFAULT_RETRY_LIMIT);
LOG.info("retryLimit: "+ retryLimit); LOG.info("retryLimit: " + retryLimit);
} }
private void buildSavePoint(TaskContext context) { private void buildSavePoint(TaskContext context) {
@ -83,7 +80,6 @@ public class ReaderTask extends CommonRdbmsReader.Task {
} }
/** /**
*
* 如果isTableMode && table有PK * 如果isTableMode && table有PK
* <p> * <p>
* 则支持断点续读 (若pk不在原始的columns中,则追加到尾部,但不传给下游) * 则支持断点续读 (若pk不在原始的columns中,则追加到尾部,但不传给下游)
@ -92,7 +88,7 @@ public class ReaderTask extends CommonRdbmsReader.Task {
*/ */
@Override @Override
public void startRead(Configuration readerSliceConfig, RecordSender recordSender, public void startRead(Configuration readerSliceConfig, RecordSender recordSender,
TaskPluginCollector taskPluginCollector, int fetchSize) { TaskPluginCollector taskPluginCollector, int fetchSize) {
String querySql = readerSliceConfig.getString(Key.QUERY_SQL); String querySql = readerSliceConfig.getString(Key.QUERY_SQL);
String table = readerSliceConfig.getString(Key.TABLE); String table = readerSliceConfig.getString(Key.TABLE);
PerfTrace.getInstance().addTaskDetails(taskId, table + "," + jdbcUrl); PerfTrace.getInstance().addTaskDetails(taskId, table + "," + jdbcUrl);
@ -131,14 +127,14 @@ public class ReaderTask extends CommonRdbmsReader.Task {
} }
private void startRead0(boolean isTableMode, TaskContext context, RecordSender recordSender, private void startRead0(boolean isTableMode, TaskContext context, RecordSender recordSender,
TaskPluginCollector taskPluginCollector) { TaskPluginCollector taskPluginCollector) {
// 不是table模式 直接使用原来的做法 // 不是table模式 直接使用原来的做法
if (!isTableMode) { if (!isTableMode) {
doRead(recordSender, taskPluginCollector, context); doRead(recordSender, taskPluginCollector, context);
return; return;
} }
// check primary key index // check primary key index
Connection conn = DBUtil.getConnection(ObReaderUtils.DATABASE_TYPE, jdbcUrl, username, password); Connection conn = DBUtil.getConnection(ObReaderUtils.databaseType, jdbcUrl, username, password);
ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds); ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
context.setConn(conn); context.setConn(conn);
try { try {
@ -184,11 +180,11 @@ public class ReaderTask extends CommonRdbmsReader.Task {
} }
} catch (Throwable e) { } catch (Throwable e) {
if (retryLimit == ++retryCount) { if (retryLimit == ++retryCount) {
throw RdbmsException.asQueryException(ObReaderUtils.DATABASE_TYPE, new Exception(e), throw RdbmsException.asQueryException(ObReaderUtils.databaseType, new Exception(e),
context.getQuerySql(), context.getTable(), username); context.getQuerySql(), context.getTable(), username);
} }
LOG.error("read fail, retry count " + retryCount + ", sleep 60 second, save point:" + LOG.error("read fail, retry count " + retryCount + ", sleep 60 second, save point:" +
context.getSavePoint() + ", error: "+ e.getMessage()); context.getSavePoint() + ", error: " + e.getMessage());
ObReaderUtils.sleep(60000); // sleep 10s ObReaderUtils.sleep(60000); // sleep 10s
} }
// 假如原来的查询有查出数据,则改成增量查询 // 假如原来的查询有查出数据,则改成增量查询
@ -227,7 +223,7 @@ public class ReaderTask extends CommonRdbmsReader.Task {
LOG.info("connection is alive, will reuse this connection."); LOG.info("connection is alive, will reuse this connection.");
} else { } else {
LOG.info("Create new connection for reader."); LOG.info("Create new connection for reader.");
conn = DBUtil.getConnection(ObReaderUtils.DATABASE_TYPE, jdbcUrl, username, password); conn = DBUtil.getConnection(ObReaderUtils.databaseType, jdbcUrl, username, password);
ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds); ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
context.setConn(conn); context.setConn(conn);
} }
@ -287,7 +283,7 @@ public class ReaderTask extends CommonRdbmsReader.Task {
ObReaderUtils.close(null, null, context.getConn()); ObReaderUtils.close(null, null, context.getConn());
context.setConn(null); context.setConn(null);
LOG.error("reader data fail", e); LOG.error("reader data fail", e);
throw RdbmsException.asQueryException(ObReaderUtils.DATABASE_TYPE, e, context.getQuerySql(), throw RdbmsException.asQueryException(ObReaderUtils.databaseType, e, context.getQuerySql(),
context.getTable(), username); context.getTable(), username);
} finally { } finally {
perfRecord.end(); perfRecord.end();

164
oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/ObReaderUtils.java
View File

File diff suppressed because one or more lines are too long

1
oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/TaskContext.java
View File

@ -162,6 +162,7 @@ public class TaskContext {
public String getUserSavePoint() { public String getUserSavePoint() {
return userSavePoint; return userSavePoint;
} }
public void setUserSavePoint(String userSavePoint) { public void setUserSavePoint(String userSavePoint) {
this.userSavePoint = userSavePoint; this.userSavePoint = userSavePoint;
} }

33
oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/OceanBaseV10Writer.java
View File

@ -1,15 +1,5 @@
package com.alibaba.datax.plugin.writer.oceanbasev10writer; package com.alibaba.datax.plugin.writer.oceanbasev10writer;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.DbUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.datax.common.plugin.RecordReceiver; import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.spi.Writer; import com.alibaba.datax.common.spi.Writer;
import com.alibaba.datax.common.util.Configuration; import com.alibaba.datax.common.util.Configuration;
@ -20,7 +10,16 @@ import com.alibaba.datax.plugin.rdbms.writer.Constant;
import com.alibaba.datax.plugin.rdbms.writer.Key; import com.alibaba.datax.plugin.rdbms.writer.Key;
import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil; import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.ConcurrentTableWriterTask; import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.ConcurrentTableWriterTask;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.DbUtils;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils; import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
/** /**
* 2016-04-07 * 2016-04-07
@ -60,6 +59,19 @@ public class OceanBaseV10Writer extends Writer {
public void init() { public void init() {
this.originalConfig = super.getPluginJobConf(); this.originalConfig = super.getPluginJobConf();
checkCompatibleMode(originalConfig); checkCompatibleMode(originalConfig);
//将config中的column和table中的关键字进行转义
List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
ObWriterUtils.escapeDatabaseKeywords(columns);
originalConfig.set(Key.COLUMN, columns);
List<JSONObject> conns = originalConfig.getList(Constant.CONN_MARK, JSONObject.class);
for (int i = 0; i < conns.size(); i++) {
JSONObject conn = conns.get(i);
Configuration connConfig = Configuration.from(conn.toString());
List<String> tables = connConfig.getList(Key.TABLE, String.class);
ObWriterUtils.escapeDatabaseKeywords(tables);
originalConfig.set(String.format("%s[%d].%s", Constant.CONN_MARK, i, Key.TABLE), tables);
}
this.commonJob = new CommonRdbmsWriter.Job(DATABASE_TYPE); this.commonJob = new CommonRdbmsWriter.Job(DATABASE_TYPE);
this.commonJob.init(this.originalConfig); this.commonJob.init(this.originalConfig);
} }
@ -222,6 +234,7 @@ public class OceanBaseV10Writer extends Writer {
/** /**
* 注意：此方法每个 Task 都会执行一次。 最佳实践：此处适当封装确保简洁清晰完成数据写入工作。 * 注意：此方法每个 Task 都会执行一次。 最佳实践：此处适当封装确保简洁清晰完成数据写入工作。
*/ */
@Override
public void startWrite(RecordReceiver recordReceiver) { public void startWrite(RecordReceiver recordReceiver) {
this.writerTask.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector()); this.writerTask.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector());
} }

125
oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java
View File

@ -1,7 +1,27 @@
package com.alibaba.datax.plugin.writer.oceanbasev10writer.task; package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.util.DBUtil;
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.writer.oceanbasev10writer.Config;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
import com.alipay.oceanbase.obproxy.data.TableEntryKey;
import com.alipay.oceanbase.obproxy.util.ObPartitionIdCalculator;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection; import java.sql.Connection;
//import java.sql.PreparedStatement;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
import java.sql.SQLException; import java.sql.SQLException;
import java.util.ArrayList; import java.util.ArrayList;
@ -16,27 +36,7 @@ import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
import com.alibaba.datax.common.element.Column; //import java.sql.PreparedStatement;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.util.DBUtil;
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.writer.oceanbasev10writer.Config;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo;
import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
import com.alipay.oceanbase.obproxy.data.TableEntryKey;
import com.alipay.oceanbase.obproxy.util.ObPartitionIdCalculator;
public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task { public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
private static final Logger LOG = LoggerFactory.getLogger(ConcurrentTableWriterTask.class); private static final Logger LOG = LoggerFactory.getLogger(ConcurrentTableWriterTask.class);
@ -62,6 +62,7 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
private ObPartitionIdCalculator partCalculator = null; private ObPartitionIdCalculator partCalculator = null;
private HashMap<Long, List<Record>> groupInsertValues; private HashMap<Long, List<Record>> groupInsertValues;
List<Record> unknownPartRecords = new ArrayList<Record>();
// private List<Record> unknownPartRecords; // private List<Record> unknownPartRecords;
private List<Integer> partitionKeyIndexes; private List<Integer> partitionKeyIndexes;
@ -104,10 +105,14 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
connectInfo.getFullUserName(), connectInfo.password); connectInfo.getFullUserName(), connectInfo.password);
checkConnHolder.initConnection(); checkConnHolder.initConnection();
if (isOracleCompatibleMode) { if (isOracleCompatibleMode) {
connectInfo.databaseName = connectInfo.databaseName.toUpperCase(); connectInfo.databaseName = connectInfo.databaseName.toUpperCase();
table = table.toUpperCase(); //在转义的情况下不翻译
LOG.info(String.format("this is oracle compatible mode, change database to %s, table to %s", if (!(table.startsWith("\"") && table.endsWith("\""))) {
connectInfo.databaseName, table)); table = table.toUpperCase();
}
LOG.info(String.format("this is oracle compatible mode, change database to %s, table to %s",
connectInfo.databaseName, table));
} }
if (config.getBool(Config.USE_PART_CALCULATOR, Config.DEFAULT_USE_PART_CALCULATOR)) { if (config.getBool(Config.USE_PART_CALCULATOR, Config.DEFAULT_USE_PART_CALCULATOR)) {
@ -289,20 +294,15 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
} }
private void addLeftRecords() { private void addLeftRecords() {
//不需要刷新Cache，已经是最后一批数据了
for (List<Record> groupValues : groupInsertValues.values()) { for (List<Record> groupValues : groupInsertValues.values()) {
if (groupValues.size() > 0 ) { if (groupValues.size() > 0 ) {
int retry = 0; addRecordsToWriteQueue(groupValues);
while (true) {
try {
concurrentWriter.addBatchRecords(groupValues);
break;
} catch (InterruptedException e) {
retry++;
LOG.info("Concurrent table writer is interrupted, retry {}", retry);
}
}
} }
} }
if (unknownPartRecords.size() > 0) {
addRecordsToWriteQueue(unknownPartRecords);
}
} }
private void addRecordToCache(final Record record) { private void addRecordToCache(final Record record) {
@ -326,41 +326,40 @@ public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
} }
groupValues.add(record); groupValues.add(record);
if (groupValues.size() >= batchSize) { if (groupValues.size() >= batchSize) {
int i = 0; groupValues = addRecordsToWriteQueue(groupValues);
while (true) {
if (i > 0) {
LOG.info("retry add batch record the {} times", i);
}
try {
concurrentWriter.addBatchRecords(groupValues);
printEveryTime();
break;
} catch (InterruptedException e) {
LOG.info("Concurrent table writer is interrupted");
}
}
groupValues = new ArrayList<Record>(batchSize);
groupInsertValues.put(partId, groupValues); groupInsertValues.put(partId, groupValues);
} }
} else { } else {
LOG.warn("add unknown part record {}", record); LOG.debug("add unknown part record {}", record);
List<Record> unknownPartRecords = new ArrayList<Record>();
unknownPartRecords.add(record); unknownPartRecords.add(record);
int i = 0; if (unknownPartRecords.size() >= batchSize) {
while (true) { unknownPartRecords = addRecordsToWriteQueue(unknownPartRecords);
if (i > 0) {
LOG.info("retry add batch record the {} times", i);
}
try {
concurrentWriter.addBatchRecords(unknownPartRecords);
break;
} catch (InterruptedException e) {
LOG.info("Concurrent table writer is interrupted");
}
} }
} }
} }
/**
*
* @param records
* @return 返回一个新的Cache用于存储接下来的数据
*/
private List<Record> addRecordsToWriteQueue(List<Record> records) {
int i = 0;
while (true) {
if (i > 0) {
LOG.info("retry add batch record the {} times", i);
}
try {
concurrentWriter.addBatchRecords(records);
break;
} catch (InterruptedException e) {
i++;
LOG.info("Concurrent table writer is interrupted");
}
}
return new ArrayList<Record>(batchSize);
}
private void checkMemStore() { private void checkMemStore() {
Connection checkConn = checkConnHolder.reconnect(); Connection checkConn = checkConnHolder.reconnect();
long now = System.currentTimeMillis(); long now = System.currentTimeMillis();

52
oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/util/ObWriterUtils.java
View File

File diff suppressed because one or more lines are too long

8
oscarwriter/pom.xml
View File

@ -39,12 +39,18 @@
<artifactId>plugin-rdbms-util</artifactId> <artifactId>plugin-rdbms-util</artifactId>
<version>${datax-project-version}</version> <version>${datax-project-version}</version>
</dependency> </dependency>
<dependency> <!-- <dependency>
<groupId>com.oscar</groupId> <groupId>com.oscar</groupId>
<artifactId>oscar</artifactId> <artifactId>oscar</artifactId>
<version>7.0.8</version> <version>7.0.8</version>
<scope>system</scope> <scope>system</scope>
<systemPath>${basedir}/src/main/lib/oscarJDBC.jar</systemPath> <systemPath>${basedir}/src/main/lib/oscarJDBC.jar</systemPath>
</dependency>-->
<!--驱动文件找不到了，暂时替换成下面这个-->
<dependency>
<groupId>com.csicit.thirdparty</groupId>
<artifactId>oscar</artifactId>
<version>1.0.1</version>
</dependency> </dependency>
</dependencies> </dependencies>

10
otsreader/pom.xml
View File

@ -34,6 +34,16 @@
<groupId>com.aliyun.openservices</groupId> <groupId>com.aliyun.openservices</groupId>
<artifactId>ots-public</artifactId> <artifactId>ots-public</artifactId>
<version>2.2.4</version> <version>2.2.4</version>
<exclusions>
<exclusion>
<artifactId>log4j-api</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-core</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
</exclusions>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.google.code.gson</groupId> <groupId>com.google.code.gson</groupId>

10
otsstreamreader/pom.xml
View File

@ -33,6 +33,16 @@
<groupId>com.aliyun.openservices</groupId> <groupId>com.aliyun.openservices</groupId>
<artifactId>tablestore-streamclient</artifactId> <artifactId>tablestore-streamclient</artifactId>
<version>1.0.0</version> <version>1.0.0</version>
<exclusions>
<exclusion>
<artifactId>log4j-api</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-core</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
</exclusions>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.google.code.gson</groupId> <groupId>com.google.code.gson</groupId>

10
otswriter/pom.xml
View File

@ -34,6 +34,16 @@
<groupId>com.aliyun.openservices</groupId> <groupId>com.aliyun.openservices</groupId>
<artifactId>ots-public</artifactId> <artifactId>ots-public</artifactId>
<version>2.2.4</version> <version>2.2.4</version>
<exclusions>
<exclusion>
<artifactId>log4j-api</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-core</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
</exclusions>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.google.code.gson</groupId> <groupId>com.google.code.gson</groupId>

7
package.xml
View File

@ -371,6 +371,13 @@
</includes> </includes>
<outputDirectory>datax</outputDirectory> <outputDirectory>datax</outputDirectory>
</fileSet> </fileSet>
<fileSet>
<directory>tsdbreader/target/datax/</directory>
<includes>
<include>**/*.*</include>
</includes>
<outputDirectory>datax</outputDirectory>
</fileSet>
<fileSet> <fileSet>
<directory>adbpgwriter/target/datax/</directory> <directory>adbpgwriter/target/datax/</directory>
<includes> <includes>

11
pom.xml
View File

@ -184,6 +184,17 @@
<version>1.9.5</version> <version>1.9.5</version>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.17.1</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.17.1</version>
</dependency>
</dependencies> </dependencies>
</dependencyManagement> </dependencyManagement>

2
postgresqlwriter/src/main/java/com/alibaba/datax/plugin/writer/postgresqlwriter/PostgresqlWriter.java
View File

@ -67,6 +67,8 @@ public class PostgresqlWriter extends Writer {
public String calcValueHolder(String columnType){ public String calcValueHolder(String columnType){
if("serial".equalsIgnoreCase(columnType)){ if("serial".equalsIgnoreCase(columnType)){
return "?::int"; return "?::int";
}else if("bigserial".equalsIgnoreCase(columnType)){
return "?::int8";
}else if("bit".equalsIgnoreCase(columnType)){ }else if("bit".equalsIgnoreCase(columnType)){
return "?::bit varying"; return "?::bit varying";
} }

2
rdbmsreader/doc/rdbmsreader.md
View File

@ -138,7 +138,7 @@ RDBMSReader插件实现了从RDBMS读取数据。在底层实现上，RDBMSReade
**rdbmswriter如何增加新的数据库支持:** **rdbmswriter如何增加新的数据库支持:**
- 进入rdbmsreader对应目录，这里${DATAX_HOME}为DataX主目录，即: ${DATAX_HOME}/plugin/reader/rdbmswriter - 进入rdbmsreader对应目录，这里${DATAX_HOME}为DataX主目录，即: ${DATAX_HOME}/plugin/reader/rdbmsreader
- 在rdbmsreader插件目录下有plugin.json配置文件，在此文件中注册您具体的数据库驱动，具体放在drivers数组中。rdbmsreader插件在任务执行时会动态选择合适的数据库驱动连接数据库。 - 在rdbmsreader插件目录下有plugin.json配置文件，在此文件中注册您具体的数据库驱动，具体放在drivers数组中。rdbmsreader插件在任务执行时会动态选择合适的数据库驱动连接数据库。

16
tdenginewriter/doc/tdenginewriter-CN.md
View File

@ -232,14 +232,14 @@ TAGS(
#### 3.2.6 类型转换 #### 3.2.6 类型转换
| MongoDB 数据类型 | DataX 内部类型 | TDengine 数据类型 | | DataX 内部类型 | TDengine 数据类型 |
| ---------------- | -------------- | ----------------- | |-------------- | ----------------- |
| int, Long | Long | BIGINT | |Long | BIGINT |
| double | Double | DOUBLE | |Double | DOUBLE |
| string, array | String | NCHAR(64) | |String | NCHAR(64) |
| date | Date | TIMESTAMP | |Date | TIMESTAMP |
| boolean | Boolean | BOOL | |Boolean | BOOL |
| bytes | Bytes | BINARY(64) | |Bytes | BINARY(64) |
### 3.3 从关系型数据库到TDengine ### 3.3 从关系型数据库到TDengine
writer部分的配置规则和上述MongoDB的示例是一样的，这里给出一个MySQL的示例。 writer部分的配置规则和上述MongoDB的示例是一样的，这里给出一个MySQL的示例。

16
tdenginewriter/doc/tdenginewriter.md
View File

@ -228,14 +228,14 @@ Then the first columns received by this writer plugin must represent timestamp,
#### 3.2.6 Type Convert #### 3.2.6 Type Convert
| MongoDB Type | DataX Type | TDengine Type | |DataX Type | TDengine Type |
| ---------------- | -------------- | ----------------- | |-------------- | ----------------- |
| int, Long | Long | BIGINT | |Long | BIGINT |
| double | Double | DOUBLE | |Double | DOUBLE |
| string, array | String | NCHAR(64) | |String | NCHAR(64) |
| date | Date | TIMESTAMP | |Date | TIMESTAMP |
| boolean | Boolean | BOOL | |Boolean | BOOL |
| bytes | Bytes | BINARY(64) | |Bytes | BINARY(64) |
### 3.3 From Relational Database to TDengine ### 3.3 From Relational Database to TDengine

2
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/Constant.java
View File

@ -16,6 +16,8 @@ public final class Constant {
static final String DEFAULT_DATA_FORMAT = "yyyy-MM-dd HH:mm:ss"; static final String DEFAULT_DATA_FORMAT = "yyyy-MM-dd HH:mm:ss";
public static final String METRIC_SPECIFY_KEY = "__metric__"; public static final String METRIC_SPECIFY_KEY = "__metric__";
public static final String METRIC_SPECIFY_KEY_PREFIX = METRIC_SPECIFY_KEY + ".";
public static final int METRIC_SPECIFY_KEY_PREFIX_LENGTH = METRIC_SPECIFY_KEY_PREFIX.length();
public static final String TS_SPECIFY_KEY = "__ts__"; public static final String TS_SPECIFY_KEY = "__ts__";
public static final String VALUE_SPECIFY_KEY = "__value__"; public static final String VALUE_SPECIFY_KEY = "__value__";

5
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/Key.java
View File

@ -17,14 +17,19 @@ public class Key {
// RDB for MySQL / ADB etc. // RDB for MySQL / ADB etc.
static final String SINK_DB_TYPE = "sinkDbType"; static final String SINK_DB_TYPE = "sinkDbType";
static final String ENDPOINT = "endpoint"; static final String ENDPOINT = "endpoint";
static final String USERNAME = "username";
static final String PASSWORD = "password";
static final String COLUMN = "column"; static final String COLUMN = "column";
static final String METRIC = "metric"; static final String METRIC = "metric";
static final String FIELD = "field"; static final String FIELD = "field";
static final String TAG = "tag"; static final String TAG = "tag";
static final String COMBINE = "combine";
static final String INTERVAL_DATE_TIME = "splitIntervalMs"; static final String INTERVAL_DATE_TIME = "splitIntervalMs";
static final String BEGIN_DATE_TIME = "beginDateTime"; static final String BEGIN_DATE_TIME = "beginDateTime";
static final String END_DATE_TIME = "endDateTime"; static final String END_DATE_TIME = "endDateTime";
static final String HINT = "hint";
static final Boolean COMBINE_DEFAULT_VALUE = false;
static final Integer INTERVAL_DATE_TIME_DEFAULT_VALUE = 60; static final Integer INTERVAL_DATE_TIME_DEFAULT_VALUE = 60;
static final String TYPE_DEFAULT_VALUE = "TSDB"; static final String TYPE_DEFAULT_VALUE = "TSDB";
static final Set<String> TYPE_SET = new HashSet<>(); static final Set<String> TYPE_SET = new HashSet<>();

109
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/TSDBReader.java
View File

@ -60,6 +60,15 @@ public class TSDBReader extends Reader {
"The parameter [" + Key.ENDPOINT + "] is not set."); "The parameter [" + Key.ENDPOINT + "] is not set.");
} }
String username = originalConfig.getString(Key.USERNAME, null);
if (StringUtils.isBlank(username)) {
LOG.warn("The parameter [" + Key.USERNAME + "] is blank.");
}
String password = originalConfig.getString(Key.PASSWORD, null);
if (StringUtils.isBlank(password)) {
LOG.warn("The parameter [" + Key.PASSWORD + "] is blank.");
}
// tagK / field could be empty // tagK / field could be empty
if ("TSDB".equals(type)) { if ("TSDB".equals(type)) {
List<String> columns = originalConfig.getList(Key.COLUMN, String.class); List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
@ -76,7 +85,14 @@ public class TSDBReader extends Reader {
"The parameter [" + Key.COLUMN + "] is not set."); "The parameter [" + Key.COLUMN + "] is not set.");
} }
for (String specifyKey : Constant.MUST_CONTAINED_SPECIFY_KEYS) { for (String specifyKey : Constant.MUST_CONTAINED_SPECIFY_KEYS) {
if (!columns.contains(specifyKey)) { boolean containSpecifyKey = false;
for (String column : columns) {
if (column.startsWith(specifyKey)) {
containSpecifyKey = true;
break;
}
}
if (!containSpecifyKey) {
throw DataXException.asDataXException( throw DataXException.asDataXException(
TSDBReaderErrorCode.ILLEGAL_VALUE, TSDBReaderErrorCode.ILLEGAL_VALUE,
"The parameter [" + Key.COLUMN + "] should contain " "The parameter [" + Key.COLUMN + "] should contain "
@ -99,6 +115,8 @@ public class TSDBReader extends Reader {
"The parameter [" + Key.INTERVAL_DATE_TIME + "] should be great than zero."); "The parameter [" + Key.INTERVAL_DATE_TIME + "] should be great than zero.");
} }
Boolean isCombine = originalConfig.getBool(Key.COMBINE, Key.COMBINE_DEFAULT_VALUE);
SimpleDateFormat format = new SimpleDateFormat(Constant.DEFAULT_DATA_FORMAT); SimpleDateFormat format = new SimpleDateFormat(Constant.DEFAULT_DATA_FORMAT);
String startTime = originalConfig.getString(Key.BEGIN_DATE_TIME); String startTime = originalConfig.getString(Key.BEGIN_DATE_TIME);
Long startDate; Long startDate;
@ -168,14 +186,14 @@ public class TSDBReader extends Reader {
startTime = format.parse(originalConfig.getString(Key.BEGIN_DATE_TIME)).getTime(); startTime = format.parse(originalConfig.getString(Key.BEGIN_DATE_TIME)).getTime();
} catch (ParseException e) { } catch (ParseException e) {
throw DataXException.asDataXException( throw DataXException.asDataXException(
TSDBReaderErrorCode.ILLEGAL_VALUE, "解析[" + Key.BEGIN_DATE_TIME + "]失败.", e); TSDBReaderErrorCode.ILLEGAL_VALUE, "Analysis [" + Key.BEGIN_DATE_TIME + "] failed.", e);
} }
long endTime; long endTime;
try { try {
endTime = format.parse(originalConfig.getString(Key.END_DATE_TIME)).getTime(); endTime = format.parse(originalConfig.getString(Key.END_DATE_TIME)).getTime();
} catch (ParseException e) { } catch (ParseException e) {
throw DataXException.asDataXException( throw DataXException.asDataXException(
TSDBReaderErrorCode.ILLEGAL_VALUE, "解析[" + Key.END_DATE_TIME + "]失败.", e); TSDBReaderErrorCode.ILLEGAL_VALUE, "Analysis [" + Key.END_DATE_TIME + "] failed.", e);
} }
if (TimeUtils.isSecond(startTime)) { if (TimeUtils.isSecond(startTime)) {
startTime *= 1000; startTime *= 1000;
@ -186,13 +204,14 @@ public class TSDBReader extends Reader {
DateTime startDateTime = new DateTime(TimeUtils.getTimeInHour(startTime)); DateTime startDateTime = new DateTime(TimeUtils.getTimeInHour(startTime));
DateTime endDateTime = new DateTime(TimeUtils.getTimeInHour(endTime)); DateTime endDateTime = new DateTime(TimeUtils.getTimeInHour(endTime));
final Boolean isCombine = originalConfig.getBool(Key.COMBINE, Key.COMBINE_DEFAULT_VALUE);
if ("TSDB".equals(type)) { if ("TSDB".equals(type)) {
// split by metric if (isCombine) {
for (String column : columns4TSDB) {
// split by time in hour // split by time in hour
while (startDateTime.isBefore(endDateTime)) { while (startDateTime.isBefore(endDateTime)) {
Configuration clone = this.originalConfig.clone(); Configuration clone = this.originalConfig.clone();
clone.set(Key.COLUMN, Collections.singletonList(column)); clone.set(Key.COLUMN, columns4TSDB);
clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis()); clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
startDateTime = startDateTime.plusMillis(splitIntervalMs); startDateTime = startDateTime.plusMillis(splitIntervalMs);
@ -202,15 +221,30 @@ public class TSDBReader extends Reader {
LOG.info("Configuration: {}", JSON.toJSONString(clone)); LOG.info("Configuration: {}", JSON.toJSONString(clone));
} }
} else {
// split by time in hour
while (startDateTime.isBefore(endDateTime)) {
// split by metric
for (String column : columns4TSDB) {
Configuration clone = this.originalConfig.clone();
clone.set(Key.COLUMN, Collections.singletonList(column));
clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
startDateTime = startDateTime.plusMillis(splitIntervalMs);
// Make sure the time interval is [start, end).
clone.set(Key.END_DATE_TIME, startDateTime.getMillis() - 1);
configurations.add(clone);
LOG.info("Configuration: {}", JSON.toJSONString(clone));
}
}
} }
} else { } else {
// split by metric if (isCombine) {
for (String metric : metrics) {
// split by time in hour
while (startDateTime.isBefore(endDateTime)) { while (startDateTime.isBefore(endDateTime)) {
Configuration clone = this.originalConfig.clone(); Configuration clone = this.originalConfig.clone();
clone.set(Key.COLUMN, columns4RDB); clone.set(Key.COLUMN, columns4RDB);
clone.set(Key.METRIC, Collections.singletonList(metric)); clone.set(Key.METRIC, metrics);
clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis()); clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
startDateTime = startDateTime.plusMillis(splitIntervalMs); startDateTime = startDateTime.plusMillis(splitIntervalMs);
@ -220,6 +254,24 @@ public class TSDBReader extends Reader {
LOG.info("Configuration: {}", JSON.toJSONString(clone)); LOG.info("Configuration: {}", JSON.toJSONString(clone));
} }
} else {
// split by time in hour
while (startDateTime.isBefore(endDateTime)) {
// split by metric
for (String metric : metrics) {
Configuration clone = this.originalConfig.clone();
clone.set(Key.COLUMN, columns4RDB);
clone.set(Key.METRIC, Collections.singletonList(metric));
clone.set(Key.BEGIN_DATE_TIME, startDateTime.getMillis());
startDateTime = startDateTime.plusMillis(splitIntervalMs);
// Make sure the time interval is [start, end).
clone.set(Key.END_DATE_TIME, startDateTime.getMillis() - 1);
configurations.add(clone);
LOG.info("Configuration: {}", JSON.toJSONString(clone));
}
}
} }
} }
return configurations; return configurations;
@ -247,6 +299,8 @@ public class TSDBReader extends Reader {
private TSDBConnection conn; private TSDBConnection conn;
private Long startTime; private Long startTime;
private Long endTime; private Long endTime;
private Boolean isCombine;
private Map<String, Object> hint;
@Override @Override
public void init() { public void init() {
@ -265,11 +319,16 @@ public class TSDBReader extends Reader {
this.tags = readerSliceConfig.getMap(Key.TAG); this.tags = readerSliceConfig.getMap(Key.TAG);
String address = readerSliceConfig.getString(Key.ENDPOINT); String address = readerSliceConfig.getString(Key.ENDPOINT);
String username = readerSliceConfig.getString(Key.USERNAME);
String password = readerSliceConfig.getString(Key.PASSWORD);
conn = new TSDBConnection(address); conn = new TSDBConnection(address, username, password);
this.startTime = readerSliceConfig.getLong(Key.BEGIN_DATE_TIME); this.startTime = readerSliceConfig.getLong(Key.BEGIN_DATE_TIME);
this.endTime = readerSliceConfig.getLong(Key.END_DATE_TIME); this.endTime = readerSliceConfig.getLong(Key.END_DATE_TIME);
this.isCombine = readerSliceConfig.getBool(Key.COMBINE, Key.COMBINE_DEFAULT_VALUE);
this.hint = readerSliceConfig.getMap(Key.HINT);
} }
@Override @Override
@ -283,29 +342,35 @@ public class TSDBReader extends Reader {
if ("TSDB".equals(type)) { if ("TSDB".equals(type)) {
for (String metric : columns4TSDB) { for (String metric : columns4TSDB) {
final Map<String, String> tags = this.tags == null ? final Map<String, String> tags = this.tags == null ?
null : (Map<String, String>) this.tags.get(metric); null : (Map<String, String>) this.tags.get(metric);
if (fields == null || !fields.containsKey(metric)) { if (fields == null || !fields.containsKey(metric)) {
conn.sendDPs(metric, tags, this.startTime, this.endTime, recordSender); conn.sendDPs(metric, tags, this.startTime, this.endTime, recordSender, hint);
} else { } else {
conn.sendDPs(metric, (List<String>) fields.get(metric), conn.sendDPs(metric, (List<String>) fields.get(metric),
tags, this.startTime, this.endTime, recordSender); tags, this.startTime, this.endTime, recordSender, hint);
} }
} }
} else { } else {
for (String metric : metrics) { if (isCombine) {
final Map<String, String> tags = this.tags == null ? final Map<String, String> tags = this.tags == null ?
null : (Map<String, String>) this.tags.get(metric); null : (Map<String, String>) this.tags.get(metrics.get(0));
if (fields == null || !fields.containsKey(metric)) { conn.sendRecords(metrics, tags, startTime, endTime, columns4RDB, recordSender, hint);
conn.sendRecords(metric, tags, startTime, endTime, columns4RDB, recordSender); } else {
} else { for (String metric : metrics) {
conn.sendRecords(metric, (List<String>) fields.get(metric), final Map<String, String> tags = this.tags == null ?
tags, startTime, endTime, columns4RDB, recordSender); null : (Map<String, String>) this.tags.get(metric);
if (fields == null || !fields.containsKey(metric)) {
conn.sendRecords(metric, tags, startTime, endTime, columns4RDB, isCombine, recordSender, hint);
} else {
conn.sendRecords(metric, (List<String>) fields.get(metric),
tags, startTime, endTime, columns4RDB, recordSender, hint);
}
} }
} }
} }
} catch (Exception e) { } catch (Exception e) {
throw DataXException.asDataXException( throw DataXException.asDataXException(
TSDBReaderErrorCode.ILLEGAL_VALUE, "获取或发送数据点的过程中出错！", e); TSDBReaderErrorCode.ILLEGAL_VALUE, "Error in getting or sending data point！", e);
} }
} }

27
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/Connection4TSDB.java
View File

@ -22,6 +22,20 @@ public interface Connection4TSDB {
*/ */
String address(); String address();
/**
* Get the address of Database.
*
* @return host+ip
*/
String username();
/**
* Get the address of Database.
*
* @return host+ip
*/
String password();
/** /**
* Get the version of Database. * Get the version of Database.
* *
@ -46,22 +60,27 @@ public interface Connection4TSDB {
/** /**
* Send data points for TSDB with single field. * Send data points for TSDB with single field.
*/ */
void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception; void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception;
/** /**
* Send data points for TSDB with multi fields. * Send data points for TSDB with multi fields.
*/ */
void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception; void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception;
/** /**
* Send data points for RDB with single field. * Send data points for RDB with single field.
*/ */
void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception; void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, Boolean isCombine, RecordSender recordSender, Map<String, Object> hint) throws Exception;
/** /**
* Send data points for RDB with multi fields. * Send data points for RDB with multi fields.
*/ */
void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception; void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception;
/**
* Send data points for RDB with single fields on combine mode.
*/
void sendRecords(List<String> metrics, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception;
/** /**
* Put data point. * Put data point.

41
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/TSDBConnection.java
View File

@ -19,9 +19,13 @@ import java.util.Map;
public class TSDBConnection implements Connection4TSDB { public class TSDBConnection implements Connection4TSDB {
private String address; private String address;
private String username;
private String password;
public TSDBConnection(String address) { public TSDBConnection(String address, String username, String password) {
this.address = address; this.address = address;
this.username = username;
this.password = password;
} }
@Override @Override
@ -29,14 +33,24 @@ public class TSDBConnection implements Connection4TSDB {
return address; return address;
} }
@Override
public String username() {
return username;
}
@Override
public String password() {
return password;
}
@Override @Override
public String version() { public String version() {
return TSDBUtils.version(address); return TSDBUtils.version(address, username, password);
} }
@Override @Override
public String config() { public String config() {
return TSDBUtils.config(address); return TSDBUtils.config(address, username, password);
} }
@Override @Override
@ -45,23 +59,28 @@ public class TSDBConnection implements Connection4TSDB {
} }
@Override @Override
public void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception { public void sendDPs(String metric, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception {
TSDBDump.dump4TSDB(this, metric, tags, start, end, recordSender); TSDBDump.dump4TSDB(this, metric, tags, start, end, recordSender, hint);
} }
@Override @Override
public void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender) throws Exception { public void sendDPs(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, RecordSender recordSender, Map<String, Object> hint) throws Exception {
TSDBDump.dump4TSDB(this, metric, fields, tags, start, end, recordSender); TSDBDump.dump4TSDB(this, metric, fields, tags, start, end, recordSender, hint);
} }
@Override @Override
public void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception { public void sendRecords(String metric, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, Boolean isCombine, RecordSender recordSender, Map<String, Object> hint) throws Exception {
TSDBDump.dump4RDB(this, metric, tags, start, end, columns4RDB, recordSender); TSDBDump.dump4RDB(this, metric, tags, start, end, columns4RDB, recordSender, hint);
} }
@Override @Override
public void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender) throws Exception { public void sendRecords(List<String> metrics, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception {
TSDBDump.dump4RDB(this, metric, fields, tags, start, end, columns4RDB, recordSender); TSDBDump.dump4RDB(this, metrics, tags, start, end, columns4RDB, recordSender, hint);
}
@Override
public void sendRecords(String metric, List<String> fields, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender recordSender, Map<String, Object> hint) throws Exception {
TSDBDump.dump4RDB(this, metric, fields, tags, start, end, columns4RDB, recordSender, hint);
} }
@Override @Override

96
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/TSDBDump.java
View File

@ -9,10 +9,9 @@ import com.alibaba.fastjson.parser.Feature;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.HashMap; import java.util.*;
import java.util.LinkedList;
import java.util.List; import static com.alibaba.datax.plugin.reader.tsdbreader.Constant.METRIC_SPECIFY_KEY_PREFIX_LENGTH;
import java.util.Map;
/** /**
* Copyright @ 2019 alibaba.com * Copyright @ 2019 alibaba.com
@ -37,10 +36,10 @@ final class TSDBDump {
} }
static void dump4TSDB(TSDBConnection conn, String metric, Map<String, String> tags, static void dump4TSDB(TSDBConnection conn, String metric, Map<String, String> tags,
Long start, Long end, RecordSender sender) throws Exception { Long start, Long end, RecordSender sender, Map<String, Object> hint) throws Exception {
LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end); LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
String res = queryRange4SingleField(conn, metric, tags, start, end); String res = queryRange4SingleField(conn, metric, tags, start, end, hint);
List<String> dps = getDps4TSDB(metric, res); List<String> dps = getDps4TSDB(metric, res);
if (dps == null || dps.isEmpty()) { if (dps == null || dps.isEmpty()) {
return; return;
@ -49,10 +48,10 @@ final class TSDBDump {
} }
static void dump4TSDB(TSDBConnection conn, String metric, List<String> fields, Map<String, String> tags, static void dump4TSDB(TSDBConnection conn, String metric, List<String> fields, Map<String, String> tags,
Long start, Long end, RecordSender sender) throws Exception { Long start, Long end, RecordSender sender, Map<String, Object> hint) throws Exception {
LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end); LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
String res = queryRange4MultiFields(conn, metric, fields, tags, start, end); String res = queryRange4MultiFields(conn, metric, fields, tags, start, end, hint);
List<String> dps = getDps4TSDB(metric, fields, res); List<String> dps = getDps4TSDB(metric, fields, res);
if (dps == null || dps.isEmpty()) { if (dps == null || dps.isEmpty()) {
return; return;
@ -61,10 +60,10 @@ final class TSDBDump {
} }
static void dump4RDB(TSDBConnection conn, String metric, Map<String, String> tags, static void dump4RDB(TSDBConnection conn, String metric, Map<String, String> tags,
Long start, Long end, List<String> columns4RDB, RecordSender sender) throws Exception { Long start, Long end, List<String> columns4RDB, RecordSender sender, Map<String, Object> hint) throws Exception {
LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end); LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
String res = queryRange4SingleField(conn, metric, tags, start, end); String res = queryRange4SingleField(conn, metric, tags, start, end, hint);
List<DataPoint4TSDB> dps = getDps4RDB(metric, res); List<DataPoint4TSDB> dps = getDps4RDB(metric, res);
if (dps == null || dps.isEmpty()) { if (dps == null || dps.isEmpty()) {
return; return;
@ -92,12 +91,71 @@ final class TSDBDump {
} }
} }
public static void dump4RDB(TSDBConnection conn, List<String> metrics, Map<String, String> tags, Long start, Long end, List<String> columns4RDB, RecordSender sender, Map<String, Object> hint) throws Exception {
LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metrics, start, end);
List<DataPoint4TSDB> dps = new LinkedList<>();
for (String metric : metrics) {
String res = queryRange4SingleField(conn, metric, tags, start, end, hint);
final List<DataPoint4TSDB> dpList = getDps4RDB(metric, res);
if (dpList == null || dpList.isEmpty()) {
continue;
}
dps.addAll(dpList);
}
if (dps.isEmpty()) {
return;
}
Map<Long, Map<String, DataPoint4TSDB>> dpsCombinedByTs = new LinkedHashMap<>();
for (DataPoint4TSDB dp : dps) {
final long ts = dp.getTimestamp();
final Map<String, DataPoint4TSDB> dpsWithSameTs = dpsCombinedByTs.computeIfAbsent(ts, k -> new LinkedHashMap<>());
dpsWithSameTs.put(dp.getMetric(), dp);
}
for (Map.Entry<Long, Map<String, DataPoint4TSDB>> entry : dpsCombinedByTs.entrySet()) {
final Long ts = entry.getKey();
final Map<String, DataPoint4TSDB> metricAndDps = entry.getValue();
final Record record = sender.createRecord();
DataPoint4TSDB tmpDp = null;
for (final String column : columns4RDB) {
if (column.startsWith(Constant.METRIC_SPECIFY_KEY)) {
final String m = column.substring(METRIC_SPECIFY_KEY_PREFIX_LENGTH);
tmpDp = metricAndDps.get(m);
if (tmpDp == null) {
continue;
}
record.addColumn(getColumn(tmpDp.getValue()));
} else if (Constant.TS_SPECIFY_KEY.equals(column)) {
record.addColumn(new LongColumn(ts));
} else if (Constant.VALUE_SPECIFY_KEY.equals(column)) {
// combine 模式下，不应该定义 __value__ 字段，因为 __metric__.xxx 字段会输出对应的 value 值
throw new RuntimeException("The " + Constant.VALUE_SPECIFY_KEY +
" column should not be specified in combine mode!");
} else {
// combine 模式下，应该确保 __metric__.xxx 字段的定义，放在 column 数组的最前面，以保证获取到 metric
if (tmpDp == null) {
throw new RuntimeException("These " + Constant.METRIC_SPECIFY_KEY_PREFIX +
" column should be placed first in the column array in combine mode!");
}
final Object tagv = tmpDp.getTags().get(column);
if (tagv == null) {
continue;
}
record.addColumn(getColumn(tagv));
}
}
sender.sendToWriter(record);
}
}
static void dump4RDB(TSDBConnection conn, String metric, List<String> fields, static void dump4RDB(TSDBConnection conn, String metric, List<String> fields,
Map<String, String> tags, Long start, Long end, Map<String, String> tags, Long start, Long end,
List<String> columns4RDB, RecordSender sender) throws Exception { List<String> columns4RDB, RecordSender sender, Map<String, Object> hint) throws Exception {
LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end); LOG.info("conn address: {}, metric: {}, start: {}, end: {}", conn.address(), metric, start, end);
String res = queryRange4MultiFields(conn, metric, fields, tags, start, end); String res = queryRange4MultiFields(conn, metric, fields, tags, start, end, hint);
List<DataPoint4TSDB> dps = getDps4RDB(metric, fields, res); List<DataPoint4TSDB> dps = getDps4RDB(metric, fields, res);
if (dps == null || dps.isEmpty()) { if (dps == null || dps.isEmpty()) {
return; return;
@ -131,14 +189,16 @@ final class TSDBDump {
valueColumn = new LongColumn((Long) value); valueColumn = new LongColumn((Long) value);
} else if (value instanceof String) { } else if (value instanceof String) {
valueColumn = new StringColumn((String) value); valueColumn = new StringColumn((String) value);
} else if (value instanceof Integer) {
valueColumn = new LongColumn(((Integer)value).longValue());
} else { } else {
throw new Exception(String.format("value 不支持类型: [%s]", value.getClass().getSimpleName())); throw new Exception(String.format("value not supported type: [%s]", value.getClass().getSimpleName()));
} }
return valueColumn; return valueColumn;
} }
private static String queryRange4SingleField(TSDBConnection conn, String metric, Map<String, String> tags, private static String queryRange4SingleField(TSDBConnection conn, String metric, Map<String, String> tags,
Long start, Long end) throws Exception { Long start, Long end, Map<String, Object> hint) throws Exception {
String tagKV = getFilterByTags(tags); String tagKV = getFilterByTags(tags);
String body = "{\n" + String body = "{\n" +
" \"start\": " + start + ",\n" + " \"start\": " + start + ",\n" +
@ -148,14 +208,15 @@ final class TSDBDump {
" \"aggregator\": \"none\",\n" + " \"aggregator\": \"none\",\n" +
" \"metric\": \"" + metric + "\"\n" + " \"metric\": \"" + metric + "\"\n" +
(tagKV == null ? "" : tagKV) + (tagKV == null ? "" : tagKV) +
(hint == null ? "" : (", \"hint\": " + JSON.toJSONString(hint))) +
" }\n" + " }\n" +
" ]\n" + " ]\n" +
"}"; "}";
return HttpUtils.post(conn.address() + QUERY, body); return HttpUtils.post(conn.address() + QUERY, conn.username(), conn.password(), body);
} }
private static String queryRange4MultiFields(TSDBConnection conn, String metric, List<String> fields, private static String queryRange4MultiFields(TSDBConnection conn, String metric, List<String> fields,
Map<String, String> tags, Long start, Long end) throws Exception { Map<String, String> tags, Long start, Long end, Map<String, Object> hint) throws Exception {
// fields // fields
StringBuilder fieldBuilder = new StringBuilder(); StringBuilder fieldBuilder = new StringBuilder();
fieldBuilder.append("\"fields\":["); fieldBuilder.append("\"fields\":[");
@ -177,10 +238,11 @@ final class TSDBDump {
" \"metric\": \"" + metric + "\",\n" + " \"metric\": \"" + metric + "\",\n" +
fieldBuilder.toString() + fieldBuilder.toString() +
(tagKV == null ? "" : tagKV) + (tagKV == null ? "" : tagKV) +
(hint == null ? "" : (", \"hint\": " + JSON.toJSONString(hint))) +
" }\n" + " }\n" +
" ]\n" + " ]\n" +
"}"; "}";
return HttpUtils.post(conn.address() + QUERY_MULTI_FIELD, body); return HttpUtils.post(conn.address() + QUERY_MULTI_FIELD, conn.username(), conn.password(), body);
} }
private static String getFilterByTags(Map<String, String> tags) { private static String getFilterByTags(Map<String, String> tags) {

50
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/util/HttpUtils.java
View File

@ -1,11 +1,13 @@
package com.alibaba.datax.plugin.reader.tsdbreader.util; package com.alibaba.datax.plugin.reader.tsdbreader.util;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.client.fluent.Content; import org.apache.http.client.fluent.Content;
import org.apache.http.client.fluent.Request; import org.apache.http.client.fluent.Request;
import org.apache.http.entity.ContentType; import org.apache.http.entity.ContentType;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.Map; import java.util.Map;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -22,13 +24,18 @@ public final class HttpUtils {
public final static int CONNECT_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60); public final static int CONNECT_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
public final static int SOCKET_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60); public final static int SOCKET_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
private static final String CREDENTIALS_FORMAT = "%s:%s";
private static final String BASIC_AUTHENTICATION_FORMAT = "Basic %s";
private HttpUtils() { private HttpUtils() {
} }
public static String get(String url) throws Exception { public static String get(String url, String username, String password) throws Exception {
Content content = Request.Get(url) final Request request = Request.Get(url)
.connectTimeout(CONNECT_TIMEOUT_DEFAULT_IN_MILL) .connectTimeout(CONNECT_TIMEOUT_DEFAULT_IN_MILL)
.socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL) .socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL);
addAuth(request, username, password);
Content content = request
.execute() .execute()
.returnContent(); .returnContent();
if (content == null) { if (content == null) {
@ -37,24 +44,21 @@ public final class HttpUtils {
return content.asString(StandardCharsets.UTF_8); return content.asString(StandardCharsets.UTF_8);
} }
public static String post(String url, Map<String, Object> params) throws Exception { public static String post(String url, String username, String password, Map<String, Object> params) throws Exception {
return post(url, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL); return post(url, username, password, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
} }
public static String post(String url, String params) throws Exception { public static String post(String url, String username, String password, String params) throws Exception {
return post(url, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL); return post(url, username, password, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
} }
public static String post(String url, Map<String, Object> params, public static String post(String url, String username, String password, String params,
int connectTimeoutInMill, int socketTimeoutInMill) throws Exception { int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
return post(url, JSON.toJSONString(params), connectTimeoutInMill, socketTimeoutInMill); Request request = Request.Post(url)
}
public static String post(String url, String params,
int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
Content content = Request.Post(url)
.connectTimeout(connectTimeoutInMill) .connectTimeout(connectTimeoutInMill)
.socketTimeout(socketTimeoutInMill) .socketTimeout(socketTimeoutInMill);
addAuth(request, username, password);
Content content = request
.addHeader("Content-Type", "application/json") .addHeader("Content-Type", "application/json")
.bodyString(params, ContentType.APPLICATION_JSON) .bodyString(params, ContentType.APPLICATION_JSON)
.execute() .execute()
@ -64,4 +68,20 @@ public final class HttpUtils {
} }
return content.asString(StandardCharsets.UTF_8); return content.asString(StandardCharsets.UTF_8);
} }
private static void addAuth(Request request, String username, String password) {
String authorization = generateHttpAuthorization(username, password);
if (authorization != null) {
request.setHeader("Authorization", authorization);
}
}
private static String generateHttpAuthorization(String username, String password) {
if (StringUtils.isBlank(username) || StringUtils.isBlank(password)) {
return null;
}
String credentials = String.format(CREDENTIALS_FORMAT, username, password);
credentials = Base64.getEncoder().encodeToString(credentials.getBytes());
return String.format(BASIC_AUTHENTICATION_FORMAT, credentials);
}
} }

38
tsdbreader/src/main/java/com/alibaba/datax/plugin/reader/tsdbreader/util/TSDBUtils.java
View File

@ -1,11 +1,5 @@
package com.alibaba.datax.plugin.reader.tsdbreader.util; package com.alibaba.datax.plugin.reader.tsdbreader.util;
import com.alibaba.datax.plugin.reader.tsdbreader.conn.DataPoint4TSDB;
import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
/** /**
* Copyright @ 2019 alibaba.com * Copyright @ 2019 alibaba.com
@ -17,52 +11,28 @@ import java.util.List;
*/ */
public final class TSDBUtils { public final class TSDBUtils {
private static final Logger LOGGER = LoggerFactory.getLogger(TSDBUtils.class);
private TSDBUtils() { private TSDBUtils() {
} }
public static String version(String address) { public static String version(String address, String username, String password) {
String url = String.format("%s/api/version", address); String url = String.format("%s/api/version", address);
String rsp; String rsp;
try { try {
rsp = HttpUtils.get(url); rsp = HttpUtils.get(url, username, password);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
return rsp; return rsp;
} }
public static String config(String address) { public static String config(String address, String username, String password) {
String url = String.format("%s/api/config", address); String url = String.format("%s/api/config", address);
String rsp; String rsp;
try { try {
rsp = HttpUtils.get(url); rsp = HttpUtils.get(url, username, password);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
return rsp; return rsp;
} }
public static boolean put(String address, List<DataPoint4TSDB> dps) {
return put(address, JSON.toJSON(dps));
}
public static boolean put(String address, DataPoint4TSDB dp) {
return put(address, JSON.toJSON(dp));
}
private static boolean put(String address, Object o) {
String url = String.format("%s/api/put", address);
String rsp;
try {
rsp = HttpUtils.post(url, o.toString());
// If successful, the returned content should be null.
assert rsp == null;
} catch (Exception e) {
LOGGER.error("Address: {}, DataPoints: {}", url, o);
throw new RuntimeException(e);
}
return true;
}
} }

4
tsdbreader/src/test/java/com/alibaba/datax/plugin/reader/tsdbreader/conn/TSDBConnectionTest.java
View File

@ -19,12 +19,12 @@ public class TSDBConnectionTest {
@Test @Test
public void testVersion() { public void testVersion() {
String version = new TSDBConnection(TSDB_ADDRESS).version(); String version = new TSDBConnection(TSDB_ADDRESS,null,null).version();
Assert.assertNotNull(version); Assert.assertNotNull(version);
} }
@Test @Test
public void testIsSupported() { public void testIsSupported() {
Assert.assertTrue(new TSDBConnection(TSDB_ADDRESS).isSupported()); Assert.assertTrue(new TSDBConnection(TSDB_ADDRESS,null,null).isSupported());
} }
} }

39
tsdbreader/src/test/java/com/alibaba/datax/plugin/reader/tsdbreader/util/HttpUtilsTest.java
View File

@ -1,39 +0,0 @@
package com.alibaba.datax.plugin.reader.tsdbreader.util;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import java.util.HashMap;
import java.util.Map;
/**
* Copyright @ 2019 alibaba.com
* All right reserved.
* Function：HttpUtils Test
*
* @author Benedict Jin
* @since 2019-10-21
*/
@Ignore
public class HttpUtilsTest {
@Test
public void testSimpleCase() throws Exception {
String url = "https://httpbin.org/post";
Map<String, Object> params = new HashMap<>();
params.put("foo", "bar");
String rsp = HttpUtils.post(url, params);
System.out.println(rsp);
Assert.assertNotNull(rsp);
}
@Test
public void testGet() throws Exception {
String url = String.format("%s/api/version", Const.TSDB_ADDRESS);
String rsp = HttpUtils.get(url);
System.out.println(rsp);
Assert.assertNotNull(rsp);
}
}

7
tsdbwriter/pom.xml
View File

@ -91,6 +91,13 @@
<version>${fastjson.version}</version> <version>${fastjson.version}</version>
</dependency> </dependency>
<!-- tsdb -->
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>hitsdb-client</artifactId>
<version>0.4.0-SNAPSHOT</version>
</dependency>
<!-- test --> <!-- test -->
<dependency> <dependency>
<groupId>junit</groupId> <groupId>junit</groupId>

32
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/conn/Connection4TSDB.java
View File

@ -21,6 +21,28 @@ public interface Connection4TSDB {
*/ */
String address(); String address();
/**
* Get the setted database name.
*
* @return database
*/
String database();
/**
* Get the username of Database.
*
* @return username
*/
String username();
/**
* Get the password of Database.
*
* @return password
*/
String password();
/** /**
* Get the version of Database. * Get the version of Database.
* *
@ -69,13 +91,21 @@ public interface Connection4TSDB {
boolean put(List<DataPoint4TSDB> dps); boolean put(List<DataPoint4TSDB> dps);
/** /**
* Put data points. * Put data points with single field.
* *
* @param dps data points * @param dps data points
* @return whether the data point is written successfully * @return whether the data point is written successfully
*/ */
boolean put(String dps); boolean put(String dps);
/**
* Put data points with multi fields.
*
* @param dps data points
* @return whether the data point is written successfully
*/
boolean mput(String dps);
/** /**
* Whether current version is supported. * Whether current version is supported.
* *

38
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/conn/TSDBConnection.java
View File

@ -18,12 +18,18 @@ import java.util.List;
public class TSDBConnection implements Connection4TSDB { public class TSDBConnection implements Connection4TSDB {
private String address; private String address;
private String username;
private String password;
private String database;
public TSDBConnection(String address) { public TSDBConnection(String address, String database, String username, String password) {
if (StringUtils.isBlank(address)) { if (StringUtils.isBlank(address)) {
throw new RuntimeException("TSDBConnection init failed because address is blank!"); throw new RuntimeException("TSDBConnection init failed because address is blank!");
} }
this.address = address; this.address = address;
this.database = database;
this.username = username;
this.password = password;
} }
@Override @Override
@ -31,14 +37,29 @@ public class TSDBConnection implements Connection4TSDB {
return address; return address;
} }
@Override
public String username() {
return username;
}
@Override
public String database() {
return database;
}
@Override
public String password() {
return password;
}
@Override @Override
public String version() { public String version() {
return TSDBUtils.version(address); return TSDBUtils.version(address, username, password);
} }
@Override @Override
public String config() { public String config() {
return TSDBUtils.config(address); return TSDBUtils.config(address, username, password);
} }
@Override @Override
@ -53,17 +74,22 @@ public class TSDBConnection implements Connection4TSDB {
@Override @Override
public boolean put(DataPoint4TSDB dp) { public boolean put(DataPoint4TSDB dp) {
return TSDBUtils.put(address, dp); return TSDBUtils.put(address, database, username, password, dp);
} }
@Override @Override
public boolean put(List<DataPoint4TSDB> dps) { public boolean put(List<DataPoint4TSDB> dps) {
return TSDBUtils.put(address, dps); return TSDBUtils.put(address, database, username, password, dps);
} }
@Override @Override
public boolean put(String dps) { public boolean put(String dps) {
return TSDBUtils.put(address, dps); return TSDBUtils.put(address, database, username, password, dps);
}
@Override
public boolean mput(String dps) {
return TSDBUtils.mput(address, database, username, password, dps);
} }
@Override @Override

16
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/Key.java
View File

@ -10,8 +10,22 @@ package com.alibaba.datax.plugin.writer.tsdbwriter;
*/ */
public class Key { public class Key {
static final String SOURCE_DB_TYPE = "sourceDbType";
static final String MULTI_FIELD = "multiField";
// common
static final String ENDPOINT = "endpoint"; static final String ENDPOINT = "endpoint";
static final String USERNAME = "username";
static final String PASSWORD = "password";
static final String IGNORE_WRITE_ERROR = "ignoreWriteError";
static final String DATABASE = "database";
// for tsdb
static final String BATCH_SIZE = "batchSize"; static final String BATCH_SIZE = "batchSize";
static final String MAX_RETRY_TIME = "maxRetryTime"; static final String MAX_RETRY_TIME = "maxRetryTime";
static final String IGNORE_WRITE_ERROR = "ignoreWriteError";
// for rdb
static final String COLUMN = "column";
static final String COLUMN_TYPE = "columnType";
static final String TABLE = "table";
} }

5
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/SourceDBType.java Normal file
View File

@ -0,0 +1,5 @@
package com.alibaba.datax.plugin.writer.tsdbwriter;
public enum SourceDBType {
TSDB, RDB
}

96
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBConverter.java Normal file
View File

@ -0,0 +1,96 @@
package com.alibaba.datax.plugin.writer.tsdbwriter;
import com.alibaba.datax.common.element.Column;
import com.alibaba.datax.common.element.Record;
import com.alibaba.fastjson.JSON;
import com.aliyun.hitsdb.client.value.request.MultiFieldPoint;
import com.aliyun.hitsdb.client.value.request.Point;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
class TSDBConverter {
private static final Logger LOG = LoggerFactory.getLogger(TSDBConverter.class);
private List<String> columnName;
private List<String> columnType;
TSDBConverter(List<String> columnName, List<String> columnType) {
this.columnName = columnName;
this.columnType = columnType;
LOG.info("columnName: {}, columnType: {}", JSON.toJSONString(columnName), JSON.toJSONString(columnType));
}
List<Point> transRecord2Point(List<Record> records) {
List<Point> dps = new ArrayList<Point>();
for (Record record : records) {
List<Point.MetricBuilder> metricBuilders = new ArrayList<Point.MetricBuilder>();
Map<String, String> tags = new HashMap<String, String>();
Long time = 0L;
for (int i = 0; i < columnType.size(); i++) {
String type = columnType.get(i);
String name = columnName.get(i);
Column column = record.getColumn(i);
if (TSDBModel.TSDB_TAG.equals(type)) {
tags.put(name, column.asString());
} else if (TSDBModel.TSDB_FIELD_DOUBLE.equals(type)) {
metricBuilders.add(new Point.MetricBuilder(name).value(column.asDouble()));
} else if (TSDBModel.TSDB_FIELD_STRING.equals(type)) {
metricBuilders.add(new Point.MetricBuilder(name).value(column.asString()));
} else if (TSDBModel.TSDB_FIELD_BOOL.equals(type)) {
metricBuilders.add(new Point.MetricBuilder(name).value(column.asBoolean()));
} else if (TSDBModel.TSDB_TIMESTAMP.equals(type)) {
time = column.asLong();
} else if (TSDBModel.TSDB_METRIC_NUM.equals(type)) {
// compatible with previous usage of TSDB_METRIC_NUM
metricBuilders.add(new Point.MetricBuilder(name).value(column.asDouble()));
} else if (TSDBModel.TSDB_METRIC_STRING.equals(type)) {
// compatible with previous usage of TSDB_METRIC_STRING
metricBuilders.add(new Point.MetricBuilder(name).value(column.asString()));
}
}
for (Point.MetricBuilder metricBuilder : metricBuilders) {
dps.add(metricBuilder.tag(tags).timestamp(time).build(false));
}
}
return dps;
}
List<MultiFieldPoint> transRecord2MultiFieldPoint(List<Record> records, String tableName) {
List<MultiFieldPoint> dps = new ArrayList<MultiFieldPoint>();
for (Record record : records) {
MultiFieldPoint.MetricBuilder builder = MultiFieldPoint.metric(tableName);
for (int i = 0; i < columnType.size(); i++) {
String type = columnType.get(i);
String name = columnName.get(i);
Column column = record.getColumn(i);
if (TSDBModel.TSDB_TAG.equals(type)) {
builder.tag(name, column.asString());
} else if (TSDBModel.TSDB_FIELD_DOUBLE.equals(type)) {
builder.field(name, column.asDouble());
} else if (TSDBModel.TSDB_FIELD_STRING.equals(type)) {
builder.field(name, column.asString());
} else if (TSDBModel.TSDB_FIELD_BOOL.equals(type)) {
builder.field(name, column.asBoolean());
} else if (TSDBModel.TSDB_TIMESTAMP.equals(type)) {
builder.timestamp(column.asLong());
} else if (TSDBModel.TSDB_METRIC_NUM.equals(type)) {
// compatible with previous usage of TSDB_METRIC_NUM
builder.field(name, column.asDouble());
} else if (TSDBModel.TSDB_METRIC_STRING.equals(type)) {
// compatible with previous usage of TSDB_METRIC_STRING
builder.field(name, column.asString());
}
}
MultiFieldPoint point = builder.build(false);
dps.add(point);
}
return dps;
}
}

11
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBModel.java Normal file
View File

@ -0,0 +1,11 @@
package com.alibaba.datax.plugin.writer.tsdbwriter;
class TSDBModel {
static final String TSDB_METRIC_NUM = "metric_num";
static final String TSDB_METRIC_STRING = "metric_string";
static final String TSDB_TAG = "tag";
static final String TSDB_TIMESTAMP = "timestamp";
static final String TSDB_FIELD_DOUBLE = "field_double";
static final String TSDB_FIELD_STRING = "field_string";
static final String TSDB_FIELD_BOOL = "field_bool";
}

324
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBWriter.java
View File

@ -7,12 +7,20 @@ import com.alibaba.datax.common.spi.Writer;
import com.alibaba.datax.common.util.Configuration; import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.RetryUtil; import com.alibaba.datax.common.util.RetryUtil;
import com.alibaba.datax.plugin.writer.conn.TSDBConnection; import com.alibaba.datax.plugin.writer.conn.TSDBConnection;
import com.aliyun.hitsdb.client.TSDB;
import com.aliyun.hitsdb.client.TSDBClientFactory;
import com.aliyun.hitsdb.client.TSDBConfig;
import com.aliyun.hitsdb.client.value.request.MultiFieldPoint;
import com.aliyun.hitsdb.client.value.request.Point;
import com.aliyun.hitsdb.client.value.response.batch.IgnoreErrorsResult;
import com.aliyun.hitsdb.client.value.response.batch.MultiFieldIgnoreErrorsResult;
import com.aliyun.hitsdb.client.value.response.batch.SummaryResult;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.ArrayList; import java.io.IOException;
import java.util.List; import java.util.*;
import java.util.concurrent.Callable; import java.util.concurrent.Callable;
/** /**
@ -26,6 +34,9 @@ import java.util.concurrent.Callable;
@SuppressWarnings("unused") @SuppressWarnings("unused")
public class TSDBWriter extends Writer { public class TSDBWriter extends Writer {
private static SourceDBType DB_TYPE;
private static TSDB tsdb = null;
public static class Job extends Writer.Job { public static class Job extends Writer.Job {
private static final Logger LOG = LoggerFactory.getLogger(Job.class); private static final Logger LOG = LoggerFactory.getLogger(Job.class);
@ -34,33 +45,99 @@ public class TSDBWriter extends Writer {
@Override @Override
public void init() { public void init() {
this.originalConfig = super.getPluginJobConf(); originalConfig = super.getPluginJobConf();
String address = this.originalConfig.getString(Key.ENDPOINT); // check source db type
if (StringUtils.isBlank(address)) { String sourceDbType = originalConfig.getString(Key.SOURCE_DB_TYPE);
if (StringUtils.isBlank(sourceDbType)) {
sourceDbType = SourceDBType.TSDB.name();
originalConfig.set(Key.SOURCE_DB_TYPE, sourceDbType);
LOG.info("The parameter [" + Key.SOURCE_DB_TYPE + "] will be default value: " + SourceDBType.TSDB);
}
try {
DB_TYPE = SourceDBType.valueOf(sourceDbType);
} catch (Exception e) {
throw DataXException.asDataXException(TSDBWriterErrorCode.REQUIRED_VALUE, throw DataXException.asDataXException(TSDBWriterErrorCode.REQUIRED_VALUE,
"The parameter [" + Key.ENDPOINT + "] is not set."); "The parameter [" + Key.SOURCE_DB_TYPE +
"] is invalid, which should be one of [" + Arrays.toString(SourceDBType.values()) + "].");
} }
Integer batchSize = this.originalConfig.getInt(Key.BATCH_SIZE); // for tsdb
if (batchSize == null || batchSize < 1) { if (DB_TYPE == SourceDBType.TSDB) {
originalConfig.set(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE); String address = originalConfig.getString(Key.ENDPOINT);
LOG.info("The parameter [" + Key.BATCH_SIZE + if (StringUtils.isBlank(address)) {
"] will be default value: " + Constant.DEFAULT_BATCH_SIZE); throw DataXException.asDataXException(TSDBWriterErrorCode.REQUIRED_VALUE,
} "The parameter [" + Key.ENDPOINT + "] is not set.");
}
Integer retrySize = this.originalConfig.getInt(Key.MAX_RETRY_TIME); String username = originalConfig.getString(Key.USERNAME, null);
if (retrySize == null || retrySize < 0) { if (StringUtils.isBlank(username)) {
originalConfig.set(Key.MAX_RETRY_TIME, Constant.DEFAULT_TRY_SIZE); LOG.warn("The parameter [" + Key.USERNAME + "] is blank.");
LOG.info("The parameter [" + Key.MAX_RETRY_TIME + }
"] will be default value: " + Constant.DEFAULT_TRY_SIZE); String password = originalConfig.getString(Key.PASSWORD, null);
} if (StringUtils.isBlank(password)) {
LOG.warn("The parameter [" + Key.PASSWORD + "] is blank.");
}
Integer batchSize = originalConfig.getInt(Key.BATCH_SIZE);
if (batchSize == null || batchSize < 1) {
originalConfig.set(Key.BATCH_SIZE, Constant.DEFAULT_BATCH_SIZE);
LOG.info("The parameter [" + Key.BATCH_SIZE +
"] will be default value: " + Constant.DEFAULT_BATCH_SIZE);
}
Integer retrySize = originalConfig.getInt(Key.MAX_RETRY_TIME);
if (retrySize == null || retrySize < 0) {
originalConfig.set(Key.MAX_RETRY_TIME, Constant.DEFAULT_TRY_SIZE);
LOG.info("The parameter [" + Key.MAX_RETRY_TIME +
"] will be default value: " + Constant.DEFAULT_TRY_SIZE);
}
Boolean ignoreWriteError = originalConfig.getBool(Key.IGNORE_WRITE_ERROR);
if (ignoreWriteError == null) {
originalConfig.set(Key.IGNORE_WRITE_ERROR, Constant.DEFAULT_IGNORE_WRITE_ERROR);
LOG.info("The parameter [" + Key.IGNORE_WRITE_ERROR +
"] will be default value: " + Constant.DEFAULT_IGNORE_WRITE_ERROR);
}
} else if (DB_TYPE == SourceDBType.RDB) {
// for rdb
originalConfig.getNecessaryValue(Key.ENDPOINT, TSDBWriterErrorCode.REQUIRED_VALUE);
originalConfig.getNecessaryValue(Key.COLUMN_TYPE, TSDBWriterErrorCode.REQUIRED_VALUE);
originalConfig.getNecessaryValue(Key.COLUMN, TSDBWriterErrorCode.REQUIRED_VALUE);
String endpoint = originalConfig.getString(Key.ENDPOINT);
String[] split = endpoint.split(":");
if (split.length != 3) {
throw DataXException.asDataXException(TSDBWriterErrorCode.REQUIRED_VALUE,
"The parameter [" + Key.ENDPOINT + "] is invalid, which should be [http://IP:Port].");
}
String ip = split[1].substring(2);
int port = Integer.parseInt(split[2]);
String username = originalConfig.getString(Key.USERNAME, null);
if (StringUtils.isBlank(username)) {
LOG.warn("The parameter [" + Key.USERNAME + "] is blank.");
}
String password = originalConfig.getString(Key.PASSWORD, null);
if (StringUtils.isBlank(password)) {
LOG.warn("The parameter [" + Key.PASSWORD + "] is blank.");
}
if (!StringUtils.isBlank(password) && !StringUtils.isBlank(username)) {
tsdb = TSDBClientFactory.connect(TSDBConfig.address(ip, port).basicAuth(username, password).config());
} else {
tsdb = TSDBClientFactory.connect(TSDBConfig.address(ip, port).config());
}
String database = originalConfig.getString(Key.DATABASE, null);
if (StringUtils.isBlank(database)) {
LOG.info("The parameter [" + Key.DATABASE + "] is blank.");
} else {
tsdb.useDatabase(database);
}
LOG.info("Tsdb config:" + originalConfig.toJSON());
Boolean ignoreWriteError = this.originalConfig.getBool(Key.IGNORE_WRITE_ERROR);
if (ignoreWriteError == null) {
originalConfig.set(Key.IGNORE_WRITE_ERROR, Constant.DEFAULT_IGNORE_WRITE_ERROR);
LOG.info("The parameter [" + Key.IGNORE_WRITE_ERROR +
"] will be default value: " + Constant.DEFAULT_IGNORE_WRITE_ERROR);
} }
} }
@ -72,7 +149,7 @@ public class TSDBWriter extends Writer {
public List<Configuration> split(int mandatoryNumber) { public List<Configuration> split(int mandatoryNumber) {
ArrayList<Configuration> configurations = new ArrayList<Configuration>(mandatoryNumber); ArrayList<Configuration> configurations = new ArrayList<Configuration>(mandatoryNumber);
for (int i = 0; i < mandatoryNumber; i++) { for (int i = 0; i < mandatoryNumber; i++) {
configurations.add(this.originalConfig.clone()); configurations.add(originalConfig.clone());
} }
return configurations; return configurations;
} }
@ -83,6 +160,14 @@ public class TSDBWriter extends Writer {
@Override @Override
public void destroy() { public void destroy() {
if (DB_TYPE == SourceDBType.RDB) {
if (tsdb != null) {
try {
tsdb.close();
} catch (IOException ignored) {
}
}
}
} }
} }
@ -91,18 +176,87 @@ public class TSDBWriter extends Writer {
private static final Logger LOG = LoggerFactory.getLogger(Task.class); private static final Logger LOG = LoggerFactory.getLogger(Task.class);
private TSDBConnection conn; private TSDBConnection conn;
private boolean multiField;
private int batchSize; private int batchSize;
private int retrySize; private int retrySize;
private boolean ignoreWriteError; private boolean ignoreWriteError;
private String tableName;
private TSDBConverter tsdbConverter;
@Override @Override
public void init() { public void init() {
Configuration writerSliceConfig = getPluginJobConf(); Configuration writerSliceConfig = getPluginJobConf();
String address = writerSliceConfig.getString(Key.ENDPOINT);
this.conn = new TSDBConnection(address); // single field | multi fields
this.batchSize = writerSliceConfig.getInt(Key.BATCH_SIZE); this.multiField = writerSliceConfig.getBool(Key.MULTI_FIELD, false);
this.retrySize = writerSliceConfig.getInt(Key.MAX_RETRY_TIME);
this.ignoreWriteError = writerSliceConfig.getBool(Key.IGNORE_WRITE_ERROR); this.ignoreWriteError = writerSliceConfig.getBool(Key.IGNORE_WRITE_ERROR);
// for tsdb
if (DB_TYPE == SourceDBType.TSDB) {
String address = writerSliceConfig.getString(Key.ENDPOINT);
String database = writerSliceConfig.getString(Key.DATABASE);
String username = writerSliceConfig.getString(Key.USERNAME);
String password = writerSliceConfig.getString(Key.PASSWORD);
this.conn = new TSDBConnection(address, database, username, password);
this.batchSize = writerSliceConfig.getInt(Key.BATCH_SIZE);
this.retrySize = writerSliceConfig.getInt(Key.MAX_RETRY_TIME);
} else if (DB_TYPE == SourceDBType.RDB) {
// for rdb
int timeSize = 0;
int fieldSize = 0;
int tagSize = 0;
batchSize = writerSliceConfig.getInt(Key.BATCH_SIZE, 100);
List<String> columnName = writerSliceConfig.getList(Key.COLUMN, String.class);
List<String> columnType = writerSliceConfig.getList(Key.COLUMN_TYPE, String.class);
Set<String> typeSet = new HashSet<String>(columnType);
if (columnName.size() != columnType.size()) {
throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
"The parameter [" + Key.COLUMN_TYPE + "] should has same length with [" + Key.COLUMN + "].");
}
for (String type : columnType) {
if (TSDBModel.TSDB_TAG.equals(type)) {
tagSize ++;
} else if (TSDBModel.TSDB_FIELD_DOUBLE.equals(type) || TSDBModel.TSDB_FIELD_STRING.equals(type)
|| TSDBModel.TSDB_FIELD_BOOL.equals(type)) {
fieldSize++;
} else if (TSDBModel.TSDB_TIMESTAMP.equals(type)) {
timeSize++;
}
}
if (fieldSize == 0) {
// compatible with previous usage of TSDB_METRIC_NUM and TSDB_METRIC_STRING
if (!typeSet.contains(TSDBModel.TSDB_METRIC_NUM) && !typeSet.contains(TSDBModel.TSDB_METRIC_STRING)) {
throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
"The parameter [" + Key.COLUMN_TYPE + "] is invalid, must set at least one of "
+ TSDBModel.TSDB_FIELD_DOUBLE + ", " + TSDBModel.TSDB_FIELD_STRING + " or " + TSDBModel.TSDB_FIELD_BOOL + ".");
}
}
if (tagSize == 0) {
throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
"The parameter [" + Key.COLUMN_TYPE + "] is invalid, must set " + TSDBModel.TSDB_TAG + ". ");
}
if (timeSize != 1) {
throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
"The parameter [" + Key.COLUMN_TYPE + "] is invalid, must set one and only one "
+ TSDBModel.TSDB_TIMESTAMP + ".");
}
if (multiField) {
// check source db type
tableName = writerSliceConfig.getString(Key.TABLE);
if (StringUtils.isBlank(tableName)) {
throw DataXException.asDataXException(TSDBWriterErrorCode.ILLEGAL_VALUE,
"The parameter [" + Key.TABLE + "] h must set when use multi field input.");
}
}
tsdbConverter = new TSDBConverter(columnName, columnType);
}
} }
@Override @Override
@ -111,30 +265,52 @@ public class TSDBWriter extends Writer {
@Override @Override
public void startWrite(RecordReceiver recordReceiver) { public void startWrite(RecordReceiver recordReceiver) {
try { // for tsdb
Record lastRecord = null; if (DB_TYPE == SourceDBType.TSDB) {
Record record; try {
int count = 0; Record lastRecord = null;
StringBuilder dps = new StringBuilder(); Record record;
while ((record = recordReceiver.getFromReader()) != null) { int count = 0;
final int recordLength = record.getColumnNumber(); StringBuilder dps = new StringBuilder();
for (int i = 0; i < recordLength; i++) { while ((record = recordReceiver.getFromReader()) != null) {
dps.append(record.getColumn(i).asString()); final int recordLength = record.getColumnNumber();
dps.append(","); for (int i = 0; i < recordLength; i++) {
count++; dps.append(record.getColumn(i).asString());
if (count == batchSize) { dps.append(",");
count = 0; count++;
batchPut(record, "[" + dps.substring(0, dps.length() - 1) + "]"); if (count == batchSize) {
dps = new StringBuilder(); count = 0;
batchPut(record, "[" + dps.substring(0, dps.length() - 1) + "]");
dps = new StringBuilder();
}
} }
lastRecord = record;
} }
lastRecord = record; if (StringUtils.isNotBlank(dps.toString())) {
batchPut(lastRecord, "[" + dps.substring(0, dps.length() - 1) + "]");
}
} catch (Exception e) {
throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION, e);
} }
if (StringUtils.isNotBlank(dps.toString())) { } else if (DB_TYPE == SourceDBType.RDB) {
batchPut(lastRecord, "[" + dps.substring(0, dps.length() - 1) + "]"); // for rdb
List<Record> writerBuffer = new ArrayList<Record>(this.batchSize);
Record record;
long total = 0;
while ((record = recordReceiver.getFromReader()) != null) {
writerBuffer.add(record);
if (writerBuffer.size() >= this.batchSize) {
total += doBatchInsert(writerBuffer);
writerBuffer.clear();
}
} }
} catch (Exception e) { if (!writerBuffer.isEmpty()) {
throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION, e); total += doBatchInsert(writerBuffer);
writerBuffer.clear();
}
getTaskPluginCollector().collectMessage("write size", total + "");
LOG.info("Task finished, write size: {}", total);
} }
} }
@ -143,12 +319,13 @@ public class TSDBWriter extends Writer {
RetryUtil.executeWithRetry(new Callable<Integer>() { RetryUtil.executeWithRetry(new Callable<Integer>() {
@Override @Override
public Integer call() { public Integer call() {
if (!conn.put(dps)) { final boolean success = multiField ? conn.mput(dps) : conn.put(dps);
getTaskPluginCollector().collectDirtyRecord(record, "Put data points failed!"); if (success) {
throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION, return 0;
"Put data points failed!");
} }
return 0; getTaskPluginCollector().collectDirtyRecord(record, "Put data points failed!");
throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION,
"Put data points failed!");
} }
}, retrySize, 60000L, true); }, retrySize, 60000L, true);
} catch (Exception e) { } catch (Exception e) {
@ -160,6 +337,47 @@ public class TSDBWriter extends Writer {
} }
} }
private long doBatchInsert(final List<Record> writerBuffer) {
int size;
if (ignoreWriteError) {
if (multiField) {
List<MultiFieldPoint> points = tsdbConverter.transRecord2MultiFieldPoint(writerBuffer, tableName);
size = points.size();
MultiFieldIgnoreErrorsResult ignoreErrorsResult = tsdb.multiFieldPutSync(points, MultiFieldIgnoreErrorsResult.class);
if (ignoreErrorsResult == null) {
LOG.error("Unexpected inner error for insert");
} else if (ignoreErrorsResult.getFailed() > 0) {
LOG.error("write TSDB failed num:" + ignoreErrorsResult.getFailed());
}
} else {
List<Point> points = tsdbConverter.transRecord2Point(writerBuffer);
size = points.size();
IgnoreErrorsResult ignoreErrorsResult = tsdb.putSync(points, IgnoreErrorsResult.class);
if (ignoreErrorsResult == null) {
LOG.error("Unexpected inner error for insert");
} else if (ignoreErrorsResult.getFailed() > 0) {
LOG.error("write TSDB failed num:" + ignoreErrorsResult.getFailed());
}
}
} else {
SummaryResult summaryResult;
if (multiField) {
List<MultiFieldPoint> points = tsdbConverter.transRecord2MultiFieldPoint(writerBuffer, tableName);
size = points.size();
summaryResult = tsdb.multiFieldPutSync(points, SummaryResult.class);
} else {
List<Point> points = tsdbConverter.transRecord2Point(writerBuffer);
size = points.size();
summaryResult = tsdb.putSync(points, SummaryResult.class);
}
if (summaryResult.getFailed() > 0) {
LOG.error("write TSDB failed num:" + summaryResult.getFailed());
throw DataXException.asDataXException(TSDBWriterErrorCode.RUNTIME_EXCEPTION, "Write TSDB failed", new Exception());
}
}
return size;
}
@Override @Override
public void post() { public void post() {
} }

1
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/tsdbwriter/TSDBWriterErrorCode.java
View File

@ -13,6 +13,7 @@ import com.alibaba.datax.common.spi.ErrorCode;
public enum TSDBWriterErrorCode implements ErrorCode { public enum TSDBWriterErrorCode implements ErrorCode {
REQUIRED_VALUE("TSDBWriter-00", "Missing the necessary value"), REQUIRED_VALUE("TSDBWriter-00", "Missing the necessary value"),
ILLEGAL_VALUE("TSDBWriter-01", "Illegal value"),
RUNTIME_EXCEPTION("TSDBWriter-01", "Runtime exception"), RUNTIME_EXCEPTION("TSDBWriter-01", "Runtime exception"),
RETRY_WRITER_EXCEPTION("TSDBWriter-02", "After repeated attempts, the write still fails"); RETRY_WRITER_EXCEPTION("TSDBWriter-02", "After repeated attempts, the write still fails");

56
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/util/HttpUtils.java
View File

@ -1,11 +1,14 @@
package com.alibaba.datax.plugin.writer.util; package com.alibaba.datax.plugin.writer.util;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.client.fluent.Content; import org.apache.http.client.fluent.Content;
import org.apache.http.client.fluent.Request; import org.apache.http.client.fluent.Request;
import org.apache.http.entity.ContentType; import org.apache.http.entity.ContentType;
import java.nio.charset.Charset; import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.Map; import java.util.Map;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -19,43 +22,44 @@ import java.util.concurrent.TimeUnit;
*/ */
public final class HttpUtils { public final class HttpUtils {
public final static Charset UTF_8 = Charset.forName("UTF-8");
public final static int CONNECT_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60); public final static int CONNECT_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
public final static int SOCKET_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60); public final static int SOCKET_TIMEOUT_DEFAULT_IN_MILL = (int) TimeUnit.SECONDS.toMillis(60);
private static final String CREDENTIALS_FORMAT = "%s:%s";
private static final String BASIC_AUTHENTICATION_FORMAT = "Basic %s";
private HttpUtils() { private HttpUtils() {
} }
public static String get(String url) throws Exception { public static String get(String url, String username, String password) throws Exception {
Content content = Request.Get(url) final Request request = Request.Get(url)
.connectTimeout(CONNECT_TIMEOUT_DEFAULT_IN_MILL) .connectTimeout(CONNECT_TIMEOUT_DEFAULT_IN_MILL)
.socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL) .socketTimeout(SOCKET_TIMEOUT_DEFAULT_IN_MILL);
addAuth(request, username, password);
Content content = request
.execute() .execute()
.returnContent(); .returnContent();
if (content == null) { if (content == null) {
return null; return null;
} }
return content.asString(UTF_8); return content.asString(StandardCharsets.UTF_8);
} }
public static String post(String url, Map<String, Object> params) throws Exception { public static String post(String url, String username, String password, Map<String, Object> params) throws Exception {
return post(url, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL); return post(url, username, password, JSON.toJSONString(params), CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
} }
public static String post(String url, String params) throws Exception { public static String post(String url, String username, String password, String params) throws Exception {
return post(url, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL); return post(url, username, password, params, CONNECT_TIMEOUT_DEFAULT_IN_MILL, SOCKET_TIMEOUT_DEFAULT_IN_MILL);
} }
public static String post(String url, Map<String, Object> params, public static String post(String url, String username, String password, String params,
int connectTimeoutInMill, int socketTimeoutInMill) throws Exception { int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
return post(url, JSON.toJSONString(params), connectTimeoutInMill, socketTimeoutInMill); Request request = Request.Post(url)
}
public static String post(String url, String params,
int connectTimeoutInMill, int socketTimeoutInMill) throws Exception {
Content content = Request.Post(url)
.connectTimeout(connectTimeoutInMill) .connectTimeout(connectTimeoutInMill)
.socketTimeout(socketTimeoutInMill) .socketTimeout(socketTimeoutInMill);
addAuth(request, username, password);
Content content = request
.addHeader("Content-Type", "application/json") .addHeader("Content-Type", "application/json")
.bodyString(params, ContentType.APPLICATION_JSON) .bodyString(params, ContentType.APPLICATION_JSON)
.execute() .execute()
@ -63,6 +67,22 @@ public final class HttpUtils {
if (content == null) { if (content == null) {
return null; return null;
} }
return content.asString(UTF_8); return content.asString(StandardCharsets.UTF_8);
}
private static void addAuth(Request request, String username, String password) {
String authorization = generateHttpAuthorization(username, password);
if (authorization != null) {
request.setHeader("Authorization", authorization);
}
}
private static String generateHttpAuthorization(String username, String password) {
if (StringUtils.isBlank(username) || StringUtils.isBlank(password)) {
return null;
}
String credentials = String.format(CREDENTIALS_FORMAT, username, password);
credentials = Base64.getEncoder().encodeToString(credentials.getBytes());
return String.format(BASIC_AUTHENTICATION_FORMAT, credentials);
} }
} }

38
tsdbwriter/src/main/java/com/alibaba/datax/plugin/writer/util/TSDBUtils.java
View File

@ -2,6 +2,7 @@ package com.alibaba.datax.plugin.writer.util;
import com.alibaba.datax.plugin.writer.conn.DataPoint4TSDB; import com.alibaba.datax.plugin.writer.conn.DataPoint4TSDB;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -22,45 +23,56 @@ public final class TSDBUtils {
private TSDBUtils() { private TSDBUtils() {
} }
public static String version(String address) { public static String version(String address, String username, String password) {
String url = String.format("%s/api/version", address); String url = String.format("%s/api/version", address);
String rsp; String rsp;
try { try {
rsp = HttpUtils.get(url); rsp = HttpUtils.get(url, username, password);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
return rsp; return rsp;
} }
public static String config(String address) { public static String config(String address, String username, String password) {
String url = String.format("%s/api/config", address); String url = String.format("%s/api/config", address);
String rsp; String rsp;
try { try {
rsp = HttpUtils.get(url); rsp = HttpUtils.get(url, username, password);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
return rsp; return rsp;
} }
public static boolean put(String address, List<DataPoint4TSDB> dps) { public static boolean put(String address, String database, String username, String password, List<DataPoint4TSDB> dps) {
return put(address, JSON.toJSON(dps)); return put(address, database, username, password, JSON.toJSON(dps));
} }
public static boolean put(String address, DataPoint4TSDB dp) { public static boolean put(String address, String database, String username, String password, DataPoint4TSDB dp) {
return put(address, JSON.toJSON(dp)); return put(address, database, username, password, JSON.toJSON(dp));
} }
private static boolean put(String address, Object o) { private static boolean put(String address, String database, String username, String password, Object o) {
return put(address, o.toString()); return put(address, database, username, password, o.toString());
} }
public static boolean put(String address, String s) { public static boolean put(String address, String database, String username, String password, String s) {
String url = String.format("%s/api/put", address); return put(address, database, username, password, s, false);
}
public static boolean mput(String address, String database, String username, String password, String s) {
return put(address, database, username, password, s, true);
}
public static boolean put(String address, String database, String username, String password, String s, boolean multiField) {
String url = address + (multiField ? "/api/mput" : "/api/put");
if (!StringUtils.isBlank(database)) {
url = url.concat("?db=" + database);
}
String rsp; String rsp;
try { try {
rsp = HttpUtils.post(url, s); rsp = HttpUtils.post(url, username, password, s);
// If successful, the returned content should be null. // If successful, the returned content should be null.
assert rsp == null; assert rsp == null;
} catch (Exception e) { } catch (Exception e) {

4
tsdbwriter/src/test/java/com/alibaba/datax/plugin/writer/conn/TSDBConnectionTest.java
View File

@ -19,12 +19,12 @@ public class TSDBConnectionTest {
@Test @Test
public void testVersion() { public void testVersion() {
String version = new TSDBConnection(TSDB_ADDRESS).version(); String version = new TSDBConnection(TSDB_ADDRESS,null,null,null).version();
Assert.assertNotNull(version); Assert.assertNotNull(version);
} }
@Test @Test
public void testIsSupported() { public void testIsSupported() {
Assert.assertTrue(new TSDBConnection(TSDB_ADDRESS).isSupported()); Assert.assertTrue(new TSDBConnection(TSDB_ADDRESS,null,null,null).isSupported());
} }
} }

4
tsdbwriter/src/test/java/com/alibaba/datax/plugin/writer/util/HttpUtilsTest.java
View File

@ -24,7 +24,7 @@ public class HttpUtilsTest {
Map<String, Object> params = new HashMap<String, Object>(); Map<String, Object> params = new HashMap<String, Object>();
params.put("foo", "bar"); params.put("foo", "bar");
String rsp = HttpUtils.post(url, params); String rsp = HttpUtils.post(url, null,null,params);
System.out.println(rsp); System.out.println(rsp);
Assert.assertNotNull(rsp); Assert.assertNotNull(rsp);
} }
@ -32,7 +32,7 @@ public class HttpUtilsTest {
@Test @Test
public void testGet() throws Exception { public void testGet() throws Exception {
String url = String.format("%s/api/version", Const.OPENTSDB_ADDRESS); String url = String.format("%s/api/version", Const.OPENTSDB_ADDRESS);
String rsp = HttpUtils.get(url); String rsp = HttpUtils.get(url,null,null);
System.out.println(rsp); System.out.println(rsp);
Assert.assertNotNull(rsp); Assert.assertNotNull(rsp);
} }

4
tsdbwriter/src/test/java/com/alibaba/datax/plugin/writer/util/TSDBTest.java
View File

@ -17,11 +17,11 @@ public class TSDBTest {
@Test @Test
public void testVersion() { public void testVersion() {
String version = TSDBUtils.version(Const.TSDB_ADDRESS); String version = TSDBUtils.version(Const.TSDB_ADDRESS,null,null);
Assert.assertNotNull(version); Assert.assertNotNull(version);
System.out.println(version); System.out.println(version);
version = TSDBUtils.version(Const.OPENTSDB_ADDRESS); version = TSDBUtils.version(Const.OPENTSDB_ADDRESS,null,null);
Assert.assertNotNull(version); Assert.assertNotNull(version);
System.out.println(version); System.out.println(version);
} }