You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2015/02/09 14:49:11 UTC
[1/4] flink git commit: [backports] Cleanup and port changes to 0.8
branch.
Repository: flink
Updated Branches:
refs/heads/release-0.8 02b6f85fe -> 944e2e3d5
[backports] Cleanup and port changes to 0.8 branch.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/944e2e3d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/944e2e3d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/944e2e3d
Branch: refs/heads/release-0.8
Commit: 944e2e3d5355e55a9c678158065d9ba81d4190ab
Parents: 95958a9
Author: Robert Metzger <rm...@apache.org>
Authored: Thu Feb 5 14:07:48 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Mon Feb 9 14:48:34 2015 +0100
----------------------------------------------------------------------
.../apache/flink/api/io/avro/AvroPojoTest.java | 178 +++++++++++--------
.../flink/api/java/typeutils/TypeExtractor.java | 4 +-
.../java/typeutils/runtime/KryoSerializer.java | 14 +-
flink-shaded/pom.xml | 2 +-
4 files changed, 114 insertions(+), 84 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/944e2e3d/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
index 6ff4836..b91a3d8 100644
--- a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
+++ b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
@@ -24,29 +24,33 @@ import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.AvroInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.test.util.JavaProgramTestBase;
import org.apache.flink.util.Collector;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
-import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
@RunWith(Parameterized.class)
-public class AvroPojoTest extends MultipleProgramsTestBase {
- public AvroPojoTest(ExecutionMode mode) {
- super(mode);
+public class AvroPojoTest extends JavaProgramTestBase {
+
+ public AvroPojoTest(Configuration config) {
+ super(config);
}
private File inFile;
- private String resultPath;
private String expected;
@Rule
@@ -59,99 +63,129 @@ public class AvroPojoTest extends MultipleProgramsTestBase {
AvroRecordInputFormatTest.writeTestFile(inFile);
}
- @After
- public void after() throws Exception{
- compareResultsByLinesInMemory(expected, resultPath);
- }
- @Test
- public void testSimpleAvroRead() throws Exception {
+ private String testField(final String fieldName) throws Exception {
+ before();
+
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
Path in = new Path(inFile.getAbsoluteFile().toURI());
AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
- DataSet<User> usersDS = env.createInput(users)
- // null map type because the order changes in different JVMs (hard to test)
- .map(new MapFunction<User, User>() {
+ DataSet<User> usersDS = env.createInput(users);
+
+ DataSet<Object> res = usersDS.groupBy(fieldName).reduceGroup(new GroupReduceFunction<User, Object>() {
@Override
- public User map(User value) throws Exception {
- value.setTypeMap(null);
- return value;
+ public void reduce(Iterable<User> values, Collector<Object> out) throws Exception {
+ for(User u : values) {
+ out.collect(u.get(fieldName));
+ }
}
});
+ res.writeAsText(resultPath);
+ env.execute("Simple Avro read job");
+ if(fieldName.equals("name")) {
+ return "Alyssa\nCharlie";
+ } else if(fieldName.equals("type_enum")) {
+ return "GREEN\nRED\n";
+ } else if(fieldName.equals("type_double_test")) {
+ return "123.45\n1.337\n";
+ } else {
+ Assert.fail("Unknown field");
+ }
- usersDS.writeAsText(resultPath);
+ postSubmit();
+ return "";
+ }
- env.execute("Simple Avro read job");
+ private static int NUM_PROGRAMS = 3;
+ private int curProgId = config.getInteger("ProgramId", -1);
+ private String resultPath;
+ private String expectedResult;
- expected = "{\"name\": \"Alyssa\", \"favorite_number\": 256, \"favorite_color\": null, \"type_long_test\": null, \"type_double_test\": 123.45, \"type_null_test\": null, \"type_bool_test\": true, \"type_array_string\": [\"ELEMENT 1\", \"ELEMENT 2\"], \"type_array_boolean\": [true, false], \"type_nullable_array\": null, \"type_enum\": \"GREEN\", \"type_map\": null, \"type_fixed\": null, \"type_union\": null}\n" +
- "{\"name\": \"Charlie\", \"favorite_number\": null, \"favorite_color\": \"blue\", \"type_long_test\": 1337, \"type_double_test\": 1.337, \"type_null_test\": null, \"type_bool_test\": false, \"type_array_string\": [], \"type_array_boolean\": [], \"type_nullable_array\": null, \"type_enum\": \"RED\", \"type_map\": null, \"type_fixed\": null, \"type_union\": null}\n";
+ @Override
+ protected void preSubmit() throws Exception {
+ resultPath = getTempDirPath("result");
}
- @Test
- public void testKeySelection() throws Exception {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Path in = new Path(inFile.getAbsoluteFile().toURI());
-
- AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
- DataSet<User> usersDS = env.createInput(users);
+ @Override
+ protected void testProgram() throws Exception {
+ expectedResult = runProgram(curProgId);
+ }
- DataSet<Tuple2<String, Integer>> res = usersDS.groupBy("name").reduceGroup(new GroupReduceFunction<User, Tuple2<String, Integer>>() {
- @Override
- public void reduce(Iterable<User> values, Collector<Tuple2<String, Integer>> out) throws Exception {
- for(User u : values) {
- out.collect(new Tuple2<String, Integer>(u.getName().toString(), 1));
+ private String runProgram(int curProgId) throws Exception {
+ switch(curProgId) {
+ case 1:
+ for (String fieldName : Arrays.asList("name", "type_enum", "type_double_test")) {
+ return testField(fieldName);
}
- }
- });
+ break;
+ case 2:
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Path in = new Path(inFile.getAbsoluteFile().toURI());
- res.writeAsText(resultPath);
- env.execute("Avro Key selection");
+ AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
+ DataSet<User> usersDS = env.createInput(users);
+ DataSet<Tuple2<String, Integer>> res = usersDS.groupBy("name").reduceGroup(new GroupReduceFunction<User, Tuple2<String, Integer>>() {
+ @Override
+ public void reduce(Iterable<User> values, Collector<Tuple2<String, Integer>> out) throws Exception {
+ for (User u : values) {
+ out.collect(new Tuple2<String, Integer>(u.getName().toString(), 1));
+ }
+ }
+ });
- expected = "(Alyssa,1)\n(Charlie,1)\n";
- }
+ res.writeAsText(resultPath);
+ env.execute("Avro Key selection");
+
+
+ return "(Alyssa,1)\n(Charlie,1)\n";
+ case 3:
+ env = ExecutionEnvironment.getExecutionEnvironment();
+ in = new Path(inFile.getAbsoluteFile().toURI());
+
+ AvroInputFormat<User> users1 = new AvroInputFormat<User>(in, User.class);
+ DataSet<User> usersDS1 = env.createInput(users1)
+ // null map type because the order changes in different JVMs (hard to test)
+ .map(new MapFunction<User, User>() {
+ @Override
+ public User map(User value) throws Exception {
+ value.setTypeMap(null);
+ return value;
+ }
+ });
+
+ usersDS1.writeAsText(resultPath);
+
+ env.execute("Simple Avro read job");
- /**
- * Test some know fields for grouping on
- */
- @Test
- public void testAllFields() throws Exception {
- for(String fieldName : Arrays.asList("name", "type_enum", "type_double_test")) {
- testField(fieldName);
+
+ return "{\"name\": \"Alyssa\", \"favorite_number\": 256, \"favorite_color\": null, \"type_long_test\": null, \"type_double_test\": 123.45, \"type_null_test\": null, \"type_bool_test\": true, \"type_array_string\": [\"ELEMENT 1\", \"ELEMENT 2\"], \"type_array_boolean\": [true, false], \"type_nullable_array\": null, \"type_enum\": \"GREEN\", \"type_map\": null, \"type_fixed\": null, \"type_union\": null}\n" +
+ "{\"name\": \"Charlie\", \"favorite_number\": null, \"favorite_color\": \"blue\", \"type_long_test\": 1337, \"type_double_test\": 1.337, \"type_null_test\": null, \"type_bool_test\": false, \"type_array_string\": [], \"type_array_boolean\": [], \"type_nullable_array\": null, \"type_enum\": \"RED\", \"type_map\": null, \"type_fixed\": null, \"type_union\": null}\n";
+
+ default:
+ throw new RuntimeException("Unknown test");
}
+ return "";
}
- private void testField(final String fieldName) throws Exception {
- before();
+ @Override
+ protected void postSubmit() throws Exception {
+ compareResultsByLinesInMemory(expectedResult, resultPath);
+ }
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- Path in = new Path(inFile.getAbsoluteFile().toURI());
+ @Parameterized.Parameters
+ public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
- AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
- DataSet<User> usersDS = env.createInput(users);
+ LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
- DataSet<Object> res = usersDS.groupBy(fieldName).reduceGroup(new GroupReduceFunction<User, Object>() {
- @Override
- public void reduce(Iterable<User> values, Collector<Object> out) throws Exception {
- for(User u : values) {
- out.collect(u.get(fieldName));
- }
- }
- });
- res.writeAsText(resultPath);
- env.execute("Simple Avro read job");
- if(fieldName.equals("name")) {
- expected = "Alyssa\nCharlie";
- } else if(fieldName.equals("type_enum")) {
- expected = "GREEN\nRED\n";
- } else if(fieldName.equals("type_double_test")) {
- expected = "123.45\n1.337\n";
- } else {
- Assert.fail("Unknown field");
+ for(int i=1; i <= NUM_PROGRAMS; i++) {
+ Configuration config = new Configuration();
+ config.setInteger("ProgramId", i);
+ tConfigs.add(config);
}
- after();
+ return toParameterList(tConfigs);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/944e2e3d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
index 2e90c8a..acf47b6 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
@@ -939,7 +939,7 @@ public class TypeExtractor {
// special case for POJOs generated by Avro.
if(SpecificRecordBase.class.isAssignableFrom(clazz)) {
- return (TypeInformation<OUT>) new AvroTypeInfo(clazz);
+ return (TypeInformation<X>) new AvroTypeInfo(clazz);
}
if (alreadySeen.contains(clazz)) {
@@ -1034,7 +1034,7 @@ public class TypeExtractor {
}
}
- private <X> TypeInformation<X> analyzePojo(Class<X> clazz, ArrayList<Type> typeHierarchy, ParameterizedType clazzTypeHint) {
+ protected <X> TypeInformation<X> analyzePojo(Class<X> clazz, ArrayList<Type> typeHierarchy, ParameterizedType clazzTypeHint) {
// try to create Type hierarchy, if the incoming only contains the most bottom one or none.
if(typeHierarchy.size() <= 1) {
getTypeHierarchy(typeHierarchy, clazz, Object.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/944e2e3d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
index da09242..b73f0b1 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
@@ -21,7 +21,6 @@ package org.apache.flink.api.java.typeutils.runtime;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoException;
import com.esotericsoftware.kryo.Serializer;
-import com.esotericsoftware.kryo.factories.ReflectionSerializerFactory;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.CollectionSerializer;
@@ -31,10 +30,14 @@ import com.twitter.chill.ScalaKryoInstantiator;
import com.twitter.chill.protobuf.ProtobufSerializer;
import com.twitter.chill.thrift.TBaseSerializer;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.specific.SpecificRecordBase;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.thrift.TBase;
+import scala.reflect.ClassTag;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -243,14 +246,7 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
// register the type of our class
kryo.register(type);
-
- // register given types. we do this first so that any registration of a
- // more specific serializer overrides this
- for (Class<?> type : registeredTypes) {
- kryo.register(type);
- }
-
-
+
kryo.setRegistrationRequired(false);
kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/944e2e3d/flink-shaded/pom.xml
----------------------------------------------------------------------
diff --git a/flink-shaded/pom.xml b/flink-shaded/pom.xml
index 51191c7..1f00b62 100644
--- a/flink-shaded/pom.xml
+++ b/flink-shaded/pom.xml
@@ -25,7 +25,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-parent</artifactId>
- <version>0.9-SNAPSHOT</version>
+ <version>0.8-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
[2/4] flink git commit: [FLINK-1392] Add Kryo serializer for Protobuf
Posted by rm...@apache.org.
[FLINK-1392] Add Kryo serializer for Protobuf
Conflicts:
flink-java/pom.xml
flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
Conflicts:
flink-shaded/pom.xml
pom.xml
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c4c3db50
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c4c3db50
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c4c3db50
Branch: refs/heads/release-0.8
Commit: c4c3db50a7aa6a7349189f547ee04c1373df5578
Parents: 84c4998
Author: Robert Metzger <rm...@apache.org>
Authored: Tue Jan 13 10:21:29 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Mon Feb 9 14:48:34 2015 +0100
----------------------------------------------------------------------
flink-java/pom.xml | 33 ++++++++++++++++++++
.../java/typeutils/runtime/KryoSerializer.java | 14 ++++++---
flink-shaded/pom.xml | 5 ++-
pom.xml | 6 +++-
4 files changed, 52 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c4c3db50/flink-java/pom.xml
----------------------------------------------------------------------
diff --git a/flink-java/pom.xml b/flink-java/pom.xml
index 4a4bbfa..cc5664b 100644
--- a/flink-java/pom.xml
+++ b/flink-java/pom.xml
@@ -69,6 +69,39 @@ under the License.
<version>0.5.1</version>
</dependency>
+ <dependency>
+ <groupId>com.twitter</groupId>
+ <artifactId>chill-protobuf</artifactId>
+ <version>0.5.1</version>
+ </dependency>
+ <!-- We need protobuf for chill-protobuf -->
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ <version>2.5.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.twitter</groupId>
+ <artifactId>chill-thrift</artifactId>
+ <version>0.5.1</version>
+ </dependency>
+ <!-- libthrift is required by chill-thrift -->
+ <dependency>
+ <groupId>org.apache.thrift</groupId>
+ <artifactId>libthrift</artifactId>
+ <version>0.6.1</version>
+ <exclusions>
+ <exclusion>
+ <groupId>javax.servlet</groupId>
+ <artifactId>servlet-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<!-- guava needs to be in "provided" scope, to make sure it is not included into the jars by the shading -->
<dependency>
http://git-wip-us.apache.org/repos/asf/flink/blob/c4c3db50/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
index f7a90a5..da09242 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
@@ -26,15 +26,15 @@ import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.CollectionSerializer;
import com.esotericsoftware.kryo.serializers.JavaSerializer;
+import com.google.protobuf.Message;
import com.twitter.chill.ScalaKryoInstantiator;
-import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericData;
-import org.apache.avro.specific.SpecificRecordBase;
+import com.twitter.chill.protobuf.ProtobufSerializer;
+import com.twitter.chill.thrift.TBaseSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
-import scala.reflect.ClassTag;
+import org.apache.thrift.TBase;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -216,6 +216,12 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
// Throwable and all subclasses should be serialized via java serialization
kryo.addDefaultSerializer(Throwable.class, new JavaSerializer());
+ // add serializers for popular other serialization frameworks
+ // Google Protobuf (FLINK-1392)
+ this.kryo.addDefaultSerializer(Message.class, ProtobufSerializer.class);
+ // thrift
+ this.kryo.addDefaultSerializer(TBase.class, TBaseSerializer.class);
+
// If the type we have to serialize as a GenricType is implementing SpecificRecordBase,
// we have to register the avro serializer
// This rule only applies if users explicitly use the GenericTypeInformation for the avro types
http://git-wip-us.apache.org/repos/asf/flink/blob/c4c3db50/flink-shaded/pom.xml
----------------------------------------------------------------------
diff --git a/flink-shaded/pom.xml b/flink-shaded/pom.xml
index 4dbae5d..51191c7 100644
--- a/flink-shaded/pom.xml
+++ b/flink-shaded/pom.xml
@@ -25,7 +25,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-parent</artifactId>
- <version>0.8-SNAPSHOT</version>
+ <version>0.9-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
@@ -71,6 +71,9 @@ under the License.
<relocation>
<pattern>com.google</pattern>
<shadedPattern>org.apache.flink.shaded.com.google</shadedPattern>
+ <excludes>
+ <exclude>com.google.protobuf.**</exclude>
+ </excludes>
</relocation>
</relocations>
</configuration>
http://git-wip-us.apache.org/repos/asf/flink/blob/c4c3db50/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index e59f340..84690b2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -277,7 +277,8 @@ under the License.
</exclusion>
</exclusions>
</dependency>
- <!-- YARN -->
+
+ <!-- Hadoop 2 Dependencies -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
@@ -853,6 +854,9 @@ under the License.
<relocation>
<pattern>com.google</pattern>
<shadedPattern>org.apache.flink.shaded.com.google</shadedPattern>
+ <excludes>
+ <exclude>com.google.protobuf.**</exclude>
+ </excludes>
</relocation>
</relocations>
</configuration>
[3/4] flink git commit: [hotfix] Also use java closure cleaner on
grouped operations
Posted by rm...@apache.org.
[hotfix] Also use java closure cleaner on grouped operations
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/95958a9b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/95958a9b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/95958a9b
Branch: refs/heads/release-0.8
Commit: 95958a9bad36574d12df8d0eb6cfbbe8d7fe8cc5
Parents: c4c3db5
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Jan 15 11:46:53 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Mon Feb 9 14:48:34 2015 +0100
----------------------------------------------------------------------
.../java/org/apache/flink/api/java/operators/SortedGrouping.java | 2 +-
.../org/apache/flink/api/java/operators/UnsortedGrouping.java | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/95958a9b/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java
index 63e5a19..57ee50c 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/SortedGrouping.java
@@ -128,7 +128,7 @@ public class SortedGrouping<T> extends Grouping<T> {
}
TypeInformation<R> resultType = TypeExtractor.getGroupReduceReturnTypes(reducer,
this.getDataSet().getType());
- return new GroupReduceOperator<T, R>(this, resultType, reducer, Utils.getCallLocationName() );
+ return new GroupReduceOperator<T, R>(this, resultType, dataSet.clean(reducer), Utils.getCallLocationName() );
}
http://git-wip-us.apache.org/repos/asf/flink/blob/95958a9b/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java
index d323eae..b8049f6 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/UnsortedGrouping.java
@@ -134,7 +134,7 @@ public class UnsortedGrouping<T> extends Grouping<T> {
if (reducer == null) {
throw new NullPointerException("Reduce function must not be null.");
}
- return new ReduceOperator<T>(this, reducer, Utils.getCallLocationName());
+ return new ReduceOperator<T>(this, dataSet.clean(reducer), Utils.getCallLocationName());
}
/**
@@ -156,7 +156,7 @@ public class UnsortedGrouping<T> extends Grouping<T> {
}
TypeInformation<R> resultType = TypeExtractor.getGroupReduceReturnTypes(reducer, this.getDataSet().getType());
- return new GroupReduceOperator<T, R>(this, resultType, reducer, Utils.getCallLocationName());
+ return new GroupReduceOperator<T, R>(this, resultType, dataSet.clean(reducer), Utils.getCallLocationName());
}
/**
[4/4] flink git commit: [FLINK-1391] Add support for using Avro-POJOs
and Avro types with Kryo
Posted by rm...@apache.org.
[FLINK-1391] Add support for using Avro-POJOs and Avro types with Kryo
Conflicts:
flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
Conflicts:
flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/84c49981
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/84c49981
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/84c49981
Branch: refs/heads/release-0.8
Commit: 84c4998125b175ee524cec4292ab29060784861c
Parents: 02b6f85
Author: Robert Metzger <me...@web.de>
Authored: Mon Jan 12 21:11:09 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Mon Feb 9 14:48:34 2015 +0100
----------------------------------------------------------------------
docs/example_connectors.md | 27 +
docs/programming_guide.md | 2 +-
flink-addons/flink-avro/pom.xml | 30 +-
.../flink/api/avro/EncoderDecoderTest.java | 9 +-
.../org/apache/flink/api/io/avro/.gitignore | 1 +
.../apache/flink/api/io/avro/AvroPojoTest.java | 157 ++++
.../api/io/avro/AvroRecordInputFormatTest.java | 144 +++-
.../flink/api/io/avro/generated/Colors.java | 32 -
.../flink/api/io/avro/generated/User.java | 755 -------------------
.../src/test/resources/avro/user.avsc | 10 +-
.../common/typeutils/ComparatorTestBase.java | 4 +-
flink-java/pom.xml | 9 +-
.../flink/api/java/ExecutionEnvironment.java | 2 +-
.../flink/api/java/typeutils/AvroTypeInfo.java | 72 ++
.../flink/api/java/typeutils/TypeExtractor.java | 8 +-
.../java/typeutils/runtime/KryoSerializer.java | 110 ++-
.../java/typeutils/runtime/PojoComparator.java | 7 +-
pom.xml | 4 +-
tools/maven/suppressions.xml | 1 +
19 files changed, 551 insertions(+), 833 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/docs/example_connectors.md
----------------------------------------------------------------------
diff --git a/docs/example_connectors.md b/docs/example_connectors.md
index 28f35bb..fcd65af 100644
--- a/docs/example_connectors.md
+++ b/docs/example_connectors.md
@@ -69,6 +69,33 @@ users to use all existing Hadoop input formats with Flink.
This section shows some examples for connecting Flink to other systems.
[Read more about Hadoop compatibility in Flink](hadoop_compatibility.html).
+## Avro support in Flink
+
+Flink has extensive build-in support for [Apache Avro](http://avro.apache.org/). This allows to easily read from Avro files with Flink.
+Also, the serialization framework of Flink is able to handle classes generated from Avro schemas.
+
+In order to read data from an Avro file, you have to specify an `AvroInputFormat`.
+
+**Example**:
+
+~~~java
+AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
+DataSet<User> usersDS = env.createInput(users);
+~~~
+
+Note that `User` is a POJO generated by Avro. Flink also allows to perform string-based key selection of these POJOs. For example:
+
+~~~java
+usersDS.groupBy("name")
+~~~
+
+
+Note that using the `GenericData.Record` type is possible with Flink, but not recommended. Since the record contains the full schema, its very data intensive and thus probably slow to use.
+
+Flink's POJO field selection also works with POJOs generated from Avro. However, the usage is only possible if the field types are written correctly to the generated class. If a field is of type `Object` you can not use the field as a join or grouping key.
+Specifying a field in Avro like this `{"name": "type_double_test", "type": "double"},` works fine, however specifying it as a UNION-type with only one field (`{"name": "type_double_test", "type": ["double"]},`) will generate a field of type `Object`. Note that specifying nullable types (`{"name": "type_double_test", "type": ["null", "double"]},`) is possible!
+
+
### Access Microsoft Azure Table Storage
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/docs/programming_guide.md
----------------------------------------------------------------------
diff --git a/docs/programming_guide.md b/docs/programming_guide.md
index 7a2211d..f41fc97 100644
--- a/docs/programming_guide.md
+++ b/docs/programming_guide.md
@@ -982,7 +982,7 @@ These are valid expressions for the example POJO above:
- `complex.word.f2`: Selects the last field in the Tuple3.
- `complex.hadoopCitizen`: Selects a Hadoop-`Writable` type as a key.
-Please note that you can only use types inside POJOs that Flink is able to serialize. Currently, we are using [Avro](http://avro.apache.org) to serialize arbitrary objects (such as `Date`).
+Please note that you can only use types inside POJOs that Flink is able to serialize. Currently, we are using [Kryo](https://github.com/EsotericSoftware/kryo) to serialize arbitrary objects (such as `java.util.Collection`).
### Define key using a Key Selector Function
{:.no_toc}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/pom.xml
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/pom.xml b/flink-addons/flink-avro/pom.xml
index a0d8d9d..dfc9105 100644
--- a/flink-addons/flink-avro/pom.xml
+++ b/flink-addons/flink-avro/pom.xml
@@ -41,7 +41,8 @@ under the License.
<artifactId>flink-java</artifactId>
<version>${project.version}</version>
</dependency>
-
+
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
@@ -53,6 +54,14 @@ under the License.
<artifactId>avro</artifactId>
<!-- version is derived from base module -->
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
@@ -116,7 +125,24 @@ under the License.
</execution>
</executions>
</plugin>
-
+ <!-- Generate Test class from avro schema -->
+ <plugin>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro-maven-plugin</artifactId>
+ <version>1.7.7</version>
+ <executions>
+ <execution>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>schema</goal>
+ </goals>
+ <configuration>
+ <testSourceDirectory>${project.basedir}/src/test/resources/avro</testSourceDirectory>
+ <testOutputDirectory>${project.basedir}/src/test/java/</testOutputDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
<pluginManagement>
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
index 0724457..8f14cb3 100644
--- a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
+++ b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/avro/EncoderDecoderTest.java
@@ -29,9 +29,11 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
+import org.apache.avro.generic.GenericData;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.flink.api.io.avro.generated.Colors;
+import org.apache.flink.api.io.avro.generated.Fixed16;
import org.apache.flink.api.io.avro.generated.User;
import org.apache.flink.util.StringUtils;
import org.junit.Test;
@@ -232,8 +234,11 @@ public class EncoderDecoderTest {
map.put("1", 1L);
map.put("2", 2L);
map.put("3", 3L);
-
- User user = new User("Freudenreich", 1337, "macintosh gray", 1234567890L, 3.1415926, null, true, strings, bools, null, Colors.GREEN, map);
+
+ byte[] b = new byte[16];
+ new Random().nextBytes(b);
+ Fixed16 f = new Fixed16(b);
+ User user = new User("Freudenreich", 1337, "macintosh gray", 1234567890L, 3.1415926, null, true, strings, bools, null, Colors.GREEN, map, f, new Boolean(true));
testObjectSerialization(user);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/.gitignore
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/.gitignore b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/.gitignore
new file mode 100644
index 0000000..dc9b237
--- /dev/null
+++ b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/.gitignore
@@ -0,0 +1 @@
+generated
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
new file mode 100644
index 0000000..6ff4836
--- /dev/null
+++ b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.flink.api.io.avro;
+
+import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.io.avro.generated.User;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.io.AvroInputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.util.Collector;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.File;
+import java.util.Arrays;
+
+@RunWith(Parameterized.class)
+public class AvroPojoTest extends MultipleProgramsTestBase {
+ public AvroPojoTest(ExecutionMode mode) {
+ super(mode);
+ }
+
+ private File inFile;
+ private String resultPath;
+ private String expected;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ public void before() throws Exception{
+ resultPath = tempFolder.newFile().toURI().toString();
+ inFile = tempFolder.newFile();
+ AvroRecordInputFormatTest.writeTestFile(inFile);
+ }
+
+ @After
+ public void after() throws Exception{
+ compareResultsByLinesInMemory(expected, resultPath);
+ }
+
+ @Test
+ public void testSimpleAvroRead() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Path in = new Path(inFile.getAbsoluteFile().toURI());
+
+ AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
+ DataSet<User> usersDS = env.createInput(users)
+ // null map type because the order changes in different JVMs (hard to test)
+ .map(new MapFunction<User, User>() {
+ @Override
+ public User map(User value) throws Exception {
+ value.setTypeMap(null);
+ return value;
+ }
+ });
+
+ usersDS.writeAsText(resultPath);
+
+ env.execute("Simple Avro read job");
+
+
+ expected = "{\"name\": \"Alyssa\", \"favorite_number\": 256, \"favorite_color\": null, \"type_long_test\": null, \"type_double_test\": 123.45, \"type_null_test\": null, \"type_bool_test\": true, \"type_array_string\": [\"ELEMENT 1\", \"ELEMENT 2\"], \"type_array_boolean\": [true, false], \"type_nullable_array\": null, \"type_enum\": \"GREEN\", \"type_map\": null, \"type_fixed\": null, \"type_union\": null}\n" +
+ "{\"name\": \"Charlie\", \"favorite_number\": null, \"favorite_color\": \"blue\", \"type_long_test\": 1337, \"type_double_test\": 1.337, \"type_null_test\": null, \"type_bool_test\": false, \"type_array_string\": [], \"type_array_boolean\": [], \"type_nullable_array\": null, \"type_enum\": \"RED\", \"type_map\": null, \"type_fixed\": null, \"type_union\": null}\n";
+ }
+
+ @Test
+ public void testKeySelection() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Path in = new Path(inFile.getAbsoluteFile().toURI());
+
+ AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
+ DataSet<User> usersDS = env.createInput(users);
+
+ DataSet<Tuple2<String, Integer>> res = usersDS.groupBy("name").reduceGroup(new GroupReduceFunction<User, Tuple2<String, Integer>>() {
+ @Override
+ public void reduce(Iterable<User> values, Collector<Tuple2<String, Integer>> out) throws Exception {
+ for(User u : values) {
+ out.collect(new Tuple2<String, Integer>(u.getName().toString(), 1));
+ }
+ }
+ });
+
+ res.writeAsText(resultPath);
+ env.execute("Avro Key selection");
+
+
+ expected = "(Alyssa,1)\n(Charlie,1)\n";
+ }
+
+ /**
+ * Test some know fields for grouping on
+ */
+ @Test
+ public void testAllFields() throws Exception {
+ for(String fieldName : Arrays.asList("name", "type_enum", "type_double_test")) {
+ testField(fieldName);
+ }
+ }
+
+ private void testField(final String fieldName) throws Exception {
+ before();
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ Path in = new Path(inFile.getAbsoluteFile().toURI());
+
+ AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
+ DataSet<User> usersDS = env.createInput(users);
+
+ DataSet<Object> res = usersDS.groupBy(fieldName).reduceGroup(new GroupReduceFunction<User, Object>() {
+ @Override
+ public void reduce(Iterable<User> values, Collector<Object> out) throws Exception {
+ for(User u : values) {
+ out.collect(u.get(fieldName));
+ }
+ }
+ });
+ res.writeAsText(resultPath);
+ env.execute("Simple Avro read job");
+ if(fieldName.equals("name")) {
+ expected = "Alyssa\nCharlie";
+ } else if(fieldName.equals("type_enum")) {
+ expected = "GREEN\nRED\n";
+ } else if(fieldName.equals("type_double_test")) {
+ expected = "123.45\n1.337\n";
+ } else {
+ Assert.fail("Unknown field");
+ }
+
+ after();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroRecordInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroRecordInputFormatTest.java b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroRecordInputFormatTest.java
index d8d8b46..1ec4a8a 100644
--- a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroRecordInputFormatTest.java
+++ b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroRecordInputFormatTest.java
@@ -27,13 +27,24 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.file.FileReader;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
+import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.util.Utf8;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.ComparatorTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.io.avro.generated.Colors;
import org.apache.flink.api.io.avro.generated.User;
import org.apache.flink.api.java.io.AvroInputFormat;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileInputSplit;
import org.apache.flink.core.fs.Path;
@@ -48,7 +59,7 @@ import org.junit.Test;
*/
public class AvroRecordInputFormatTest {
- private File testFile;
+ public File testFile;
final static String TEST_NAME = "Alyssa";
@@ -65,24 +76,25 @@ public class AvroRecordInputFormatTest {
final static String TEST_MAP_KEY2 = "KEY 2";
final static long TEST_MAP_VALUE2 = 17554L;
- @Before
- public void createFiles() throws IOException {
- testFile = File.createTempFile("AvroInputFormatTest", null);
-
+ private Schema userSchema = new User().getSchema();
+
+
+ public static void writeTestFile(File testFile) throws IOException {
ArrayList<CharSequence> stringArray = new ArrayList<CharSequence>();
stringArray.add(TEST_ARRAY_STRING_1);
stringArray.add(TEST_ARRAY_STRING_2);
-
+
ArrayList<Boolean> booleanArray = new ArrayList<Boolean>();
booleanArray.add(TEST_ARRAY_BOOLEAN_1);
booleanArray.add(TEST_ARRAY_BOOLEAN_2);
-
+
HashMap<CharSequence, Long> longMap = new HashMap<CharSequence, Long>();
longMap.put(TEST_MAP_KEY1, TEST_MAP_VALUE1);
longMap.put(TEST_MAP_KEY2, TEST_MAP_VALUE2);
-
-
+
+
User user1 = new User();
+
user1.setName(TEST_NAME);
user1.setFavoriteNumber(256);
user1.setTypeDoubleTest(123.45d);
@@ -91,22 +103,24 @@ public class AvroRecordInputFormatTest {
user1.setTypeArrayBoolean(booleanArray);
user1.setTypeEnum(TEST_ENUM_COLOR);
user1.setTypeMap(longMap);
-
+
// Construct via builder
User user2 = User.newBuilder()
- .setName("Charlie")
- .setFavoriteColor("blue")
- .setFavoriteNumber(null)
- .setTypeBoolTest(false)
- .setTypeDoubleTest(1.337d)
- .setTypeNullTest(null)
- .setTypeLongTest(1337L)
- .setTypeArrayString(new ArrayList<CharSequence>())
- .setTypeArrayBoolean(new ArrayList<Boolean>())
- .setTypeNullableArray(null)
- .setTypeEnum(Colors.RED)
- .setTypeMap(new HashMap<CharSequence, Long>())
- .build();
+ .setName("Charlie")
+ .setFavoriteColor("blue")
+ .setFavoriteNumber(null)
+ .setTypeBoolTest(false)
+ .setTypeDoubleTest(1.337d)
+ .setTypeNullTest(null)
+ .setTypeLongTest(1337L)
+ .setTypeArrayString(new ArrayList<CharSequence>())
+ .setTypeArrayBoolean(new ArrayList<Boolean>())
+ .setTypeNullableArray(null)
+ .setTypeEnum(Colors.RED)
+ .setTypeMap(new HashMap<CharSequence, Long>())
+ .setTypeFixed(null)
+ .setTypeUnion(null)
+ .build();
DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);
DataFileWriter<User> dataFileWriter = new DataFileWriter<User>(userDatumWriter);
dataFileWriter.create(user1.getSchema(), testFile);
@@ -114,7 +128,17 @@ public class AvroRecordInputFormatTest {
dataFileWriter.append(user2);
dataFileWriter.close();
}
-
+ @Before
+ public void createFiles() throws IOException {
+ testFile = File.createTempFile("AvroInputFormatTest", null);
+ writeTestFile(testFile);
+ }
+
+
+ /**
+ * Test if the AvroInputFormat is able to properly read data from an avro file.
+ * @throws IOException
+ */
@Test
public void testDeserialisation() throws IOException {
Configuration parameters = new Configuration();
@@ -159,9 +183,79 @@ public class AvroRecordInputFormatTest {
format.close();
}
-
+
+ /**
+ * Test if the Flink serialization is able to properly process GenericData.Record types.
+ * Usually users of Avro generate classes (POJOs) from Avro schemas.
+ * However, if generated classes are not available, one can also use GenericData.Record.
+ * It is an untyped key-value record which is using a schema to validate the correctness of the data.
+ *
+ * It is not recommended to use GenericData.Record with Flink. Use generated POJOs instead.
+ */
+ @Test
+ public void testDeserializeToGenericType() throws IOException {
+ DatumReader<GenericData.Record> datumReader = new GenericDatumReader<GenericData.Record>(userSchema);
+
+ FileReader<GenericData.Record> dataFileReader = DataFileReader.openReader(testFile, datumReader);
+ // initialize Record by reading it from disk (thats easier than creating it by hand)
+ GenericData.Record rec = new GenericData.Record(userSchema);
+ dataFileReader.next(rec);
+ // check if record has been read correctly
+ assertNotNull(rec);
+ assertEquals("name not equal", TEST_NAME, rec.get("name").toString() );
+ assertEquals("enum not equal", TEST_ENUM_COLOR.toString(), rec.get("type_enum").toString());
+ assertEquals(null, rec.get("type_long_test")); // it is null for the first record.
+
+ // now serialize it with our framework:
+ TypeInformation<GenericData.Record> te = (TypeInformation<GenericData.Record>) TypeExtractor.createTypeInfo(GenericData.Record.class);
+ TypeSerializer<GenericData.Record> tser = te.createSerializer();
+ ComparatorTestBase.TestOutputView target = new ComparatorTestBase.TestOutputView();
+ tser.serialize(rec, target);
+
+ GenericData.Record newRec = tser.deserialize(target.getInputView());
+
+ // check if it is still the same
+ assertNotNull(newRec);
+ assertEquals("enum not equal", TEST_ENUM_COLOR.toString(), newRec.get("type_enum").toString());
+ assertEquals("name not equal", TEST_NAME, newRec.get("name").toString() );
+ assertEquals(null, newRec.get("type_long_test"));
+
+ }
+
+ /**
+ * This test validates proper serialization with specific (generated POJO) types.
+ */
+ @Test
+ public void testDeserializeToSpecificType() throws IOException {
+
+ DatumReader<User> datumReader = new SpecificDatumReader<User>(userSchema);
+
+ FileReader<User> dataFileReader = DataFileReader.openReader(testFile, datumReader);
+ User rec = dataFileReader.next();
+
+ // check if record has been read correctly
+ assertNotNull(rec);
+ assertEquals("name not equal", TEST_NAME, rec.get("name").toString() );
+ assertEquals("enum not equal", TEST_ENUM_COLOR.toString(), rec.get("type_enum").toString());
+
+ // now serialize it with our framework:
+ TypeInformation<User> te = (TypeInformation<User>) TypeExtractor.createTypeInfo(User.class);
+ TypeSerializer<User> tser = te.createSerializer();
+ ComparatorTestBase.TestOutputView target = new ComparatorTestBase.TestOutputView();
+ tser.serialize(rec, target);
+
+ User newRec = tser.deserialize(target.getInputView());
+
+ // check if it is still the same
+ assertNotNull(newRec);
+ assertEquals("name not equal", TEST_NAME, newRec.getName().toString() );
+ assertEquals("enum not equal", TEST_ENUM_COLOR.toString(), newRec.getTypeEnum().toString() );
+ }
+
+
@After
public void deleteFiles() {
testFile.delete();
}
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/Colors.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/Colors.java b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/Colors.java
deleted file mode 100644
index 58e1f5c..0000000
--- a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/Colors.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * 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.
- */
-
-
-/**
- * Autogenerated by Avro
- *
- * DO NOT EDIT DIRECTLY
- */
-package org.apache.flink.api.io.avro.generated;
-@SuppressWarnings("all")
-@org.apache.avro.specific.AvroGenerated
-public enum Colors {
- RED, GREEN, BLUE ;
- public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"enum\",\"name\":\"Colors\",\"namespace\":\"org.apache.flink.api.io.avro.generated\",\"symbols\":[\"RED\",\"GREEN\",\"BLUE\"]}");
- public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/User.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/User.java b/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/User.java
deleted file mode 100644
index 505857e..0000000
--- a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/User.java
+++ /dev/null
@@ -1,755 +0,0 @@
-/*
- * 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
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * 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.
- */
-
-
-/**
- * Autogenerated by Avro
- *
- * DO NOT EDIT DIRECTLY
- */
-package org.apache.flink.api.io.avro.generated;
-@SuppressWarnings("all")
-@org.apache.avro.specific.AvroGenerated
-public class User extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
- public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"org.apache.flink.api.io.avro.generated\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"favorite_number\",\"type\":[\"int\",\"null\"]},{\"name\":\"favorite_color\",\"type\":[\"string\",\"null\"]},{\"name\":\"type_long_test\",\"type\":[\"long\",\"null\"]},{\"name\":\"type_double_test\",\"type\":[\"double\"]},{\"name\":\"type_null_test\",\"type\":[\"null\"]},{\"name\":\"type_bool_test\",\"type\":[\"boolean\"]},{\"name\":\"type_array_string\",\"type\":{\"type\":\"array\",\"items\":\"string\"}},{\"name\":\"type_array_boolean\",\"type\":{\"type\":\"array\",\"items\":\"boolean\"}},{\"name\":\"type_nullable_array\",\"type\":[\"null\",{\"type\":\"array\",\"items\":\"string\"}],\"default\":null},{\"name\":\"type_enum\",\"type\":{\"type\":\"enum\",\"name\":\"Colors\",\"symbols\":[\"RED\",\"GREEN\",\"BLUE\"]}},{\"name\":\"type
_map\",\"type\":{\"type\":\"map\",\"values\":\"long\"}}]}");
- public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
- @Deprecated public java.lang.CharSequence name;
- @Deprecated public java.lang.Integer favorite_number;
- @Deprecated public java.lang.CharSequence favorite_color;
- @Deprecated public java.lang.Long type_long_test;
- @Deprecated public java.lang.Object type_double_test;
- @Deprecated public java.lang.Object type_null_test;
- @Deprecated public java.lang.Object type_bool_test;
- @Deprecated public java.util.List<java.lang.CharSequence> type_array_string;
- @Deprecated public java.util.List<java.lang.Boolean> type_array_boolean;
- @Deprecated public java.util.List<java.lang.CharSequence> type_nullable_array;
- @Deprecated public org.apache.flink.api.io.avro.generated.Colors type_enum;
- @Deprecated public java.util.Map<java.lang.CharSequence,java.lang.Long> type_map;
-
- /**
- * Default constructor. Note that this does not initialize fields
- * to their default values from the schema. If that is desired then
- * one should use {@link \#newBuilder()}.
- */
- public User() {}
-
- /**
- * All-args constructor.
- */
- public User(java.lang.CharSequence name, java.lang.Integer favorite_number, java.lang.CharSequence favorite_color, java.lang.Long type_long_test, java.lang.Object type_double_test, java.lang.Object type_null_test, java.lang.Object type_bool_test, java.util.List<java.lang.CharSequence> type_array_string, java.util.List<java.lang.Boolean> type_array_boolean, java.util.List<java.lang.CharSequence> type_nullable_array, org.apache.flink.api.io.avro.generated.Colors type_enum, java.util.Map<java.lang.CharSequence,java.lang.Long> type_map) {
- this.name = name;
- this.favorite_number = favorite_number;
- this.favorite_color = favorite_color;
- this.type_long_test = type_long_test;
- this.type_double_test = type_double_test;
- this.type_null_test = type_null_test;
- this.type_bool_test = type_bool_test;
- this.type_array_string = type_array_string;
- this.type_array_boolean = type_array_boolean;
- this.type_nullable_array = type_nullable_array;
- this.type_enum = type_enum;
- this.type_map = type_map;
- }
-
- public org.apache.avro.Schema getSchema() { return SCHEMA$; }
- // Used by DatumWriter. Applications should not call.
- public java.lang.Object get(int field$) {
- switch (field$) {
- case 0: return name;
- case 1: return favorite_number;
- case 2: return favorite_color;
- case 3: return type_long_test;
- case 4: return type_double_test;
- case 5: return type_null_test;
- case 6: return type_bool_test;
- case 7: return type_array_string;
- case 8: return type_array_boolean;
- case 9: return type_nullable_array;
- case 10: return type_enum;
- case 11: return type_map;
- default: throw new org.apache.avro.AvroRuntimeException("Bad index");
- }
- }
- // Used by DatumReader. Applications should not call.
- @SuppressWarnings(value="unchecked")
- public void put(int field$, java.lang.Object value$) {
- switch (field$) {
- case 0: name = (java.lang.CharSequence)value$; break;
- case 1: favorite_number = (java.lang.Integer)value$; break;
- case 2: favorite_color = (java.lang.CharSequence)value$; break;
- case 3: type_long_test = (java.lang.Long)value$; break;
- case 4: type_double_test = (java.lang.Object)value$; break;
- case 5: type_null_test = (java.lang.Object)value$; break;
- case 6: type_bool_test = (java.lang.Object)value$; break;
- case 7: type_array_string = (java.util.List<java.lang.CharSequence>)value$; break;
- case 8: type_array_boolean = (java.util.List<java.lang.Boolean>)value$; break;
- case 9: type_nullable_array = (java.util.List<java.lang.CharSequence>)value$; break;
- case 10: type_enum = (org.apache.flink.api.io.avro.generated.Colors)value$; break;
- case 11: type_map = (java.util.Map<java.lang.CharSequence,java.lang.Long>)value$; break;
- default: throw new org.apache.avro.AvroRuntimeException("Bad index");
- }
- }
-
- /**
- * Gets the value of the 'name' field.
- */
- public java.lang.CharSequence getName() {
- return name;
- }
-
- /**
- * Sets the value of the 'name' field.
- * @param value the value to set.
- */
- public void setName(java.lang.CharSequence value) {
- this.name = value;
- }
-
- /**
- * Gets the value of the 'favorite_number' field.
- */
- public java.lang.Integer getFavoriteNumber() {
- return favorite_number;
- }
-
- /**
- * Sets the value of the 'favorite_number' field.
- * @param value the value to set.
- */
- public void setFavoriteNumber(java.lang.Integer value) {
- this.favorite_number = value;
- }
-
- /**
- * Gets the value of the 'favorite_color' field.
- */
- public java.lang.CharSequence getFavoriteColor() {
- return favorite_color;
- }
-
- /**
- * Sets the value of the 'favorite_color' field.
- * @param value the value to set.
- */
- public void setFavoriteColor(java.lang.CharSequence value) {
- this.favorite_color = value;
- }
-
- /**
- * Gets the value of the 'type_long_test' field.
- */
- public java.lang.Long getTypeLongTest() {
- return type_long_test;
- }
-
- /**
- * Sets the value of the 'type_long_test' field.
- * @param value the value to set.
- */
- public void setTypeLongTest(java.lang.Long value) {
- this.type_long_test = value;
- }
-
- /**
- * Gets the value of the 'type_double_test' field.
- */
- public java.lang.Object getTypeDoubleTest() {
- return type_double_test;
- }
-
- /**
- * Sets the value of the 'type_double_test' field.
- * @param value the value to set.
- */
- public void setTypeDoubleTest(java.lang.Object value) {
- this.type_double_test = value;
- }
-
- /**
- * Gets the value of the 'type_null_test' field.
- */
- public java.lang.Object getTypeNullTest() {
- return type_null_test;
- }
-
- /**
- * Sets the value of the 'type_null_test' field.
- * @param value the value to set.
- */
- public void setTypeNullTest(java.lang.Object value) {
- this.type_null_test = value;
- }
-
- /**
- * Gets the value of the 'type_bool_test' field.
- */
- public java.lang.Object getTypeBoolTest() {
- return type_bool_test;
- }
-
- /**
- * Sets the value of the 'type_bool_test' field.
- * @param value the value to set.
- */
- public void setTypeBoolTest(java.lang.Object value) {
- this.type_bool_test = value;
- }
-
- /**
- * Gets the value of the 'type_array_string' field.
- */
- public java.util.List<java.lang.CharSequence> getTypeArrayString() {
- return type_array_string;
- }
-
- /**
- * Sets the value of the 'type_array_string' field.
- * @param value the value to set.
- */
- public void setTypeArrayString(java.util.List<java.lang.CharSequence> value) {
- this.type_array_string = value;
- }
-
- /**
- * Gets the value of the 'type_array_boolean' field.
- */
- public java.util.List<java.lang.Boolean> getTypeArrayBoolean() {
- return type_array_boolean;
- }
-
- /**
- * Sets the value of the 'type_array_boolean' field.
- * @param value the value to set.
- */
- public void setTypeArrayBoolean(java.util.List<java.lang.Boolean> value) {
- this.type_array_boolean = value;
- }
-
- /**
- * Gets the value of the 'type_nullable_array' field.
- */
- public java.util.List<java.lang.CharSequence> getTypeNullableArray() {
- return type_nullable_array;
- }
-
- /**
- * Sets the value of the 'type_nullable_array' field.
- * @param value the value to set.
- */
- public void setTypeNullableArray(java.util.List<java.lang.CharSequence> value) {
- this.type_nullable_array = value;
- }
-
- /**
- * Gets the value of the 'type_enum' field.
- */
- public org.apache.flink.api.io.avro.generated.Colors getTypeEnum() {
- return type_enum;
- }
-
- /**
- * Sets the value of the 'type_enum' field.
- * @param value the value to set.
- */
- public void setTypeEnum(org.apache.flink.api.io.avro.generated.Colors value) {
- this.type_enum = value;
- }
-
- /**
- * Gets the value of the 'type_map' field.
- */
- public java.util.Map<java.lang.CharSequence,java.lang.Long> getTypeMap() {
- return type_map;
- }
-
- /**
- * Sets the value of the 'type_map' field.
- * @param value the value to set.
- */
- public void setTypeMap(java.util.Map<java.lang.CharSequence,java.lang.Long> value) {
- this.type_map = value;
- }
-
- /** Creates a new User RecordBuilder */
- public static org.apache.flink.api.io.avro.generated.User.Builder newBuilder() {
- return new org.apache.flink.api.io.avro.generated.User.Builder();
- }
-
- /** Creates a new User RecordBuilder by copying an existing Builder */
- public static org.apache.flink.api.io.avro.generated.User.Builder newBuilder(org.apache.flink.api.io.avro.generated.User.Builder other) {
- return new org.apache.flink.api.io.avro.generated.User.Builder(other);
- }
-
- /** Creates a new User RecordBuilder by copying an existing User instance */
- public static org.apache.flink.api.io.avro.generated.User.Builder newBuilder(org.apache.flink.api.io.avro.generated.User other) {
- return new org.apache.flink.api.io.avro.generated.User.Builder(other);
- }
-
- /**
- * RecordBuilder for User instances.
- */
- public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<User>
- implements org.apache.avro.data.RecordBuilder<User> {
-
- private java.lang.CharSequence name;
- private java.lang.Integer favorite_number;
- private java.lang.CharSequence favorite_color;
- private java.lang.Long type_long_test;
- private java.lang.Object type_double_test;
- private java.lang.Object type_null_test;
- private java.lang.Object type_bool_test;
- private java.util.List<java.lang.CharSequence> type_array_string;
- private java.util.List<java.lang.Boolean> type_array_boolean;
- private java.util.List<java.lang.CharSequence> type_nullable_array;
- private org.apache.flink.api.io.avro.generated.Colors type_enum;
- private java.util.Map<java.lang.CharSequence,java.lang.Long> type_map;
-
- /** Creates a new Builder */
- private Builder() {
- super(org.apache.flink.api.io.avro.generated.User.SCHEMA$);
- }
-
- /** Creates a Builder by copying an existing Builder */
- private Builder(org.apache.flink.api.io.avro.generated.User.Builder other) {
- super(other);
- if (isValidValue(fields()[0], other.name)) {
- this.name = data().deepCopy(fields()[0].schema(), other.name);
- fieldSetFlags()[0] = true;
- }
- if (isValidValue(fields()[1], other.favorite_number)) {
- this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number);
- fieldSetFlags()[1] = true;
- }
- if (isValidValue(fields()[2], other.favorite_color)) {
- this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color);
- fieldSetFlags()[2] = true;
- }
- if (isValidValue(fields()[3], other.type_long_test)) {
- this.type_long_test = data().deepCopy(fields()[3].schema(), other.type_long_test);
- fieldSetFlags()[3] = true;
- }
- if (isValidValue(fields()[4], other.type_double_test)) {
- this.type_double_test = data().deepCopy(fields()[4].schema(), other.type_double_test);
- fieldSetFlags()[4] = true;
- }
- if (isValidValue(fields()[5], other.type_null_test)) {
- this.type_null_test = data().deepCopy(fields()[5].schema(), other.type_null_test);
- fieldSetFlags()[5] = true;
- }
- if (isValidValue(fields()[6], other.type_bool_test)) {
- this.type_bool_test = data().deepCopy(fields()[6].schema(), other.type_bool_test);
- fieldSetFlags()[6] = true;
- }
- if (isValidValue(fields()[7], other.type_array_string)) {
- this.type_array_string = data().deepCopy(fields()[7].schema(), other.type_array_string);
- fieldSetFlags()[7] = true;
- }
- if (isValidValue(fields()[8], other.type_array_boolean)) {
- this.type_array_boolean = data().deepCopy(fields()[8].schema(), other.type_array_boolean);
- fieldSetFlags()[8] = true;
- }
- if (isValidValue(fields()[9], other.type_nullable_array)) {
- this.type_nullable_array = data().deepCopy(fields()[9].schema(), other.type_nullable_array);
- fieldSetFlags()[9] = true;
- }
- if (isValidValue(fields()[10], other.type_enum)) {
- this.type_enum = data().deepCopy(fields()[10].schema(), other.type_enum);
- fieldSetFlags()[10] = true;
- }
- if (isValidValue(fields()[11], other.type_map)) {
- this.type_map = data().deepCopy(fields()[11].schema(), other.type_map);
- fieldSetFlags()[11] = true;
- }
- }
-
- /** Creates a Builder by copying an existing User instance */
- private Builder(org.apache.flink.api.io.avro.generated.User other) {
- super(org.apache.flink.api.io.avro.generated.User.SCHEMA$);
- if (isValidValue(fields()[0], other.name)) {
- this.name = data().deepCopy(fields()[0].schema(), other.name);
- fieldSetFlags()[0] = true;
- }
- if (isValidValue(fields()[1], other.favorite_number)) {
- this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number);
- fieldSetFlags()[1] = true;
- }
- if (isValidValue(fields()[2], other.favorite_color)) {
- this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color);
- fieldSetFlags()[2] = true;
- }
- if (isValidValue(fields()[3], other.type_long_test)) {
- this.type_long_test = data().deepCopy(fields()[3].schema(), other.type_long_test);
- fieldSetFlags()[3] = true;
- }
- if (isValidValue(fields()[4], other.type_double_test)) {
- this.type_double_test = data().deepCopy(fields()[4].schema(), other.type_double_test);
- fieldSetFlags()[4] = true;
- }
- if (isValidValue(fields()[5], other.type_null_test)) {
- this.type_null_test = data().deepCopy(fields()[5].schema(), other.type_null_test);
- fieldSetFlags()[5] = true;
- }
- if (isValidValue(fields()[6], other.type_bool_test)) {
- this.type_bool_test = data().deepCopy(fields()[6].schema(), other.type_bool_test);
- fieldSetFlags()[6] = true;
- }
- if (isValidValue(fields()[7], other.type_array_string)) {
- this.type_array_string = data().deepCopy(fields()[7].schema(), other.type_array_string);
- fieldSetFlags()[7] = true;
- }
- if (isValidValue(fields()[8], other.type_array_boolean)) {
- this.type_array_boolean = data().deepCopy(fields()[8].schema(), other.type_array_boolean);
- fieldSetFlags()[8] = true;
- }
- if (isValidValue(fields()[9], other.type_nullable_array)) {
- this.type_nullable_array = data().deepCopy(fields()[9].schema(), other.type_nullable_array);
- fieldSetFlags()[9] = true;
- }
- if (isValidValue(fields()[10], other.type_enum)) {
- this.type_enum = data().deepCopy(fields()[10].schema(), other.type_enum);
- fieldSetFlags()[10] = true;
- }
- if (isValidValue(fields()[11], other.type_map)) {
- this.type_map = data().deepCopy(fields()[11].schema(), other.type_map);
- fieldSetFlags()[11] = true;
- }
- }
-
- /** Gets the value of the 'name' field */
- public java.lang.CharSequence getName() {
- return name;
- }
-
- /** Sets the value of the 'name' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setName(java.lang.CharSequence value) {
- validate(fields()[0], value);
- this.name = value;
- fieldSetFlags()[0] = true;
- return this;
- }
-
- /** Checks whether the 'name' field has been set */
- public boolean hasName() {
- return fieldSetFlags()[0];
- }
-
- /** Clears the value of the 'name' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearName() {
- name = null;
- fieldSetFlags()[0] = false;
- return this;
- }
-
- /** Gets the value of the 'favorite_number' field */
- public java.lang.Integer getFavoriteNumber() {
- return favorite_number;
- }
-
- /** Sets the value of the 'favorite_number' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setFavoriteNumber(java.lang.Integer value) {
- validate(fields()[1], value);
- this.favorite_number = value;
- fieldSetFlags()[1] = true;
- return this;
- }
-
- /** Checks whether the 'favorite_number' field has been set */
- public boolean hasFavoriteNumber() {
- return fieldSetFlags()[1];
- }
-
- /** Clears the value of the 'favorite_number' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearFavoriteNumber() {
- favorite_number = null;
- fieldSetFlags()[1] = false;
- return this;
- }
-
- /** Gets the value of the 'favorite_color' field */
- public java.lang.CharSequence getFavoriteColor() {
- return favorite_color;
- }
-
- /** Sets the value of the 'favorite_color' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setFavoriteColor(java.lang.CharSequence value) {
- validate(fields()[2], value);
- this.favorite_color = value;
- fieldSetFlags()[2] = true;
- return this;
- }
-
- /** Checks whether the 'favorite_color' field has been set */
- public boolean hasFavoriteColor() {
- return fieldSetFlags()[2];
- }
-
- /** Clears the value of the 'favorite_color' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearFavoriteColor() {
- favorite_color = null;
- fieldSetFlags()[2] = false;
- return this;
- }
-
- /** Gets the value of the 'type_long_test' field */
- public java.lang.Long getTypeLongTest() {
- return type_long_test;
- }
-
- /** Sets the value of the 'type_long_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeLongTest(java.lang.Long value) {
- validate(fields()[3], value);
- this.type_long_test = value;
- fieldSetFlags()[3] = true;
- return this;
- }
-
- /** Checks whether the 'type_long_test' field has been set */
- public boolean hasTypeLongTest() {
- return fieldSetFlags()[3];
- }
-
- /** Clears the value of the 'type_long_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeLongTest() {
- type_long_test = null;
- fieldSetFlags()[3] = false;
- return this;
- }
-
- /** Gets the value of the 'type_double_test' field */
- public java.lang.Object getTypeDoubleTest() {
- return type_double_test;
- }
-
- /** Sets the value of the 'type_double_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeDoubleTest(java.lang.Object value) {
- validate(fields()[4], value);
- this.type_double_test = value;
- fieldSetFlags()[4] = true;
- return this;
- }
-
- /** Checks whether the 'type_double_test' field has been set */
- public boolean hasTypeDoubleTest() {
- return fieldSetFlags()[4];
- }
-
- /** Clears the value of the 'type_double_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeDoubleTest() {
- type_double_test = null;
- fieldSetFlags()[4] = false;
- return this;
- }
-
- /** Gets the value of the 'type_null_test' field */
- public java.lang.Object getTypeNullTest() {
- return type_null_test;
- }
-
- /** Sets the value of the 'type_null_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeNullTest(java.lang.Object value) {
- validate(fields()[5], value);
- this.type_null_test = value;
- fieldSetFlags()[5] = true;
- return this;
- }
-
- /** Checks whether the 'type_null_test' field has been set */
- public boolean hasTypeNullTest() {
- return fieldSetFlags()[5];
- }
-
- /** Clears the value of the 'type_null_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeNullTest() {
- type_null_test = null;
- fieldSetFlags()[5] = false;
- return this;
- }
-
- /** Gets the value of the 'type_bool_test' field */
- public java.lang.Object getTypeBoolTest() {
- return type_bool_test;
- }
-
- /** Sets the value of the 'type_bool_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeBoolTest(java.lang.Object value) {
- validate(fields()[6], value);
- this.type_bool_test = value;
- fieldSetFlags()[6] = true;
- return this;
- }
-
- /** Checks whether the 'type_bool_test' field has been set */
- public boolean hasTypeBoolTest() {
- return fieldSetFlags()[6];
- }
-
- /** Clears the value of the 'type_bool_test' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeBoolTest() {
- type_bool_test = null;
- fieldSetFlags()[6] = false;
- return this;
- }
-
- /** Gets the value of the 'type_array_string' field */
- public java.util.List<java.lang.CharSequence> getTypeArrayString() {
- return type_array_string;
- }
-
- /** Sets the value of the 'type_array_string' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeArrayString(java.util.List<java.lang.CharSequence> value) {
- validate(fields()[7], value);
- this.type_array_string = value;
- fieldSetFlags()[7] = true;
- return this;
- }
-
- /** Checks whether the 'type_array_string' field has been set */
- public boolean hasTypeArrayString() {
- return fieldSetFlags()[7];
- }
-
- /** Clears the value of the 'type_array_string' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeArrayString() {
- type_array_string = null;
- fieldSetFlags()[7] = false;
- return this;
- }
-
- /** Gets the value of the 'type_array_boolean' field */
- public java.util.List<java.lang.Boolean> getTypeArrayBoolean() {
- return type_array_boolean;
- }
-
- /** Sets the value of the 'type_array_boolean' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeArrayBoolean(java.util.List<java.lang.Boolean> value) {
- validate(fields()[8], value);
- this.type_array_boolean = value;
- fieldSetFlags()[8] = true;
- return this;
- }
-
- /** Checks whether the 'type_array_boolean' field has been set */
- public boolean hasTypeArrayBoolean() {
- return fieldSetFlags()[8];
- }
-
- /** Clears the value of the 'type_array_boolean' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeArrayBoolean() {
- type_array_boolean = null;
- fieldSetFlags()[8] = false;
- return this;
- }
-
- /** Gets the value of the 'type_nullable_array' field */
- public java.util.List<java.lang.CharSequence> getTypeNullableArray() {
- return type_nullable_array;
- }
-
- /** Sets the value of the 'type_nullable_array' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeNullableArray(java.util.List<java.lang.CharSequence> value) {
- validate(fields()[9], value);
- this.type_nullable_array = value;
- fieldSetFlags()[9] = true;
- return this;
- }
-
- /** Checks whether the 'type_nullable_array' field has been set */
- public boolean hasTypeNullableArray() {
- return fieldSetFlags()[9];
- }
-
- /** Clears the value of the 'type_nullable_array' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeNullableArray() {
- type_nullable_array = null;
- fieldSetFlags()[9] = false;
- return this;
- }
-
- /** Gets the value of the 'type_enum' field */
- public org.apache.flink.api.io.avro.generated.Colors getTypeEnum() {
- return type_enum;
- }
-
- /** Sets the value of the 'type_enum' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeEnum(org.apache.flink.api.io.avro.generated.Colors value) {
- validate(fields()[10], value);
- this.type_enum = value;
- fieldSetFlags()[10] = true;
- return this;
- }
-
- /** Checks whether the 'type_enum' field has been set */
- public boolean hasTypeEnum() {
- return fieldSetFlags()[10];
- }
-
- /** Clears the value of the 'type_enum' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeEnum() {
- type_enum = null;
- fieldSetFlags()[10] = false;
- return this;
- }
-
- /** Gets the value of the 'type_map' field */
- public java.util.Map<java.lang.CharSequence,java.lang.Long> getTypeMap() {
- return type_map;
- }
-
- /** Sets the value of the 'type_map' field */
- public org.apache.flink.api.io.avro.generated.User.Builder setTypeMap(java.util.Map<java.lang.CharSequence,java.lang.Long> value) {
- validate(fields()[11], value);
- this.type_map = value;
- fieldSetFlags()[11] = true;
- return this;
- }
-
- /** Checks whether the 'type_map' field has been set */
- public boolean hasTypeMap() {
- return fieldSetFlags()[11];
- }
-
- /** Clears the value of the 'type_map' field */
- public org.apache.flink.api.io.avro.generated.User.Builder clearTypeMap() {
- type_map = null;
- fieldSetFlags()[11] = false;
- return this;
- }
-
- @Override
- public User build() {
- try {
- User record = new User();
- record.name = fieldSetFlags()[0] ? this.name : (java.lang.CharSequence) defaultValue(fields()[0]);
- record.favorite_number = fieldSetFlags()[1] ? this.favorite_number : (java.lang.Integer) defaultValue(fields()[1]);
- record.favorite_color = fieldSetFlags()[2] ? this.favorite_color : (java.lang.CharSequence) defaultValue(fields()[2]);
- record.type_long_test = fieldSetFlags()[3] ? this.type_long_test : (java.lang.Long) defaultValue(fields()[3]);
- record.type_double_test = fieldSetFlags()[4] ? this.type_double_test : (java.lang.Object) defaultValue(fields()[4]);
- record.type_null_test = fieldSetFlags()[5] ? this.type_null_test : (java.lang.Object) defaultValue(fields()[5]);
- record.type_bool_test = fieldSetFlags()[6] ? this.type_bool_test : (java.lang.Object) defaultValue(fields()[6]);
- record.type_array_string = fieldSetFlags()[7] ? this.type_array_string : (java.util.List<java.lang.CharSequence>) defaultValue(fields()[7]);
- record.type_array_boolean = fieldSetFlags()[8] ? this.type_array_boolean : (java.util.List<java.lang.Boolean>) defaultValue(fields()[8]);
- record.type_nullable_array = fieldSetFlags()[9] ? this.type_nullable_array : (java.util.List<java.lang.CharSequence>) defaultValue(fields()[9]);
- record.type_enum = fieldSetFlags()[10] ? this.type_enum : (org.apache.flink.api.io.avro.generated.Colors) defaultValue(fields()[10]);
- record.type_map = fieldSetFlags()[11] ? this.type_map : (java.util.Map<java.lang.CharSequence,java.lang.Long>) defaultValue(fields()[11]);
- return record;
- } catch (Exception e) {
- throw new org.apache.avro.AvroRuntimeException(e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-addons/flink-avro/src/test/resources/avro/user.avsc
----------------------------------------------------------------------
diff --git a/flink-addons/flink-avro/src/test/resources/avro/user.avsc b/flink-addons/flink-avro/src/test/resources/avro/user.avsc
index af3cb75..6801b10 100644
--- a/flink-addons/flink-avro/src/test/resources/avro/user.avsc
+++ b/flink-addons/flink-avro/src/test/resources/avro/user.avsc
@@ -1,5 +1,5 @@
-{"namespace": "org.apache.flink.api.java.record.io.avro.generated",
+{"namespace": "org.apache.flink.api.io.avro.generated",
"type": "record",
"name": "User",
"fields": [
@@ -7,13 +7,17 @@
{"name": "favorite_number", "type": ["int", "null"]},
{"name": "favorite_color", "type": ["string", "null"]},
{"name": "type_long_test", "type": ["long", "null"]},
- {"name": "type_double_test", "type": ["double"]},
+ {"name": "type_double_test", "type": "double"},
{"name": "type_null_test", "type": ["null"]},
{"name": "type_bool_test", "type": ["boolean"]},
{"name": "type_array_string", "type" : {"type" : "array", "items" : "string"}},
{"name": "type_array_boolean", "type" : {"type" : "array", "items" : "boolean"}},
{"name": "type_nullable_array", "type": ["null", {"type":"array", "items":"string"}], "default":null},
{"name": "type_enum", "type": {"type": "enum", "name": "Colors", "symbols" : ["RED", "GREEN", "BLUE"]}},
- {"name": "type_map", "type": {"type": "map", "values": "long"}}
+ {"name": "type_map", "type": {"type": "map", "values": "long"}},
+ {"name": "type_fixed",
+ "size": 16,
+ "type": ["null", {"name": "Fixed16", "size": 16, "type": "fixed"}] },
+ {"name": "type_union", "type": ["null", "boolean", "long", "double"]}
]
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
index e6a8cb6..bc5c6b6 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
@@ -448,7 +448,7 @@ public abstract class ComparatorTestBase<T> {
}
// --------------------------------------------------------------------------------------------
- protected static final class TestOutputView extends DataOutputStream implements DataOutputView {
+ public static final class TestOutputView extends DataOutputStream implements DataOutputView {
public TestOutputView() {
super(new ByteArrayOutputStream(4096));
@@ -474,7 +474,7 @@ public abstract class ComparatorTestBase<T> {
}
}
- protected static final class TestInputView extends DataInputStream implements DataInputView {
+ public static final class TestInputView extends DataInputStream implements DataInputView {
public TestInputView(byte[] data) {
super(new ByteArrayInputStream(data));
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-java/pom.xml
----------------------------------------------------------------------
diff --git a/flink-java/pom.xml b/flink-java/pom.xml
index 1469dac..4a4bbfa 100644
--- a/flink-java/pom.xml
+++ b/flink-java/pom.xml
@@ -63,6 +63,13 @@ under the License.
<version>0.5.1</version>
</dependency>
+ <dependency>
+ <groupId>com.twitter</groupId>
+ <artifactId>chill-avro_2.10</artifactId>
+ <version>0.5.1</version>
+ </dependency>
+
+
<!-- guava needs to be in "provided" scope, to make sure it is not included into the jars by the shading -->
<dependency>
<groupId>com.google.guava</groupId>
@@ -80,7 +87,7 @@ under the License.
</dependency>
</dependencies>
- <!-- Because flink-scala uses it in tests -->
+ <!-- Because flink-scala and flink-avro uses it in tests -->
<build>
<plugins>
<plugin>
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index c19e9aa..563787f 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -371,7 +371,7 @@ public abstract class ExecutionEnvironment {
}
catch (Exception e) {
throw new InvalidProgramException("The type returned by the input format could not be automatically determined. " +
- "Please specify the TypeInformation of the produced type explicitly.");
+ "Please specify the TypeInformation of the produced type explicitly.", e);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-java/src/main/java/org/apache/flink/api/java/typeutils/AvroTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/AvroTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/AvroTypeInfo.java
new file mode 100644
index 0000000..ccdf7f7
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/AvroTypeInfo.java
@@ -0,0 +1,72 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.flink.api.java.typeutils;
+
+import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Special type information to generate a POJO type info from an avro schema.
+ *
+ * Proceeding: It uses a regular pojo type analysis and replaces all GenericType<CharSequence>
+ * with a GenericType<avro.Utf8>
+ * @param <T>
+ */
+public class AvroTypeInfo<T extends SpecificRecordBase> extends PojoTypeInfo<T> {
+ public AvroTypeInfo(Class<T> typeClass) {
+ super(typeClass, generateFieldsFromAvroSchema(typeClass));
+ }
+
+ private static <T extends SpecificRecordBase> List<PojoField> generateFieldsFromAvroSchema(Class<T> typeClass) {
+ PojoTypeExtractor pte = new PojoTypeExtractor();
+ TypeInformation ti = pte.analyzePojo(typeClass, new ArrayList<Type>(), null);
+
+ if(!(ti instanceof PojoTypeInfo)) {
+ throw new IllegalStateException("Expecting type to be a PojoTypeInfo");
+ }
+ PojoTypeInfo pti = (PojoTypeInfo) ti;
+ List<PojoField> newFields = new ArrayList<PojoField>(pti.getTotalFields());
+
+ for(int i = 0; i < pti.getTotalFields(); i++) {
+ PojoField f = pti.getPojoFieldAt(i);
+ TypeInformation newType = f.type;
+ // check if type is a CharSequence
+ if(newType instanceof GenericTypeInfo) {
+ if((newType).getTypeClass().equals(CharSequence.class)) {
+ // replace the type by a org.apache.avro.util.Utf8
+ newType = new GenericTypeInfo(org.apache.avro.util.Utf8.class);
+ }
+ }
+ PojoField newField = new PojoField(f.field, newType);
+ newFields.add(newField);
+ }
+ return newFields;
+ }
+
+ private static class PojoTypeExtractor extends TypeExtractor {
+ private PojoTypeExtractor() {
+ super();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
index b528d00..2e90c8a 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
@@ -30,6 +30,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import org.apache.avro.specific.SpecificRecordBase;
import org.apache.commons.lang3.Validate;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.CrossFunction;
@@ -66,7 +67,7 @@ public class TypeExtractor {
// in an endless recursion
private Set<Class<?>> alreadySeen;
- private TypeExtractor() {
+ protected TypeExtractor() {
alreadySeen = new HashSet<Class<?>>();
}
@@ -936,6 +937,11 @@ public class TypeExtractor {
return (TypeInformation<X>) new EnumTypeInfo(clazz);
}
+ // special case for POJOs generated by Avro.
+ if(SpecificRecordBase.class.isAssignableFrom(clazz)) {
+ return (TypeInformation<OUT>) new AvroTypeInfo(clazz);
+ }
+
if (alreadySeen.contains(clazz)) {
return new GenericTypeInfo<X>(clazz);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
index a5f5f65..f7a90a5 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
@@ -20,21 +20,39 @@ package org.apache.flink.api.java.typeutils.runtime;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoException;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.factories.ReflectionSerializerFactory;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.serializers.CollectionSerializer;
import com.esotericsoftware.kryo.serializers.JavaSerializer;
import com.twitter.chill.ScalaKryoInstantiator;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.specific.SpecificRecordBase;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
+import scala.reflect.ClassTag;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.EOFException;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
import java.lang.reflect.Modifier;
+/**
+ * A type serializer that serializes its type using the Kryo serialization
+ * framework (https://github.com/EsotericSoftware/kryo).
+ *
+ * This serializer is intended as a fallback serializer for the cases that are
+ * not covered by the basic types, tuples, and POJOs.
+ *
+ * @param <T> The type to be serialized.
+ */
public class KryoSerializer<T> extends TypeSerializer<T> {
private static final long serialVersionUID = 2L;
@@ -48,6 +66,8 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
private transient Input input;
private transient Output output;
+
+ // ------------------------------------------------------------------------
public KryoSerializer(Class<T> type){
if(type == null){
@@ -56,6 +76,7 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
this.type = type;
}
+ // ------------------------------------------------------------------------
@Override
public boolean isImmutableType() {
@@ -191,15 +212,92 @@ public class KryoSerializer<T> extends TypeSerializer<T> {
private void checkKryoInitialized() {
if (this.kryo == null) {
this.kryo = new ScalaKryoInstantiator().newKryo();
- this.kryo.addDefaultSerializer(Throwable.class, new JavaSerializer());
- this.kryo.setRegistrationRequired(false);
- this.kryo.register(type);
- this.kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
+
+ // Throwable and all subclasses should be serialized via java serialization
+ kryo.addDefaultSerializer(Throwable.class, new JavaSerializer());
+
+ // If the type we have to serialize as a GenricType is implementing SpecificRecordBase,
+ // we have to register the avro serializer
+ // This rule only applies if users explicitly use the GenericTypeInformation for the avro types
+ // usually, we are able to handle Avro POJOs with the POJO serializer.
+ if(SpecificRecordBase.class.isAssignableFrom(type)) {
+ ClassTag<SpecificRecordBase> tag = scala.reflect.ClassTag$.MODULE$.apply(type);
+ this.kryo.register(type, com.twitter.chill.avro.AvroSerializer.SpecificRecordSerializer(tag));
+
+ }
+ // Avro POJOs contain java.util.List which have GenericData.Array as their runtime type
+ // because Kryo is not able to serialize them properly, we use this serializer for them
+ this.kryo.register(GenericData.Array.class, new SpecificInstanceCollectionSerializer(ArrayList.class));
+ // We register this serializer for users who want to use untyped Avro records (GenericData.Record).
+ // Kryo is able to serialize everything in there, except for the Schema.
+ // This serializer is very slow, but using the GenericData.Records of Kryo is in general a bad idea.
+ // we add the serializer as a default serializer because Avro is using a private sub-type at runtime.
+ this.kryo.addDefaultSerializer(Schema.class, new AvroSchemaSerializer());
+
+
+ // register the type of our class
+ kryo.register(type);
+
+ // register given types. we do this first so that any registration of a
+ // more specific serializer overrides this
+ for (Class<?> type : registeredTypes) {
+ kryo.register(type);
+ }
+
+
+ kryo.setRegistrationRequired(false);
+ kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Custom Serializers
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Special serializer for Java collections enforcing certain instance types.
+ * Avro is serializing collections with an "GenericData.Array" type. Kryo is not able to handle
+ * this type, so we use ArrayLists.
+ */
+ public static class SpecificInstanceCollectionSerializer<T extends Collection> extends CollectionSerializer {
+ Class<T> type;
+ public SpecificInstanceCollectionSerializer(Class<T> type) {
+ this.type = type;
+ }
+
+ @Override
+ protected Collection create(Kryo kryo, Input input, Class<Collection> type) {
+ return kryo.newInstance(this.type);
+ }
+
+ @Override
+ protected Collection createCopy(Kryo kryo, Collection original) {
+ return kryo.newInstance(this.type);
+ }
+ }
+
+ /**
+ * Slow serialization approach for Avro schemas.
+ * This is only used with {{@link org.apache.avro.generic.GenericData.Record}} types.
+ * Having this serializer, we are able to handle avro Records.
+ */
+ public static class AvroSchemaSerializer extends Serializer<Schema> {
+ @Override
+ public void write(Kryo kryo, Output output, Schema object) {
+ String schemaAsString = object.toString(false);
+ output.writeString(schemaAsString);
+ }
+
+ @Override
+ public Schema read(Kryo kryo, Input input, Class<Schema> type) {
+ String schemaAsString = input.readString();
+ // the parser seems to be stateful, to we need a new one for every type.
+ Schema.Parser sParser = new Schema.Parser();
+ return sParser.parse(schemaAsString);
}
}
-
// --------------------------------------------------------------------------------------------
- // for testing
+ // For testing
// --------------------------------------------------------------------------------------------
Kryo getKryo() {
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
index ae4a806..c0c7797 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
@@ -210,7 +210,12 @@ public final class PojoComparator<T> extends CompositeTypeComparator<T> implemen
int code = 0;
for (; i < this.keyFields.length; i++) {
code *= TupleComparatorBase.HASH_SALT[i & 0x1F];
- code += this.comparators[i].hash(accessField(keyFields[i], value));
+ try {
+ code += this.comparators[i].hash(accessField(keyFields[i], value));
+ }catch(NullPointerException npe) {
+ throw new RuntimeException("A NullPointerException occured while accessing a key field in a POJO. " +
+ "Most likely, the value grouped/joined on is null. Field name: "+keyFields[i].getName(), npe);
+ }
}
return code;
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index b729e17..e59f340 100644
--- a/pom.xml
+++ b/pom.xml
@@ -923,7 +923,9 @@ under the License.
<exclude>flink-runtime/resources/web-docs-infoserver/js/timeline.js</exclude>
<!-- Test Data. -->
<exclude>flink-tests/src/test/resources/testdata/terainput.txt</exclude>
- <exclude>flink-addons/flink-avro/src/test/resources/avro/user.avsc</exclude>
+ <exclude>flink-addons/flink-avro/src/test/resources/avro/*.avsc</exclude>
+ <!-- Generated files -->
+ <exclude>flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/*.java</exclude>
<!-- Configuration Files. -->
<exclude>**/flink-bin/conf/slaves</exclude>
<!-- Administrative files in the main trunk. -->
http://git-wip-us.apache.org/repos/asf/flink/blob/84c49981/tools/maven/suppressions.xml
----------------------------------------------------------------------
diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml
index 377cbfd..b17dbce 100644
--- a/tools/maven/suppressions.xml
+++ b/tools/maven/suppressions.xml
@@ -24,4 +24,5 @@ under the License.
<suppressions>
<suppress files="org[\\/]apache[\\/]flink[\\/]api[\\/]io[\\/]avro[\\/]example[\\/]User.java" checks="[a-zA-Z0-9]*"/>
+ <suppress files="org[\\/]apache[\\/]flink[\\/]api[\\/]io[\\/]avro[\\/]generated[\\/].*.java" checks="[a-zA-Z0-9]*"/>
</suppressions>
\ No newline at end of file