make time column configurable
Adding the header of the time column to the CsvReaderSettings. This is necessary to parse arbitrary CSV files.
This commit is contained in:
@@ -12,30 +12,45 @@ public class CsvReaderSettings {
|
|||||||
|
|
||||||
private Set<String> ignoreColumnNames = new HashSet<String>();
|
private Set<String> ignoreColumnNames = new HashSet<String>();
|
||||||
|
|
||||||
|
private String timeColumn;
|
||||||
|
|
||||||
public CsvReaderSettings() {
|
public CsvReaderSettings() {
|
||||||
separator = ',';
|
separator = ',';
|
||||||
|
timeColumn = "@timestamp";
|
||||||
}
|
}
|
||||||
|
|
||||||
private CsvReaderSettings(final byte separator, final Collection<String> ignoreColumns) {
|
private CsvReaderSettings(final String timeColumn, final byte separator, final Collection<String> ignoreColumns) {
|
||||||
|
|
||||||
|
this.timeColumn = timeColumn;
|
||||||
this.separator = separator;
|
this.separator = separator;
|
||||||
this.ignoreColumnNames.addAll(ignoreColumns);
|
this.ignoreColumnNames.addAll(ignoreColumns);
|
||||||
}
|
}
|
||||||
|
|
||||||
public static CsvReaderSettings create(final byte separator, final String... ignoreColumnNames) {
|
public static CsvReaderSettings create(final String timeColumn, final byte separator,
|
||||||
return new CsvReaderSettings(separator, List.of(ignoreColumnNames));
|
final String... ignoreColumnNames) {
|
||||||
|
return new CsvReaderSettings(timeColumn, separator, List.of(ignoreColumnNames));
|
||||||
}
|
}
|
||||||
|
|
||||||
public static CsvReaderSettings create(final char separator, final String... ignoreColumnNames) {
|
public static CsvReaderSettings create(final String timeColumn, final char separator,
|
||||||
return CsvReaderSettings.create(separator, List.of(ignoreColumnNames));
|
final String... ignoreColumnNames) {
|
||||||
|
return CsvReaderSettings.create(timeColumn, separator, List.of(ignoreColumnNames));
|
||||||
}
|
}
|
||||||
|
|
||||||
public static CsvReaderSettings create(final char separator, final Collection<String> ignoreColumnNames) {
|
public static CsvReaderSettings create(final String timeColumn, final char separator,
|
||||||
|
final Collection<String> ignoreColumnNames) {
|
||||||
Preconditions.checkTrue(separator == (byte) separator,
|
Preconditions.checkTrue(separator == (byte) separator,
|
||||||
"Only separators that fulfill separator == (byte)separator are supported. "
|
"Only separators that fulfill separator == (byte)separator are supported. "
|
||||||
+ "This restriction is because the parsing algorithm skips the overhead of "
|
+ "This restriction is because the parsing algorithm skips the overhead of "
|
||||||
+ "translating bytes to characters.");
|
+ "translating bytes to characters.");
|
||||||
return new CsvReaderSettings((byte) separator, ignoreColumnNames);
|
return new CsvReaderSettings(timeColumn, (byte) separator, ignoreColumnNames);
|
||||||
|
}
|
||||||
|
|
||||||
|
public String getTimeColumn() {
|
||||||
|
return timeColumn;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void setTimeColumn(final String timeColumn) {
|
||||||
|
this.timeColumn = timeColumn;
|
||||||
}
|
}
|
||||||
|
|
||||||
public byte getSeparator() {
|
public byte getSeparator() {
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ class CsvToEntryTransformer {
|
|||||||
|
|
||||||
int[] columns = null;
|
int[] columns = null;
|
||||||
final byte[] buffer = new byte[4096 * 16];
|
final byte[] buffer = new byte[4096 * 16];
|
||||||
final int keyTimestamp = Tags.STRING_COMPRESSOR.put("@timestamp");
|
final int keyTimestamp = Tags.STRING_COMPRESSOR.put(settings.getTimeColumn());
|
||||||
final int keyDuration = Tags.STRING_COMPRESSOR.put("duration");
|
final int keyDuration = Tags.STRING_COMPRESSOR.put("duration");
|
||||||
final FastISODateParser dateParser = new FastISODateParser();
|
final FastISODateParser dateParser = new FastISODateParser();
|
||||||
|
|
||||||
|
|||||||
@@ -66,7 +66,7 @@ public final class IngestionHandler implements Callable<Void> {
|
|||||||
} else {
|
} else {
|
||||||
in.reset();
|
in.reset();
|
||||||
final CsvToEntryTransformer csvTransformer = new CsvToEntryTransformer(queue,
|
final CsvToEntryTransformer csvTransformer = new CsvToEntryTransformer(queue,
|
||||||
CsvReaderSettings.create(','));
|
CsvReaderSettings.create("@timestamp", ','));
|
||||||
csvTransformer.readCSV(in);
|
csvTransformer.readCSV(in);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -48,7 +48,7 @@ public class CsvToEntryTransformerTest {
|
|||||||
+ dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",2,tagValue\n";
|
+ dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",2,tagValue\n";
|
||||||
|
|
||||||
final ArrayBlockingQueue<Entries> queue = db.getQueue();
|
final ArrayBlockingQueue<Entries> queue = db.getQueue();
|
||||||
final CsvReaderSettings settings = CsvReaderSettings.create(',');
|
final CsvReaderSettings settings = CsvReaderSettings.create("@timestamp", ',');
|
||||||
final CsvToEntryTransformer csvToEntryTransformer = new CsvToEntryTransformer(queue, settings);
|
final CsvToEntryTransformer csvToEntryTransformer = new CsvToEntryTransformer(queue, settings);
|
||||||
csvToEntryTransformer.readCSV(new ByteArrayInputStream(csv.getBytes(StandardCharsets.UTF_8)));
|
csvToEntryTransformer.readCSV(new ByteArrayInputStream(csv.getBytes(StandardCharsets.UTF_8)));
|
||||||
queue.put(Entries.POISON);
|
queue.put(Entries.POISON);
|
||||||
@@ -56,13 +56,13 @@ public class CsvToEntryTransformerTest {
|
|||||||
|
|
||||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||||
final LongList result = db.get(new Query("tag=tagValue", DateTimeRange.max())).singleGroup().flatMap();
|
final LongList result = db.get(new Query("tag=tagValue", DateTimeRange.max())).singleGroup().flatMap();
|
||||||
Assert.assertEquals(result.size(), 4);
|
Assert.assertEquals(4, result.size());
|
||||||
|
|
||||||
Assert.assertEquals(result.get(0), dateA.toInstant().toEpochMilli());
|
Assert.assertEquals(dateA.toInstant().toEpochMilli(), result.get(0));
|
||||||
Assert.assertEquals(result.get(1), 1);
|
Assert.assertEquals(1, result.get(1));
|
||||||
|
|
||||||
Assert.assertEquals(result.get(2), dateB.toInstant().truncatedTo(ChronoUnit.MILLIS).toEpochMilli());
|
Assert.assertEquals(dateB.toInstant().truncatedTo(ChronoUnit.MILLIS).toEpochMilli(), result.get(2));
|
||||||
Assert.assertEquals(result.get(3), 2);
|
Assert.assertEquals(2, result.get(3));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -85,7 +85,7 @@ public class CsvToEntryTransformerTest {
|
|||||||
+ "2000-01-01T00:00:00.001Z,2,ignoreValue,ignoreValue,tagValue\n";
|
+ "2000-01-01T00:00:00.001Z,2,ignoreValue,ignoreValue,tagValue\n";
|
||||||
|
|
||||||
final ArrayBlockingQueue<Entries> queue = db.getQueue();
|
final ArrayBlockingQueue<Entries> queue = db.getQueue();
|
||||||
final CsvReaderSettings settings = CsvReaderSettings.create(',', "ignoredColumn");
|
final CsvReaderSettings settings = CsvReaderSettings.create("@timestamp", ',', "ignoredColumn");
|
||||||
final CsvToEntryTransformer csvToEntryTransformer = new CsvToEntryTransformer(queue, settings);
|
final CsvToEntryTransformer csvToEntryTransformer = new CsvToEntryTransformer(queue, settings);
|
||||||
csvToEntryTransformer.readCSV(new ByteArrayInputStream(csv.getBytes(StandardCharsets.UTF_8)));
|
csvToEntryTransformer.readCSV(new ByteArrayInputStream(csv.getBytes(StandardCharsets.UTF_8)));
|
||||||
queue.put(Entries.POISON);
|
queue.put(Entries.POISON);
|
||||||
|
|||||||
@@ -43,14 +43,15 @@ public class PdbControllerTest {
|
|||||||
public void testUploadCsv() {
|
public void testUploadCsv() {
|
||||||
|
|
||||||
final String ignoredColumn = "ignoredColumn";
|
final String ignoredColumn = "ignoredColumn";
|
||||||
|
final String timeColumn = "time";
|
||||||
final OffsetDateTime dateA = OffsetDateTime.now();
|
final OffsetDateTime dateA = OffsetDateTime.now();
|
||||||
final OffsetDateTime dateB = OffsetDateTime.now();
|
final OffsetDateTime dateB = OffsetDateTime.now();
|
||||||
|
|
||||||
final String csv = "@timestamp,duration,tag," + ignoredColumn + "\n"//
|
final String csv = timeColumn + ",duration,tag," + ignoredColumn + "\n"//
|
||||||
+ dateA.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",1,tagValue,ignoredValue\n"//
|
+ dateA.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",1,tagValue,ignoredValue\n"//
|
||||||
+ dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",2,tagValue,ignoredValue\n";
|
+ dateB.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + ",2,tagValue,ignoredValue\n";
|
||||||
|
|
||||||
final CsvReaderSettings settings = CsvReaderSettings.create(',', ignoredColumn);
|
final CsvReaderSettings settings = CsvReaderSettings.create(timeColumn, ',', ignoredColumn);
|
||||||
uploadCsv(settings, csv);
|
uploadCsv(settings, csv);
|
||||||
{
|
{
|
||||||
final GroupResult groupResult = performanceDb.get(new Query("tag=tagValue", DateTimeRange.ofDay(dateA)))
|
final GroupResult groupResult = performanceDb.get(new Query("tag=tagValue", DateTimeRange.ofDay(dateA)))
|
||||||
|
|||||||
Reference in New Issue
Block a user