You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2016/11/14 12:52:10 UTC
[3/8] storm git commit: [STORM-2177][SQL] TSV format support user
defined delimiter
[STORM-2177][SQL] TSV format support user defined delimiter
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f14e0b21
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f14e0b21
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f14e0b21
Branch: refs/heads/master
Commit: f14e0b21666e009f3138c228d9ea0727488a895f
Parents: d165464
Author: vesense <be...@163.com>
Authored: Wed Nov 2 14:13:31 2016 +0800
Committer: vesense <be...@163.com>
Committed: Wed Nov 2 16:06:19 2016 +0800
----------------------------------------------------------------------
.../apache/storm/sql/runtime/serde/csv/CsvScheme.java | 2 +-
.../storm/sql/runtime/serde/csv/CsvSerializer.java | 4 +++-
.../apache/storm/sql/runtime/serde/tsv/TsvScheme.java | 5 +++--
.../storm/sql/runtime/serde/tsv/TsvSerializer.java | 7 ++++---
.../org/apache/storm/sql/runtime/utils/SerdeUtils.java | 6 ++++--
.../test/org/apache/storm/sql/TestCsvSerializer.java | 12 ++++++------
.../test/org/apache/storm/sql/TestTsvSerializer.java | 6 ++++--
7 files changed, 25 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/f14e0b21/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvScheme.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvScheme.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvScheme.java
index b258b40..ce1849f 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvScheme.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvScheme.java
@@ -55,7 +55,7 @@ public class CsvScheme implements Scheme {
ArrayList<Object> list = new ArrayList<>(fieldNames.size());
for (int i = 0; i < record.size(); i++) {
- list.add(record.get(i++));
+ list.add(record.get(i));
}
return list;
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/storm/blob/f14e0b21/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvSerializer.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvSerializer.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvSerializer.java
index 0f54c7f..223036b 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvSerializer.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvSerializer.java
@@ -47,7 +47,9 @@ public class CsvSerializer implements IOutputSerializer, Serializable {
try {
StringWriter writer = new StringWriter();
CSVPrinter printer = new CSVPrinter(writer, CSVFormat.RFC4180);
- printer.printRecords(data);
+ for (Object o : data) {
+ printer.print(o);
+ }
//since using StringWriter, we do not need to close it.
return ByteBuffer.wrap(writer.getBuffer().toString().getBytes(StandardCharsets.UTF_8));
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/storm/blob/f14e0b21/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvScheme.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvScheme.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvScheme.java
index ac22bcd..8800078 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvScheme.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvScheme.java
@@ -29,10 +29,11 @@ import java.util.List;
public class TsvScheme implements Scheme {
private final List<String> fieldNames;
- protected char delimiter = '\t';
+ private final char delimiter;
- public TsvScheme(List<String> fieldNames) {
+ public TsvScheme(List<String> fieldNames, char delimiter) {
this.fieldNames = fieldNames;
+ this.delimiter = delimiter;
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/f14e0b21/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvSerializer.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvSerializer.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvSerializer.java
index c9085d2..492064e 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvSerializer.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvSerializer.java
@@ -26,10 +26,11 @@ import java.util.List;
public class TsvSerializer implements IOutputSerializer, Serializable {
private final List<String> fields; //reserved for future
- protected char delimiter = '\t';
+ private final char delimiter;
- public TsvSerializer(List<String> fields) {
- this.fields = fields;
+ public TsvSerializer(List<String> fields, char delimiter) {
+ this.fields = fields;
+ this.delimiter = delimiter;
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/f14e0b21/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/SerdeUtils.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/SerdeUtils.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/SerdeUtils.java
index fc8985f..05dbf67 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/SerdeUtils.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/SerdeUtils.java
@@ -44,7 +44,8 @@ public final class SerdeUtils {
scheme = new JsonScheme(fieldNames);
break;
case "org.apache.storm.sql.runtime.serde.json.TsvScheme" :
- scheme = new TsvScheme(fieldNames);
+ String delimiter = properties.getProperty("tsv.delimiter", "\t");
+ scheme = new TsvScheme(fieldNames, delimiter.charAt(0));
break;
case "org.apache.storm.sql.runtime.serde.json.CsvScheme" :
scheme = new CsvScheme(fieldNames);
@@ -75,7 +76,8 @@ public final class SerdeUtils {
serializer = new JsonSerializer(fieldNames);
break;
case "org.apache.storm.sql.runtime.serde.json.TsvSerializer" :
- serializer = new TsvSerializer(fieldNames);
+ String delimiter = properties.getProperty("tsv.delimiter", "\t");
+ serializer = new TsvSerializer(fieldNames, delimiter.charAt(0));
break;
case "org.apache.storm.sql.runtime.serde.json.CsvSerializer" :
serializer = new CsvSerializer(fieldNames);
http://git-wip-us.apache.org/repos/asf/storm/blob/f14e0b21/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestCsvSerializer.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestCsvSerializer.java b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestCsvSerializer.java
index 1c3e8e5..0108949 100644
--- a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestCsvSerializer.java
+++ b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestCsvSerializer.java
@@ -18,8 +18,8 @@
package org.apache.storm.sql;
import com.google.common.collect.Lists;
-import org.apache.storm.sql.runtime.serde.tsv.TsvScheme;
-import org.apache.storm.sql.runtime.serde.tsv.TsvSerializer;
+import org.apache.storm.sql.runtime.serde.csv.CsvScheme;
+import org.apache.storm.sql.runtime.serde.csv.CsvSerializer;
import org.junit.Test;
import java.nio.ByteBuffer;
@@ -34,20 +34,20 @@ public class TestCsvSerializer {
List<String> fields = Lists.newArrayList("ID", "val");
List<Object> o = Lists.newArrayList("1", "2");
- TsvSerializer serializer = new TsvSerializer(fields);
+ CsvSerializer serializer = new CsvSerializer(fields);
ByteBuffer byteBuffer = serializer.write(o, null);
- TsvScheme scheme = new TsvScheme(fields);
+ CsvScheme scheme = new CsvScheme(fields);
assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
// Fields with embedded commas or double-quote characters
fields = Lists.newArrayList("ID", "val", "v");
o = Lists.newArrayList("1,9", "2,\"3\",5", "\"8\"");
- serializer = new TsvSerializer(fields);
+ serializer = new CsvSerializer(fields);
byteBuffer = serializer.write(o, null);
- scheme = new TsvScheme(fields);
+ scheme = new CsvScheme(fields);
assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
}
http://git-wip-us.apache.org/repos/asf/storm/blob/f14e0b21/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestTsvSerializer.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestTsvSerializer.java b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestTsvSerializer.java
index 07d842f..1798828 100644
--- a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestTsvSerializer.java
+++ b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestTsvSerializer.java
@@ -31,13 +31,15 @@ public class TestTsvSerializer {
@Test
public void testTsvSchemeAndSerializer() {
+ final char delimiter = '\t';
+
List<String> fields = Lists.newArrayList("ID", "val");
List<Object> o = Lists.newArrayList("1", "2");
- TsvSerializer serializer = new TsvSerializer(fields);
+ TsvSerializer serializer = new TsvSerializer(fields, delimiter);
ByteBuffer byteBuffer = serializer.write(o, null);
- TsvScheme scheme = new TsvScheme(fields);
+ TsvScheme scheme = new TsvScheme(fields, delimiter);
assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
}