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

Merge 7b87720d7d into 0824b45c5e

Browse Source
This commit is contained in:
rubybj 2025-04-10 16:23:21 +08:00 committed by GitHub
commit 4f6fa84f0a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 10 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
opentsdbreader/src/main/java/com/alibaba/datax/plugin/reader/opentsdbreader/OpenTSDBReader.java
View File

@ -140,12 +140,13 @@ public class OpenTSDBReader extends Reader {
if (TimeUtils.isSecond(endTime)) { if (TimeUtils.isSecond(endTime)) {
endTime *= 1000; endTime *= 1000;
} }
DateTime startDateTime = new DateTime(TimeUtils.getTimeInHour(startTime)); DateTime startDateTime=null;
DateTime endDateTime = new DateTime(TimeUtils.getTimeInHour(endTime)); DateTime endDateTime=new DateTime(TimeUtils.getTimeInHour(endTime));
// split by metric // split by metric
for (String column : columns) { for (String column : columns) {
// split by time in hour // split by time in hour
//When the metric is multiple, the startDateTime needs to be reset
startDateTime = new DateTime(TimeUtils.getTimeInHour(startTime));
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, Collections.singletonList(column));
@ -204,7 +205,12 @@ public class OpenTSDBReader extends Reader {
public void startRead(RecordSender recordSender) { public void startRead(RecordSender recordSender) {
try { try {
for (String column : columns) { for (String column : columns) {
conn.sendDPs(column, this.startTime, this.endTime, recordSender); //If some metric does not exist, it is not a fatal error. Just log it and do not affect other jobs.
try {
conn.sendDPs(column, this.startTime, this.endTime, recordSender);
}catch(net.opentsdb.uid.NoSuchUniqueName e){
LOG.warn("在时间段{}-{}无{}指标", this.startTime,this.endTime,column);
}
} }
} catch (Exception e) { } catch (Exception e) {
throw DataXException.asDataXException( throw DataXException.asDataXException(