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/11 14:15:06 UTC
[4/9] storm git commit: [STORM-2172][SQL] support converting Avro
Utf8 type in map or array to Java String
[STORM-2172][SQL] support converting Avro Utf8 type in map or array to Java String
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/3fe3de7d
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/3fe3de7d
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/3fe3de7d
Branch: refs/heads/master
Commit: 3fe3de7d5a256dcb93435fbae8552558f667e346
Parents: dbc95a0
Author: Xin Wang <be...@163.com>
Authored: Sun Nov 6 08:35:53 2016 +0800
Committer: Xin Wang <be...@163.com>
Committed: Sun Nov 6 09:32:46 2016 +0800
----------------------------------------------------------------------
.../sql/runtime/serde/avro/AvroScheme.java | 16 +++-----
.../sql/runtime/serde/avro/AvroSerializer.java | 3 ++
.../storm/sql/runtime/utils/SerdeUtils.java | 43 +++++++++++++-------
.../apache/storm/sql/TestAvroSerializer.java | 24 +++++++++++
4 files changed, 61 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/3fe3de7d/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroScheme.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroScheme.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroScheme.java
index ccc717b..92a371b 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroScheme.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroScheme.java
@@ -23,7 +23,6 @@ import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
-import org.apache.avro.util.Utf8;
import org.apache.storm.spout.Scheme;
import org.apache.storm.sql.runtime.utils.SerdeUtils;
import org.apache.storm.tuple.Fields;
@@ -33,8 +32,10 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
+/**
+ * AvroScheme uses generic(without code generation) instead of specific(with code generation) readers.
+ */
public class AvroScheme implements Scheme {
private final String schemaString;
private final List<String> fieldNames;
@@ -58,15 +59,8 @@ public class AvroScheme implements Scheme {
ArrayList<Object> list = new ArrayList<>(fieldNames.size());
for (String field : fieldNames) {
Object value = record.get(field);
- // Avro strings are stored using a special Avro type instead of using Java primitives
- if (value instanceof Utf8) {
- list.add(value.toString());
- } else if (value instanceof Map<?, ?>) {
- Map<Object, Object> map = SerdeUtils.convertAvroUtf8Map((Map<Object, Object>)value);
- list.add(map);
- } else {
- list.add(value);
- }
+ // Avro strings are stored using a special Avro Utf8 type instead of using Java primitives
+ list.add(SerdeUtils.convertAvroUtf8(value));
}
return list;
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/storm/blob/3fe3de7d/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroSerializer.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroSerializer.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroSerializer.java
index 141258f..01dcf84 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroSerializer.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/serde/avro/AvroSerializer.java
@@ -33,6 +33,9 @@ import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.List;
+/**
+ * AvroSerializer uses generic(without code generation) instead of specific(with code generation) writers.
+ */
public class AvroSerializer implements IOutputSerializer, Serializable {
private final String schemaString;
private final List<String> fieldNames;
http://git-wip-us.apache.org/repos/asf/storm/blob/3fe3de7d/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 411d90d..0b774ed 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
@@ -20,6 +20,7 @@ package org.apache.storm.sql.runtime.utils;
import static org.apache.commons.lang.StringUtils.isNotEmpty;
import com.google.common.base.Preconditions;
+import org.apache.avro.generic.GenericData;
import org.apache.avro.util.Utf8;
import org.apache.storm.spout.Scheme;
import org.apache.storm.sql.runtime.IOutputSerializer;
@@ -29,6 +30,7 @@ import org.apache.storm.sql.runtime.serde.json.JsonScheme;
import org.apache.storm.sql.runtime.serde.json.JsonSerializer;
import org.apache.storm.utils.Utils;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -79,22 +81,35 @@ public final class SerdeUtils {
return serializer;
}
- public static Map convertAvroUtf8Map(Map<Object, Object> value) {
- Map<Object, Object> map = new HashMap<>();
- for (Map.Entry<Object, Object> entry : value.entrySet()) {
- // Avro only allows maps with Strings for keys, so we only have to worry
- // about deserializing the values
- Object key = entry.getKey().toString();
- Object val = entry.getValue();
+ public static Object convertAvroUtf8(Object value){
+ Object ret;
+ if (value instanceof Utf8) {
+ ret = value.toString();
+ } else if (value instanceof Map<?, ?>) {
+ ret = convertAvroUtf8Map((Map<Object,Object>)value);
+ } else if (value instanceof GenericData.Array) {
+ ret = convertAvroUtf8Array((GenericData.Array)value);
+ } else {
+ ret = value;
+ }
+ return ret;
+ }
- if (val instanceof Utf8) {
- map.put(key, val.toString());
- } else if (val instanceof Map<?, ?>) {
- map.put(key, convertAvroUtf8Map((Map<Object, Object>)val));
- } else {
- map.put(key, val);
- }
+ public static Object convertAvroUtf8Map(Map<Object,Object> value) {
+ Map<Object, Object> map = new HashMap<>(value.size());
+ for (Map.Entry<Object, Object> entry : value.entrySet()) {
+ Object k = convertAvroUtf8(entry.getKey());
+ Object v = convertAvroUtf8(entry.getValue());
+ map.put(k, v);
}
return map;
}
+
+ public static Object convertAvroUtf8Array(GenericData.Array value){
+ List<Object> ls = new ArrayList<>(value.size());
+ for(Object o : value){
+ ls.add(convertAvroUtf8(o));
+ }
+ return ls;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/3fe3de7d/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestAvroSerializer.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestAvroSerializer.java b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestAvroSerializer.java
index 18efd1e..174bfde 100644
--- a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestAvroSerializer.java
+++ b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestAvroSerializer.java
@@ -18,12 +18,14 @@
package org.apache.storm.sql;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import org.apache.storm.sql.runtime.serde.avro.AvroScheme;
import org.apache.storm.sql.runtime.serde.avro.AvroSerializer;
import org.junit.Test;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Map;
import static org.junit.Assert.assertArrayEquals;
@@ -33,6 +35,12 @@ public class TestAvroSerializer {
"\"fields\":[{\"name\":\"ID\",\"type\":\"int\"}," +
"{ \"name\":\"val\", \"type\":\"string\" }]}";
+ private static final String schemaString1 = "{\"type\":\"record\"," +
+ "\"name\":\"avrotest\"," +
+ "\"fields\":[{\"name\":\"ID\",\"type\":\"int\"}," +
+ "{ \"type\":{\"type\":\"map\",\"values\": \"long\"}, \"name\":\"val1\" }," +
+ "{ \"type\":{\"type\":\"array\",\"items\": \"string\"}, \"name\":\"val2\" }]}";
+
@Test
public void testAvroSchemeAndSerializer() {
List<String> fields = Lists.newArrayList("ID", "val");
@@ -45,4 +53,20 @@ public class TestAvroSerializer {
assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
}
+ @Test
+ public void testAvroComplexSchemeAndSerializer() {
+ List<String> fields = Lists.newArrayList("ID", "val1", "val2");
+
+ Map<String,Long> mp = Maps.newHashMap();;
+ mp.put("l1",1234L);
+ mp.put("l2",56789L);
+ List<String> ls = Lists.newArrayList("s1", "s2");
+ List<Object> o = Lists.newArrayList(1, mp, ls);
+
+ AvroSerializer serializer = new AvroSerializer(schemaString1, fields);
+ ByteBuffer byteBuffer = serializer.write(o, null);
+
+ AvroScheme scheme = new AvroScheme(schemaString1, fields);
+ assertArrayEquals(o.toArray(), scheme.deserialize(byteBuffer).toArray());
+ }
}