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

tdenginewriter

Browse Source
This commit is contained in:
zyyang 2021-10-19 15:58:47 +08:00
parent 5e54aee590
commit ab526ca5c4
1 changed files with 11 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
tdenginewriter/src/main/java/com/alibaba/datax/plugin/writer/TDengineWriter.java
View File

@ -45,7 +45,6 @@ public class TDengineWriter extends Writer {
public static class Task extends Writer.Task { public static class Task extends Writer.Task {
private static final Logger LOG = LoggerFactory.getLogger(Job.class); private static final Logger LOG = LoggerFactory.getLogger(Job.class);
private static final String NEWLINE_FLAG = System.getProperty("line.separator", "\n");
private Configuration writerSliceConfig; private Configuration writerSliceConfig;
@Override @Override
@ -70,25 +69,30 @@ public class TDengineWriter extends Writer {
String peerPluginName = this.writerSliceConfig.getString(PEER_PLUGIN_NAME); String peerPluginName = this.writerSliceConfig.getString(PEER_PLUGIN_NAME);
if (peerPluginName.equals("opentsdbreader")) { if (peerPluginName.equals("opentsdbreader")) {
// opentsdb json protocol use JNI and schemaless API to write // opentsdb json protocol use JNI and schemaless API to write
String host = properties.getProperty(Key.HOST); String host = properties.getProperty(Key.HOST);
int port = Integer.parseInt(properties.getProperty(Key.PORT)); int port = Integer.parseInt(properties.getProperty(Key.PORT));
String dbname = properties.getProperty(Key.DBNAME); String dbname = properties.getProperty(Key.DBNAME);
String user = properties.getProperty(Key.USER); String user = properties.getProperty(Key.USER);
String password = properties.getProperty(Key.PASSWORD); String password = properties.getProperty(Key.PASSWORD);
JniConnection conn = null;
try { try {
JniConnection conn = new JniConnection(properties); conn = new JniConnection(properties);
conn.open(host, port, dbname, user, password); conn.open(host, port, dbname, user, password);
LOG.info("TDengine connection established, host: " + host + ", port: " + port + ", dbname: " + dbname + ", user: " + user); LOG.info("TDengine connection established, host: " + host + ", port: " + port + ", dbname: " + dbname + ", user: " + user);
int batchSize = Integer.parseInt(properties.getProperty(Key.BATCH_SIZE, DEFAULT_BATCH_SIZE)); int batchSize = Integer.parseInt(properties.getProperty(Key.BATCH_SIZE, DEFAULT_BATCH_SIZE));
writeOpentsdb(lineReceiver, conn, batchSize); writeOpentsdb(lineReceiver, conn, batchSize);
conn.close();
LOG.info("TDengine connection closed");
} catch (Exception e) { } catch (Exception e) {
LOG.error(e.getMessage()); LOG.error(e.getMessage());
e.printStackTrace(); e.printStackTrace();
} finally {
try {
if (conn != null)
conn.close();
} catch (Exception e) {
e.printStackTrace();
}
LOG.info("TDengine connection closed");
} }
} else { } else {
// other // other
@ -132,7 +136,7 @@ public class TDengineWriter extends Writer {
private String recordToString(Record record) { private String recordToString(Record record) {
int recordLength = record.getColumnNumber(); int recordLength = record.getColumnNumber();
if (0 == recordLength) { if (0 == recordLength) {
return NEWLINE_FLAG; return "";
} }
Column column; Column column;
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
@ -141,7 +145,6 @@ public class TDengineWriter extends Writer {
sb.append(column.asString()).append("\t"); sb.append(column.asString()).append("\t");
} }
sb.setLength(sb.length() - 1); sb.setLength(sb.length() - 1);
sb.append(NEWLINE_FLAG);
return sb.toString(); return sb.toString();
} }
} }