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:51:34 UTC
[1/4] storm git commit: Merge branch 'STORM-2177' of
https://github.com/vesense/storm into STORM-2173-2177-merge
Repository: storm
Updated Branches:
refs/heads/1.x-branch fe03b706f -> 81164be8e
Merge branch 'STORM-2177' of https://github.com/vesense/storm into STORM-2173-2177-merge
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c967c170
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c967c170
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c967c170
Branch: refs/heads/1.x-branch
Commit: c967c170c0f2d29aa642afa621b1157bb7c2e2d2
Parents: fe03b70
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Nov 14 19:02:57 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Nov 14 21:26:43 2016 +0900
----------------------------------------------------------------------
external/sql/storm-sql-runtime/pom.xml | 5 ++
.../storm/sql/runtime/serde/csv/CsvScheme.java | 70 ++++++++++++++++++++
.../sql/runtime/serde/csv/CsvSerializer.java | 59 +++++++++++++++++
.../storm/sql/runtime/serde/tsv/TsvScheme.java | 58 ++++++++++++++++
.../sql/runtime/serde/tsv/TsvSerializer.java | 54 +++++++++++++++
.../storm/sql/runtime/utils/SerdeUtils.java | 14 ++++
.../apache/storm/sql/runtime/utils/Utils.java | 55 +++++++++++++++
.../org/apache/storm/sql/TestCsvSerializer.java | 54 +++++++++++++++
.../org/apache/storm/sql/TestTsvSerializer.java | 46 +++++++++++++
9 files changed, 415 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/c967c170/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 63520ad..9a63557 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/c967c170/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..34fb1bb
--- /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 the standard RFC4180 CSV Parser
+ * One of 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/c967c170/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..0d3bd74
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/csv/CsvSerializer.java
@@ -0,0 +1,59 @@
+/*
+ * 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 the standard RFC4180 CSV Parser
+ * One of 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);
+ 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) {
+ throw new RuntimeException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/c967c170/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
new file mode 100644
index 0000000..310494c
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvScheme.java
@@ -0,0 +1,58 @@
+/*
+ * 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.tsv;
+
+import com.google.common.base.Preconditions;
+import org.apache.storm.spout.Scheme;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * TsvScheme uses a simple delimited format implemention by splitting string,
+ * and it supports user defined delimiter.
+ */
+public class TsvScheme implements Scheme {
+ private final List<String> fieldNames;
+ private final char delimiter;
+
+ public TsvScheme(List<String> fieldNames, char delimiter) {
+ this.fieldNames = fieldNames;
+ this.delimiter = delimiter;
+ }
+
+ @Override
+ public List<Object> deserialize(ByteBuffer ser) {
+ String data = new String(Utils.toByteArray(ser), StandardCharsets.UTF_8);
+ List<String> parts = org.apache.storm.sql.runtime.utils.Utils.split(data, delimiter);
+ Preconditions.checkArgument(parts.size() == fieldNames.size(), "Invalid schema");
+
+ ArrayList<Object> list = new ArrayList<>(fieldNames.size());
+ list.addAll(parts);
+ return list;
+ }
+
+ @Override
+ public Fields getOutputFields() {
+ return new Fields(fieldNames);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/c967c170/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
new file mode 100644
index 0000000..1cf1c76
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/tsv/TsvSerializer.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.runtime.serde.tsv;
+
+import org.apache.storm.sql.runtime.IOutputSerializer;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+/**
+ * TsvSerializer uses a simple delimited format implemention by splitting string,
+ * and it supports user defined delimiter.
+ */
+public class TsvSerializer implements IOutputSerializer, Serializable {
+ private final List<String> fields; //reserved for future
+ private final char delimiter;
+
+ public TsvSerializer(List<String> fields, char delimiter) {
+ this.fields = fields;
+ this.delimiter = delimiter;
+ }
+
+ @Override
+ public ByteBuffer write(List<Object> data, ByteBuffer buffer) {
+ StringBuilder sb = new StringBuilder(512); // 512: for most scenes to avoid inner array resizing
+ for (int i = 0; i < data.size(); i++) {
+ Object o = data.get(i);
+ if (i == 0) {
+ sb.append(o);
+ } else {
+ sb.append(delimiter);
+ sb.append(o);
+ }
+ }
+ return ByteBuffer.wrap(sb.toString().getBytes(StandardCharsets.UTF_8));
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/c967c170/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 c53fb1f..6b3dfc9 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
@@ -26,8 +26,12 @@ 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;
+import org.apache.storm.sql.runtime.serde.tsv.TsvSerializer;
import org.apache.storm.utils.Utils;
import java.util.ArrayList;
@@ -42,6 +46,11 @@ public final class SerdeUtils {
if (isNotEmpty(inputFormatClass)) {
if (JsonScheme.class.getName().equals(inputFormatClass)) {
scheme = new JsonScheme(fieldNames);
+ } else if (TsvScheme.class.getName().equals(inputFormatClass)) {
+ String delimiter = properties.getProperty("input.tsv.delimiter", "\t");
+ scheme = new TsvScheme(fieldNames, delimiter.charAt(0));
+ } else if (CsvScheme.class.getName().equals(inputFormatClass)) {
+ scheme = new CsvScheme(fieldNames);
} else if (AvroScheme.class.getName().equals(inputFormatClass)) {
String schemaString = properties.getProperty("input.avro.schema");
Preconditions.checkArgument(isNotEmpty(schemaString), "input.avro.schema can not be empty");
@@ -61,6 +70,11 @@ public final class SerdeUtils {
if (isNotEmpty(outputFormatClass)) {
if (JsonSerializer.class.getName().equals(outputFormatClass)) {
serializer = new JsonSerializer(fieldNames);
+ } else if (TsvSerializer.class.getName().equals(outputFormatClass)) {
+ String delimiter = properties.getProperty("output.tsv.delimiter", "\t");
+ serializer = new TsvSerializer(fieldNames, delimiter.charAt(0));
+ } else if (CsvSerializer.class.getName().equals(outputFormatClass)) {
+ serializer = new CsvSerializer(fieldNames);
} else if (AvroSerializer.class.getName().equals(outputFormatClass)) {
String schemaString = properties.getProperty("output.avro.schema");
Preconditions.checkArgument(isNotEmpty(schemaString), "output.avro.schema can not be empty");
http://git-wip-us.apache.org/repos/asf/storm/blob/c967c170/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/Utils.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/Utils.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/Utils.java
new file mode 100644
index 0000000..a0f3af3
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/utils/Utils.java
@@ -0,0 +1,55 @@
+/*
+ * 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.utils;
+
+import java.util.LinkedList;
+import java.util.List;
+
+public final class Utils {
+
+ /**
+ * This method for splitting string into parts by a delimiter
+ * It has higher performance than String.split(String regex)
+ *
+ * @param data
+ * @param delimiter
+ * @return
+ */
+ public static List<String> split(String data, char delimiter){
+ List<String> list = new LinkedList<>();
+ //do not use .toCharArray avoid system copy
+ StringBuilder sb = new StringBuilder(512);
+ int len = data.length();
+ for (int i=0; i < len; i++) {
+ char ch = data.charAt(i);
+ if(ch == delimiter){
+ list.add(sb.toString());
+ sb.setLength(0);
+ if(i == len - 1){
+ list.add("");
+ }
+ }else{
+ sb.append(ch);
+ }
+ }
+ if (sb.length() > 0) {
+ list.add(sb.toString());
+ }
+ return list;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/c967c170/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..0108949
--- /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.csv.CsvScheme;
+import org.apache.storm.sql.runtime.serde.csv.CsvSerializer;
+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");
+
+ CsvSerializer serializer = new CsvSerializer(fields);
+ ByteBuffer byteBuffer = serializer.write(o, null);
+
+ 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 CsvSerializer(fields);
+ byteBuffer = serializer.write(o, null);
+
+ scheme = new CsvScheme(fields);
+ assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/c967c170/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
new file mode 100644
index 0000000..1798828
--- /dev/null
+++ b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestTsvSerializer.java
@@ -0,0 +1,46 @@
+/*
+ * 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 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, delimiter);
+ ByteBuffer byteBuffer = serializer.write(o, null);
+
+ TsvScheme scheme = new TsvScheme(fields, delimiter);
+ assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
+ }
+
+}
[3/4] storm git commit: Merge branch 'STORM-2173-2177-1.x-merge' into
1.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-2173-2177-1.x-merge' into 1.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/bac2e4ee
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/bac2e4ee
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/bac2e4ee
Branch: refs/heads/1.x-branch
Commit: bac2e4ee75c6f5e6d8ada09d5efd9570191752b9
Parents: fe03b70 5009f78
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Nov 14 21:50:26 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Nov 14 21:50:26 2016 +0900
----------------------------------------------------------------------
external/sql/storm-sql-runtime/pom.xml | 5 ++
.../storm/sql/runtime/serde/csv/CsvScheme.java | 70 ++++++++++++++++++++
.../sql/runtime/serde/csv/CsvSerializer.java | 59 +++++++++++++++++
.../storm/sql/runtime/serde/tsv/TsvScheme.java | 58 ++++++++++++++++
.../sql/runtime/serde/tsv/TsvSerializer.java | 54 +++++++++++++++
.../storm/sql/runtime/utils/SerdeUtils.java | 14 ++++
.../apache/storm/sql/runtime/utils/Utils.java | 55 +++++++++++++++
.../org/apache/storm/sql/TestCsvSerializer.java | 67 +++++++++++++++++++
.../org/apache/storm/sql/TestTsvSerializer.java | 52 +++++++++++++++
9 files changed, 434 insertions(+)
----------------------------------------------------------------------
[4/4] storm git commit: add STORM-2173 and STORM-2177 to CHANGELOG
Posted by ka...@apache.org.
add STORM-2173 and STORM-2177 to CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/81164be8
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/81164be8
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/81164be8
Branch: refs/heads/1.x-branch
Commit: 81164be8e2e951c30fa07a6d7545040d12e3e4d2
Parents: bac2e4e
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Nov 14 21:51:13 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Nov 14 21:51:13 2016 +0900
----------------------------------------------------------------------
CHANGELOG.md | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/81164be8/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7483085..ecdd399 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,6 @@
## 1.1.0
+ * STORM-2173: [SQL] Support CSV as input / output format
+ * STORM-2177: [SQL] Support TSV as input / output format
* STORM-2172: [SQL] Support Avro as input / output format
* STORM-2197: NimbusClient connectins leak due to leakage in ThriftClient
* STORM-2185: Storm Supervisor doesn't delete directories properly sometimes
[2/4] storm git commit: STORM-2173 & STORM-2177 test fix for JDK 1.7
Posted by ka...@apache.org.
STORM-2173 & STORM-2177 test fix for JDK 1.7
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5009f788
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5009f788
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5009f788
Branch: refs/heads/1.x-branch
Commit: 5009f7887c206cb81c071b08d4379ecf58e04a3e
Parents: c967c17
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Mon Nov 14 21:49:51 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Mon Nov 14 21:49:51 2016 +0900
----------------------------------------------------------------------
.../org/apache/storm/sql/TestCsvSerializer.java | 21 ++++++++++++++++----
.../org/apache/storm/sql/TestTsvSerializer.java | 10 ++++++++--
2 files changed, 25 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/5009f788/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 0108949..2c9262b 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
@@ -23,6 +23,7 @@ import org.apache.storm.sql.runtime.serde.csv.CsvSerializer;
import org.junit.Test;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertArrayEquals;
@@ -31,8 +32,13 @@ public class TestCsvSerializer {
@Test
public void testCsvSchemeAndSerializer() {
- List<String> fields = Lists.newArrayList("ID", "val");
- List<Object> o = Lists.newArrayList("1", "2");
+ List<String> fields = new ArrayList<>();
+ fields.add("ID");
+ fields.add("val");
+
+ List<Object> o = new ArrayList<>();
+ o.add("1");
+ o.add("2");
CsvSerializer serializer = new CsvSerializer(fields);
ByteBuffer byteBuffer = serializer.write(o, null);
@@ -41,8 +47,15 @@ public class TestCsvSerializer {
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\"");
+ fields = new ArrayList<>();
+ fields.add("ID");
+ fields.add("val");
+ fields.add("v");
+
+ o = new ArrayList<>();
+ o.add("1,9");
+ o.add("2,\"3\",5");
+ o.add("\"8\"");
serializer = new CsvSerializer(fields);
byteBuffer = serializer.write(o, null);
http://git-wip-us.apache.org/repos/asf/storm/blob/5009f788/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 1798828..68b639d 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
@@ -23,6 +23,7 @@ import org.apache.storm.sql.runtime.serde.tsv.TsvSerializer;
import org.junit.Test;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertArrayEquals;
@@ -33,8 +34,13 @@ public class TestTsvSerializer {
public void testTsvSchemeAndSerializer() {
final char delimiter = '\t';
- List<String> fields = Lists.newArrayList("ID", "val");
- List<Object> o = Lists.newArrayList("1", "2");
+ List<String> fields = new ArrayList<>();
+ fields.add("ID");
+ fields.add("val");
+
+ List<Object> o = new ArrayList<>();
+ o.add("1");
+ o.add("2");
TsvSerializer serializer = new TsvSerializer(fields, delimiter);
ByteBuffer byteBuffer = serializer.write(o, null);