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:09 UTC
[2/8] storm git commit: [STORM-2173][SQL] Support CSV as input /
output format
[STORM-2173][SQL] Support CSV as input / output format
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d165464b
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d165464b
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d165464b
Branch: refs/heads/master
Commit: d165464b1b3e6c58f37a6632689cf3a3eb80c196
Parents: 2d1785b
Author: vesense <be...@163.com>
Authored: Tue Nov 1 19:26:22 2016 +0800
Committer: vesense <be...@163.com>
Committed: Tue Nov 1 19:26:22 2016 +0800
----------------------------------------------------------------------
external/sql/storm-sql-runtime/pom.xml | 5 ++
.../storm/sql/runtime/serde/csv/CsvScheme.java | 70 ++++++++++++++++++++
.../sql/runtime/serde/csv/CsvSerializer.java | 57 ++++++++++++++++
.../storm/sql/runtime/utils/SerdeUtils.java | 11 ++-
.../org/apache/storm/sql/TestCsvSerializer.java | 54 +++++++++++++++
5 files changed, 196 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d165464b/external/sql/storm-sql-runtime/pom.xml
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/pom.xml b/external/sql/storm-sql-runtime/pom.xml
index 2734ea7..c127067 100644
--- a/external/sql/storm-sql-runtime/pom.xml
+++ b/external/sql/storm-sql-runtime/pom.xml
@@ -79,6 +79,11 @@
<artifactId>commons-lang</artifactId>
</dependency>
<dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-csv</artifactId>
+ <version>1.4</version>
+ </dependency>
+ <dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/storm/blob/d165464b/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
new file mode 100644
index 0000000..b258b40
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvScheme.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.sql.runtime.serde.csv;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVParser;
+import org.apache.commons.csv.CSVRecord;
+import org.apache.storm.spout.Scheme;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * CsvScheme uses standard RFC4180 CSV Parser
+ * The difference from Tsv format is that fields with embedded commas will be quoted.
+ * eg: a,"b,c",d is allowed.
+ *
+ * @see <a href="https://tools.ietf.org/html/rfc4180">RFC4180</a>
+ */
+public class CsvScheme implements Scheme {
+ private final List<String> fieldNames;
+
+ public CsvScheme(List<String> fieldNames) {
+ this.fieldNames = fieldNames;
+ }
+
+ @Override
+ public List<Object> deserialize(ByteBuffer ser) {
+ try {
+ String data = new String(Utils.toByteArray(ser), StandardCharsets.UTF_8);
+ CSVParser parser = CSVParser.parse(data, CSVFormat.RFC4180);
+ CSVRecord record = parser.getRecords().get(0);
+ Preconditions.checkArgument(record.size() == fieldNames.size(), "Invalid schema");
+
+ ArrayList<Object> list = new ArrayList<>(fieldNames.size());
+ for (int i = 0; i < record.size(); i++) {
+ list.add(record.get(i++));
+ }
+ return list;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Fields getOutputFields() {
+ return new Fields(fieldNames);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/d165464b/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
new file mode 100644
index 0000000..0f54c7f
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvSerializer.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.sql.runtime.serde.csv;
+
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVPrinter;
+import org.apache.storm.sql.runtime.IOutputSerializer;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.StringWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+/**
+ * CsvSerializer uses standard RFC4180 CSV Parser
+ * The difference from Tsv format is that fields with embedded commas will be quoted.
+ * eg: a,"b,c",d is allowed.
+ *
+ * @see <a href="https://tools.ietf.org/html/rfc4180">RFC4180</a>
+ */
+public class CsvSerializer implements IOutputSerializer, Serializable {
+ private final List<String> fields; //reserved for future
+
+ public CsvSerializer(List<String> fields) {
+ this.fields = fields;
+ }
+
+ @Override
+ public ByteBuffer write(List<Object> data, ByteBuffer buffer) {
+ try {
+ StringWriter writer = new StringWriter();
+ CSVPrinter printer = new CSVPrinter(writer, CSVFormat.RFC4180);
+ printer.printRecords(data);
+ //since using StringWriter, we do not need to close it.
+ return ByteBuffer.wrap(writer.getBuffer().toString().getBytes(StandardCharsets.UTF_8));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/d165464b/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 d8ac242..fc8985f 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
@@ -24,6 +24,8 @@ import org.apache.storm.spout.Scheme;
import org.apache.storm.sql.runtime.IOutputSerializer;
import org.apache.storm.sql.runtime.serde.avro.AvroScheme;
import org.apache.storm.sql.runtime.serde.avro.AvroSerializer;
+import org.apache.storm.sql.runtime.serde.csv.CsvScheme;
+import org.apache.storm.sql.runtime.serde.csv.CsvSerializer;
import org.apache.storm.sql.runtime.serde.json.JsonScheme;
import org.apache.storm.sql.runtime.serde.json.JsonSerializer;
import org.apache.storm.sql.runtime.serde.tsv.TsvScheme;
@@ -44,13 +46,17 @@ public final class SerdeUtils {
case "org.apache.storm.sql.runtime.serde.json.TsvScheme" :
scheme = new TsvScheme(fieldNames);
break;
+ case "org.apache.storm.sql.runtime.serde.json.CsvScheme" :
+ scheme = new CsvScheme(fieldNames);
+ break;
case "org.apache.storm.sql.runtime.serde.avro.AvroScheme" :
String schemaString = properties.getProperty("avro.schema");
Preconditions.checkArgument(isNotEmpty(schemaString), "avro.schema can not be empty");
scheme = new AvroScheme(schemaString, fieldNames);
break;
default:
- //TODO this can use type.getDeclaredConstructor(List.class).newInstance(fieldNames)
+ //TODO we can use type.getDeclaredConstructor(List.class).newInstance(fieldNames)
+ //and replace above known XXXScheme(fieldNames)
//I will update this after previous PRs getting merged
scheme = Utils.newInstance(inputFormatClass);
}
@@ -71,6 +77,9 @@ public final class SerdeUtils {
case "org.apache.storm.sql.runtime.serde.json.TsvSerializer" :
serializer = new TsvSerializer(fieldNames);
break;
+ case "org.apache.storm.sql.runtime.serde.json.CsvSerializer" :
+ serializer = new CsvSerializer(fieldNames);
+ break;
case "org.apache.storm.sql.runtime.serde.avro.AvroSerializer" :
String schemaString = properties.getProperty("avro.schema");
Preconditions.checkArgument(isNotEmpty(schemaString), "avro.schema can not be empty");
http://git-wip-us.apache.org/repos/asf/storm/blob/d165464b/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
new file mode 100644
index 0000000..1c3e8e5
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestCsvSerializer.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+
+public class TestCsvSerializer {
+
+ @Test
+ public void testCsvSchemeAndSerializer() {
+ List<String> fields = Lists.newArrayList("ID", "val");
+ List<Object> o = Lists.newArrayList("1", "2");
+
+ TsvSerializer serializer = new TsvSerializer(fields);
+ ByteBuffer byteBuffer = serializer.write(o, null);
+
+ TsvScheme scheme = new TsvScheme(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);
+ byteBuffer = serializer.write(o, null);
+
+ scheme = new TsvScheme(fields);
+ assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
+ }
+
+}