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/18 16:37:23 UTC
[3/7] flink git commit: [FLINK-1417] Automatically register types
with Kryo
[FLINK-1417] Automatically register types with Kryo
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/354efec0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/354efec0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/354efec0
Branch: refs/heads/master
Commit: 354efec0f9da0fa03ea9b337b02a1a2a03a9ac16
Parents: 5015ab4
Author: Robert Metzger <rm...@apache.org>
Authored: Mon Jan 26 17:54:25 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed Feb 18 15:52:08 2015 +0100
----------------------------------------------------------------------
.gitignore | 1 +
.../org/apache/flink/api/io/avro/.gitignore | 1 -
.../apache/flink/api/io/avro/AvroPojoTest.java | 157 ------
.../flink/api/common/ExecutionConfig.java | 193 ++++++-
.../api/common/typeutils/CompositeType.java | 3 +-
.../api/common/typeutils/TypeSerializer.java | 2 +-
flink-dist/src/main/flink-bin/LICENSE | 2 +
.../src/main/resources/log4j.properties | 23 +
flink-java/pom.xml | 19 +-
.../flink/api/java/ExecutionEnvironment.java | 102 +++-
.../java/org/apache/flink/api/java/Utils.java | 25 +
.../flink/api/java/typeutils/AvroTypeInfo.java | 12 +-
.../api/java/typeutils/GenericTypeInfo.java | 2 +-
.../flink/api/java/typeutils/TypeExtractor.java | 12 +-
.../java/typeutils/runtime/KryoSerializer.java | 291 -----------
.../java/typeutils/runtime/PojoSerializer.java | 3 +-
.../typeutils/runtime/kryo/KryoSerializer.java | 312 +++++++++++
.../typeutils/runtime/kryo/Serializers.java | 242 +++++++++
.../AbstractGenericTypeSerializerTest.java | 3 -
.../runtime/KryoGenericArraySerializerTest.java | 29 --
.../runtime/KryoGenericTypeComparatorTest.java | 29 --
.../runtime/KryoGenericTypeSerializerTest.java | 165 ------
.../runtime/KryoVersusAvroMinibenchmark.java | 520 ------------------
.../runtime/KryoWithCustomSerializersTest.java | 74 ---
.../SubclassFromInterfaceSerializerTest.java | 1 +
.../kryo/KryoGenericArraySerializerTest.java | 30 ++
.../kryo/KryoGenericTypeComparatorTest.java | 30 ++
.../kryo/KryoGenericTypeSerializerTest.java | 169 ++++++
.../kryo/KryoVersusAvroMinibenchmark.java | 522 +++++++++++++++++++
.../kryo/KryoWithCustomSerializersTest.java | 74 +++
.../typeutils/runtime/kryo/SerializersTest.java | 66 +++
.../src/main/resources/log4j.properties | 23 +
.../src/main/resources/log4j.properties | 23 +
.../flink/api/scala/ExecutionEnvironment.scala | 39 +-
.../api/scala/typeutils/TrySerializer.scala | 2 +-
.../flink/api/scala/typeutils/TryTypeInfo.scala | 2 +-
.../apache/flink/api/io/avro/AvroPojoTest.java | 164 ++++++
.../api/io/avro/AvroRecordInputFormatTest.java | 19 +-
.../io/avro/AvroSplittableInputFormatTest.java | 11 +-
.../environment/StreamExecutionEnvironment.java | 38 +-
.../api/scala/StreamExecutionEnvironment.scala | 37 +-
flink-tests/pom.xml | 13 -
.../javaApiOperators/GroupReduceITCase.java | 17 +-
.../runtime/KryoGenericTypeSerializerTest.scala | 9 +-
.../ScalaSpecialTypesSerializerTest.scala | 2 +-
.../api/scala/runtime/TupleSerializerTest.scala | 3 +-
pom.xml | 7 +-
47 files changed, 2123 insertions(+), 1400 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index 522cff5..deaa1ee 100644
--- a/.gitignore
+++ b/.gitignore
@@ -19,3 +19,4 @@ tmp
_site
docs/api
build-target
+flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/generated/
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/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
deleted file mode 100644
index dc9b237..0000000
--- a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/.gitignore
+++ /dev/null
@@ -1 +0,0 @@
-generated
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/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
deleted file mode 100644
index 6ff4836..0000000
--- a/flink-addons/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
+++ /dev/null
@@ -1,157 +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.
- */
-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/354efec0/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index 5fa01b7..4a583e1 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
@@ -21,10 +21,8 @@ package org.apache.flink.api.common;
import com.esotericsoftware.kryo.Serializer;
import java.io.Serializable;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
+import java.util.ArrayList;
+import java.util.List;
/**
* A configuration config for configuring behavior of the system, such as whether to use
@@ -44,12 +42,17 @@ public class ExecutionConfig implements Serializable {
private boolean objectReuse = false;
+ private boolean disableAutoTypeRegistration = false;
+
// Serializers and types registered with Kryo and the PojoSerializer
- private final Map<Class<?>, Serializer<?>> registeredKryoSerializers = new HashMap<Class<?>, Serializer<?>>();
- private final Map<Class<?>, Class<? extends Serializer<?>>> registeredKryoSerializersClasses = new HashMap<Class<?>, Class<? extends Serializer<?>>>();
- private final Set<Class<?>> registeredKryoTypes = new HashSet<Class<?>>();
- private final Set<Class<?>> registeredPojoTypes = new HashSet<Class<?>>();
+ // we store them in lists to ensure they are registered in order in all kryo instances.
+ private final List<Entry<Class<?>, Serializer<?>>> registeredTypesWithKryoSerializers = new ArrayList<Entry<Class<?>, Serializer<?>>>();
+ private final List<Entry<Class<?>, Class<? extends Serializer<?>>>> registeredTypesWithKryoSerializerClasses = new ArrayList<Entry<Class<?>, Class<? extends Serializer<?>>>>();
+ private final List<Entry<Class<?>, Serializer<?>>> defaultKryoSerializers = new ArrayList<Entry<Class<?>, Serializer<?>>>();
+ private final List<Entry<Class<?>, Class<? extends Serializer<?>>>> defaultKryoSerializerClasses = new ArrayList<Entry<Class<?>, Class<? extends Serializer<?>>>>();
+ private final List<Class<?>> registeredKryoTypes = new ArrayList<Class<?>>();
+ private final List<Class<?>> registeredPojoTypes = new ArrayList<Class<?>>();
/**
* Enables the ClosureCleaner. This analyzes user code functions and sets fields to null
@@ -191,9 +194,10 @@ public class ExecutionConfig implements Serializable {
// Registry for types and serializers
// --------------------------------------------------------------------------------------------
+
+
/**
- * Registers the given Serializer as a default serializer for the given type at the
- * {@link org.apache.flink.api.common.typeutils.runtime.KryoSerializer}.
+ * Adds a new Kryo default serializer to the Runtime.
*
* Note that the serializer instance must be serializable (as defined by java.io.Serializable),
* because it may be distributed to the worker nodes by java serialization.
@@ -201,7 +205,7 @@ public class ExecutionConfig implements Serializable {
* @param type The class of the types serialized with the given serializer.
* @param serializer The serializer to use.
*/
- public void registerKryoSerializer(Class<?> type, Serializer<?> serializer) {
+ public void addDefaultKryoSerializer(Class<?> type, Serializer<?> serializer) {
if (type == null || serializer == null) {
throw new NullPointerException("Cannot register null class or serializer.");
}
@@ -210,23 +214,66 @@ public class ExecutionConfig implements Serializable {
+ "as defined by java.io.Serializable. For stateless serializers, you can use the "
+ "'registerSerializer(Class, Class)' method to register the serializer via its class.");
}
-
- registeredKryoSerializers.put(type, serializer);
+ Entry<Class<?>, Serializer<?>> e = new Entry<Class<?>, Serializer<?>>(type, serializer);
+ if(!defaultKryoSerializers.contains(e)) {
+ defaultKryoSerializers.add(e);
+ }
}
/**
- * Registers the given Serializer via its class as a serializer for the given type at the
- * {@link org.apache.flink.api.common.typeutils.runtime.KryoSerializer}.
+ * Adds a new Kryo default serializer to the Runtime.
*
* @param type The class of the types serialized with the given serializer.
* @param serializerClass The class of the serializer to use.
*/
- public void registerKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass) {
+ public void addDefaultKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass) {
if (type == null || serializerClass == null) {
throw new NullPointerException("Cannot register null class or serializer.");
}
+ Entry<Class<?>, Class<? extends Serializer<?>>> e = new Entry<Class<?>, Class<? extends Serializer<?>>>(type, serializerClass);
+ if(!defaultKryoSerializerClasses.contains(e)) {
+ defaultKryoSerializerClasses.add(e);
+ }
+ }
+
+ /**
+ * Registers the given type with a Kryo Serializer.
+ *
+ * Note that the serializer instance must be serializable (as defined by java.io.Serializable),
+ * because it may be distributed to the worker nodes by java serialization.
+ *
+ * @param type The class of the types serialized with the given serializer.
+ * @param serializer The serializer to use.
+ */
+ public void registerTypeWithKryoSerializer(Class<?> type, Serializer<?> serializer) {
+ if (type == null || serializer == null) {
+ throw new NullPointerException("Cannot register null class or serializer.");
+ }
+ if (!(serializer instanceof java.io.Serializable)) {
+ throw new IllegalArgumentException("The serializer instance must be serializable, (for distributing it in the cluster), "
+ + "as defined by java.io.Serializable. For stateless serializers, you can use the "
+ + "'registerSerializer(Class, Class)' method to register the serializer via its class.");
+ }
+ Entry<Class<?>, Serializer<?>> e = new Entry<Class<?>, Serializer<?>>(type, serializer);
+ if(!registeredTypesWithKryoSerializers.contains(e)) {
+ registeredTypesWithKryoSerializers.add(e);
+ }
+ }
- registeredKryoSerializersClasses.put(type, serializerClass);
+ /**
+ * Registers the given Serializer via its class as a serializer for the given type at the KryoSerializer
+ *
+ * @param type The class of the types serialized with the given serializer.
+ * @param serializerClass The class of the serializer to use.
+ */
+ public void registerTypeWithKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass) {
+ if (type == null || serializerClass == null) {
+ throw new NullPointerException("Cannot register null class or serializer.");
+ }
+ Entry<Class<?>, Class<? extends Serializer<?>>> e = new Entry<Class<?>, Class<? extends Serializer<?>>>(type, serializerClass);
+ if(!registeredTypesWithKryoSerializerClasses.contains(e)) {
+ registeredTypesWithKryoSerializerClasses.add(e);
+ }
}
/**
@@ -241,7 +288,9 @@ public class ExecutionConfig implements Serializable {
if (type == null) {
throw new NullPointerException("Cannot register null type class.");
}
- registeredPojoTypes.add(type);
+ if(!registeredPojoTypes.contains(type)) {
+ registeredPojoTypes.add(type);
+ }
}
/**
@@ -260,29 +309,48 @@ public class ExecutionConfig implements Serializable {
}
/**
- * Returns the registered Kryo Serializers.
+ * Returns the registered types with Kryo Serializers.
*/
- public Map<Class<?>, Serializer<?>> getRegisteredKryoSerializers() {
- return registeredKryoSerializers;
+ public List<Entry<Class<?>, Serializer<?>>> getRegisteredTypesWithKryoSerializers() {
+ return registeredTypesWithKryoSerializers;
}
/**
- * Returns the registered Kryo Serializer classes.
+ * Returns the registered types with their Kryo Serializer classes.
*/
- public Map<Class<?>, Class<? extends Serializer<?>>> getRegisteredKryoSerializersClasses() {
- return registeredKryoSerializersClasses;
+ public List<Entry<Class<?>, Class<? extends Serializer<?>>>> getRegisteredTypesWithKryoSerializerClasses() {
+ return registeredTypesWithKryoSerializerClasses;
+ }
+
+
+ /**
+ * Returns the registered default Kryo Serializers.
+ */
+ public List<Entry<Class<?>, Serializer<?>>> getDefaultKryoSerializers() {
+ return defaultKryoSerializers;
+ }
+
+ /**
+ * Returns the registered default Kryo Serializer classes.
+ */
+ public List<Entry<Class<?>, Class<? extends Serializer<?>>>> getDefaultKryoSerializerClasses() {
+ return defaultKryoSerializerClasses;
}
/**
* Returns the registered Kryo types.
*/
- public Set<Class<?>> getRegisteredKryoTypes() {
+ public List<Class<?>> getRegisteredKryoTypes() {
if (isForceKryoEnabled()) {
// if we force kryo, we must also return all the types that
// were previously only registered as POJO
- Set<Class<?>> result = new HashSet<Class<?>>();
+ List<Class<?>> result = new ArrayList<Class<?>>();
result.addAll(registeredKryoTypes);
- result.addAll(registeredPojoTypes);
+ for(Class<?> t : registeredPojoTypes) {
+ if (!result.contains(t)) {
+ result.add(t);
+ }
+ }
return result;
} else {
return registeredKryoTypes;
@@ -292,7 +360,76 @@ public class ExecutionConfig implements Serializable {
/**
* Returns the registered POJO types.
*/
- public Set<Class<?>> getRegisteredPojoTypes() {
+ public List<Class<?>> getRegisteredPojoTypes() {
return registeredPojoTypes;
}
+
+
+ public boolean isDisableAutoTypeRegistration() {
+ return disableAutoTypeRegistration;
+ }
+
+ /**
+ * Control whether Flink is automatically registering all types in the user programs with
+ * Kryo.
+ *
+ * @param disableAutoTypeRegistration
+ */
+ public void setDisableAutoTypeRegistration(boolean disableAutoTypeRegistration) {
+ this.disableAutoTypeRegistration = disableAutoTypeRegistration;
+ }
+
+
+ public static class Entry<K, V> implements Serializable {
+ private final K k;
+ private final V v;
+ public Entry(K k, V v) {
+ this.k = k;
+ this.v = v;
+ }
+
+ public K getKey() {
+ return k;
+ }
+
+ public V getValue() {
+ return v;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ Entry entry = (Entry) o;
+
+ if (k != null ? !k.equals(entry.k) : entry.k != null) {
+ return false;
+ }
+ if (v != null ? !v.equals(entry.v) : entry.v != null) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = k != null ? k.hashCode() : 0;
+ result = 31 * result + (v != null ? v.hashCode() : 0);
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return "Entry{" +
+ "k=" + k +
+ ", v=" + v +
+ '}';
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
index 17b8dab..3bd1fff 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
@@ -132,9 +132,8 @@ public abstract class CompositeType<T> extends TypeInformation<T> {
}
return getNewComparator(config);
}
-
-
+
public static class FlatFieldDescriptor {
private int keyPosition;
private TypeInformation<?> type;
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
index 329e826..542b059 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
@@ -42,7 +42,7 @@ public abstract class TypeSerializer<T> implements Serializable {
// --------------------------------------------------------------------------------------------
// General information about the type and the serializer
// --------------------------------------------------------------------------------------------
-
+
/**
* Gets whether the type is an immutable type.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-dist/src/main/flink-bin/LICENSE
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/LICENSE b/flink-dist/src/main/flink-bin/LICENSE
index 25f91d4..89d8eca 100644
--- a/flink-dist/src/main/flink-bin/LICENSE
+++ b/flink-dist/src/main/flink-bin/LICENSE
@@ -233,7 +233,9 @@ under the Apache License (v 2.0):
- Apache Kafka (http://kafka.apache.org)
- Apache Flume (http://flume.apache.org)
- Apache Sling (http://sling.apache.org)
+ - Apache Thrift (http://thrift.apache.org)
- Google Guava (https://code.google.com/p/guava-libraries/)
+ - Google Protocol Buffers (https://github.com/google/protobuf/)
- Netty v4.0.21 (http://netty.io)
- Powermock (http://www.powermock.org)
- Javassist (http://www.javassist.org)
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-examples/flink-java-examples/src/main/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/flink-examples/flink-java-examples/src/main/resources/log4j.properties b/flink-examples/flink-java-examples/src/main/resources/log4j.properties
new file mode 100644
index 0000000..da32ea0
--- /dev/null
+++ b/flink-examples/flink-java-examples/src/main/resources/log4j.properties
@@ -0,0 +1,23 @@
+################################################################################
+# 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.
+################################################################################
+
+log4j.rootLogger=INFO, console
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/pom.xml
----------------------------------------------------------------------
diff --git a/flink-java/pom.xml b/flink-java/pom.xml
index dc5076a..640c527 100644
--- a/flink-java/pom.xml
+++ b/flink-java/pom.xml
@@ -33,6 +33,9 @@ under the License.
<name>flink-java</name>
<packaging>jar</packaging>
+ <properties>
+ <chill.version>0.5.2</chill.version>
+ </properties>
<dependencies>
<dependency>
@@ -60,19 +63,19 @@ under the License.
<dependency>
<groupId>com.twitter</groupId>
<artifactId>chill_2.10</artifactId>
- <version>0.5.1</version>
+ <version>${chill.version}</version>
</dependency>
<dependency>
<groupId>com.twitter</groupId>
<artifactId>chill-avro_2.10</artifactId>
- <version>0.5.1</version>
+ <version>${chill.version}</version>
</dependency>
<dependency>
<groupId>com.twitter</groupId>
<artifactId>chill-protobuf</artifactId>
- <version>0.5.1</version>
+ <version>${chill.version}</version>
</dependency>
<!-- We need protobuf for chill-protobuf -->
<dependency>
@@ -84,7 +87,7 @@ under the License.
<dependency>
<groupId>com.twitter</groupId>
<artifactId>chill-thrift</artifactId>
- <version>0.5.1</version>
+ <version>${chill.version}</version>
</dependency>
<!-- libthrift is required by chill-thrift -->
<dependency>
@@ -112,7 +115,6 @@ under the License.
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
- <version>2.7</version>
</dependency>
<!-- guava needs to be in "provided" scope, to make sure it is not included into the jars by the shading -->
@@ -130,13 +132,6 @@ under the License.
<type>test-jar</type>
<scope>test</scope>
</dependency>
-
- <dependency>
- <groupId>joda-time</groupId>
- <artifactId>joda-time</artifactId>
- <scope>test</scope>
- </dependency>
-
</dependencies>
<!-- Because flink-scala and flink-avro uses it in tests -->
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/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 1105ab9..21e1f24 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
@@ -30,6 +30,7 @@ import java.util.UUID;
import com.esotericsoftware.kryo.Serializer;
+import com.google.common.base.Joiner;
import org.apache.commons.lang3.Validate;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.InvalidProgramException;
@@ -38,8 +39,10 @@ import org.apache.flink.api.common.Plan;
import org.apache.flink.api.common.cache.DistributedCache.DistributedCacheEntry;
import org.apache.flink.api.common.io.FileInputFormat;
import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.operators.OperatorInformation;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormat;
import org.apache.flink.api.java.io.CollectionInputFormat;
import org.apache.flink.api.java.io.CsvReader;
@@ -54,16 +57,21 @@ import org.apache.flink.api.java.operators.Operator;
import org.apache.flink.api.java.operators.OperatorTranslation;
import org.apache.flink.api.java.operators.translation.JavaPlan;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.ValueTypeInfo;
+import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers;
import org.apache.flink.core.fs.Path;
import org.apache.flink.types.StringValue;
import org.apache.flink.util.NumberSequenceIterator;
import org.apache.flink.util.SplittableIterator;
+import org.apache.flink.util.Visitor;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* The ExecutionEnviroment is the context in which a program is executed. A
@@ -84,6 +92,8 @@ import org.apache.hadoop.mapreduce.Job;
* @see RemoteEnvironment
*/
public abstract class ExecutionEnvironment {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExecutionEnvironment.class);
/** The environment of the context (local by default, cluster if invoked through command line) */
private static ExecutionEnvironmentFactory contextEnvironmentFactory;
@@ -200,29 +210,51 @@ public abstract class ExecutionEnvironment {
// Registry for types and serializers
// --------------------------------------------------------------------------------------------
+
/**
- * Registers the given Serializer as a default serializer for the given type at the
- * {@link org.apache.flink.api.java.typeutils.runtime.KryoSerializer}.
- *
+ * Adds a new Kryo default serializer to the Runtime.
+ *
* Note that the serializer instance must be serializable (as defined by java.io.Serializable),
* because it may be distributed to the worker nodes by java serialization.
- *
+ *
* @param type The class of the types serialized with the given serializer.
* @param serializer The serializer to use.
*/
- public void registerKryoSerializer(Class<?> type, Serializer<?> serializer) {
- config.registerKryoSerializer(type, serializer);
+ public void addDefaultKryoSerializer(Class<?> type, Serializer<?> serializer) {
+ config.addDefaultKryoSerializer(type, serializer);
}
/**
- * Registers the given Serializer via its class as a serializer for the given type at the
- * {@link org.apache.flink.api.java.typeutils.runtime.KryoSerializer}.
- *
+ * Adds a new Kryo default serializer to the Runtime.
+ *
+ * @param type The class of the types serialized with the given serializer.
+ * @param serializerClass The class of the serializer to use.
+ */
+ public void addDefaultKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass) {
+ config.addDefaultKryoSerializer(type, serializerClass);
+ }
+
+ /**
+ * Registers the given type with a Kryo Serializer.
+ *
+ * Note that the serializer instance must be serializable (as defined by java.io.Serializable),
+ * because it may be distributed to the worker nodes by java serialization.
+ *
+ * @param type The class of the types serialized with the given serializer.
+ * @param serializer The serializer to use.
+ */
+ public void registerTypeWithKryoSerializer(Class<?> type, Serializer<?> serializer) {
+ config.registerTypeWithKryoSerializer(type, serializer);
+ }
+
+ /**
+ * Registers the given Serializer via its class as a serializer for the given type at the KryoSerializer
+ *
* @param type The class of the types serialized with the given serializer.
* @param serializerClass The class of the serializer to use.
*/
- public void registerKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass) {
- config.registerKryoSerializer(type, serializerClass);
+ public void registerTypeWithKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass) {
+ config.registerTypeWithKryoSerializer(type, serializerClass);
}
/**
@@ -743,7 +775,7 @@ public abstract class ExecutionEnvironment {
* @throws Exception Thrown, if the program executions fails.
*/
public abstract JobExecutionResult execute(String jobName) throws Exception;
-
+
/**
* Creates the plan with which the system will execute the program, and returns it as
* a String using a JSON representation of the execution data flow graph.
@@ -856,11 +888,35 @@ public abstract class ExecutionEnvironment {
OperatorTranslation translator = new OperatorTranslation();
JavaPlan plan = translator.translateToPlan(this.sinks, jobName);
-
+
if (getDegreeOfParallelism() > 0) {
plan.setDefaultParallelism(getDegreeOfParallelism());
}
plan.setExecutionConfig(getConfig());
+ // Check plan for GenericTypeInfo's and register the types at the serializers.
+ plan.accept(new Visitor<org.apache.flink.api.common.operators.Operator<?>>() {
+ @Override
+ public boolean preVisit(org.apache.flink.api.common.operators.Operator<?> visitable) {
+ OperatorInformation<?> opInfo = visitable.getOperatorInfo();
+ TypeInformation<?> typeInfo = opInfo.getOutputType();
+ if(typeInfo instanceof GenericTypeInfo) {
+ GenericTypeInfo<?> genericTypeInfo = (GenericTypeInfo<?>) typeInfo;
+ if(!config.isDisableAutoTypeRegistration()) {
+ Serializers.recursivelyRegisterType(genericTypeInfo.getTypeClass(), config);
+ }
+ }
+ if(typeInfo instanceof CompositeType) {
+ List<GenericTypeInfo<?>> genericTypesInComposite = new ArrayList<GenericTypeInfo<?>>();
+ Utils.getContainedGenericTypes((CompositeType)typeInfo, genericTypesInComposite);
+ for(GenericTypeInfo<?> gt : genericTypesInComposite) {
+ Serializers.recursivelyRegisterType(gt.getTypeClass(), config);
+ }
+ }
+ return true;
+ }
+ @Override
+ public void postVisit(org.apache.flink.api.common.operators.Operator<?> visitable) {}
+ });
try {
registerCachedFilesWithPlan(plan);
@@ -872,7 +928,25 @@ public abstract class ExecutionEnvironment {
if (clearSinks) {
this.sinks.clear();
}
-
+
+ // All types are registered now. Print information.
+ int registeredTypes = config.getRegisteredKryoTypes().size() +
+ config.getRegisteredPojoTypes().size() +
+ config.getRegisteredTypesWithKryoSerializerClasses().size() +
+ config.getRegisteredTypesWithKryoSerializers().size();
+ int defaultKryoSerializers = config.getDefaultKryoSerializers().size() +
+ config.getDefaultKryoSerializerClasses().size();
+ LOG.info("The job has {} registered types and {} default Kryo serializers", registeredTypes, defaultKryoSerializers);
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Registered Kryo types: {}", Joiner.on(',').join(config.getRegisteredKryoTypes()));
+ LOG.debug("Registered Kryo with Serializers types: {}", Joiner.on(',').join(config.getRegisteredTypesWithKryoSerializers()));
+ LOG.debug("Registered Kryo with Serializer Classes types: {}", Joiner.on(',').join(config.getRegisteredTypesWithKryoSerializerClasses()));
+ LOG.debug("Registered Kryo default Serializers: {}", Joiner.on(',').join(config.getDefaultKryoSerializers()));
+ LOG.debug("Registered Kryo default Serializers Classes {}", Joiner.on(',').join(config.getDefaultKryoSerializerClasses()));
+ LOG.debug("Registered POJO types: {}", Joiner.on(',').join(config.getRegisteredPojoTypes()));
+ }
+
return plan;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
index ba39aec..21df168 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
@@ -18,6 +18,12 @@
package org.apache.flink.api.java;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+
+import java.util.List;
+
public class Utils {
public static String getCallLocationName() {
@@ -35,4 +41,23 @@ public class Utils {
return String.format("%s(%s:%d)", elem.getMethodName(), elem.getFileName(), elem.getLineNumber());
}
+
+ /**
+ * Returns all GenericTypeInfos contained in a composite type.
+ *
+ * @param typeInfo
+ * @return
+ */
+ public static void getContainedGenericTypes(CompositeType typeInfo, List<GenericTypeInfo<?>> target) {
+ for(int i = 0; i < typeInfo.getArity(); i++) {
+ TypeInformation<?> type = typeInfo.getTypeAt(i);
+ if(type instanceof CompositeType) {
+ getContainedGenericTypes((CompositeType) type, target);
+ } else if(type instanceof GenericTypeInfo) {
+ if(!target.contains(type)) {
+ target.add((GenericTypeInfo<?>) type);
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/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
index ccdf7f7..bf580d1 100644
--- 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
@@ -27,10 +27,16 @@ import java.util.ArrayList;
import java.util.List;
/**
- * Special type information to generate a POJO type info from an avro schema.
+ * Special type information to generate a special AvroTypeInfo for Avro POJOs (implementing SpecificRecordBase, the typed Avro POJOs)
*
* Proceeding: It uses a regular pojo type analysis and replaces all GenericType<CharSequence>
- * with a GenericType<avro.Utf8>
+ * with a GenericType<avro.Utf8>.
+ * All other types used by Avro are standard Java types.
+ * Only strings are represented as CharSequence fields and represented as Utf8 classes at runtime.
+ * CharSequence is not comparable. To make them nicely usable with field expressions, we replace them here
+ * by generic type infos containing Utf8 classes (which are comparable),
+ *
+ * This class is checked by the AvroPojoTest.
* @param <T>
*/
public class AvroTypeInfo<T extends SpecificRecordBase> extends PojoTypeInfo<T> {
@@ -40,7 +46,7 @@ public class AvroTypeInfo<T extends SpecificRecordBase> extends PojoTypeInfo<T>
private static <T extends SpecificRecordBase> List<PojoField> generateFieldsFromAvroSchema(Class<T> typeClass) {
PojoTypeExtractor pte = new PojoTypeExtractor();
- TypeInformation ti = pte.analyzePojo(typeClass, new ArrayList<Type>(), null);
+ TypeInformation ti = pte.analyzePojo(typeClass, new ArrayList<Type>(), null, null, null);
if(!(ti instanceof PojoTypeInfo)) {
throw new IllegalStateException("Expecting type to be a PojoTypeInfo");
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
index c629ba3..34226ce 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
@@ -24,7 +24,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.typeutils.runtime.GenericTypeComparator;
-import org.apache.flink.api.java.typeutils.runtime.KryoSerializer;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType<T> {
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/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 73cedd1..5f52f98 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
@@ -1259,7 +1259,7 @@ public class TypeExtractor {
}
@SuppressWarnings("unchecked")
- private <OUT, IN1, IN2> TypeInformation<OUT> analyzePojo(Class<OUT> clazz, ArrayList<Type> typeHierarchy,
+ protected <OUT, IN1, IN2> TypeInformation<OUT> analyzePojo(Class<OUT> clazz, ArrayList<Type> typeHierarchy,
ParameterizedType parameterizedType, TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
// add the hierarchy of the POJO itself if it is generic
@@ -1385,8 +1385,9 @@ public class TypeExtractor {
}
return result;
}
-
- private static Class<?> typeToClass(Type t) {
+
+ // not public to users
+ public static Class<?> typeToClass(Type t) {
if (t instanceof Class) {
return (Class<?>)t;
}
@@ -1395,8 +1396,9 @@ public class TypeExtractor {
}
throw new IllegalArgumentException("Cannot convert type to class");
}
-
- private static boolean isClassType(Type t) {
+
+ // not public to users
+ public static boolean isClassType(Type t) {
return t instanceof Class<?> || t instanceof ParameterizedType;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/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
deleted file mode 100644
index d8411a0..0000000
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/KryoSerializer.java
+++ /dev/null
@@ -1,291 +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.
- */
-
-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.JavaSerializer;
-import com.twitter.chill.ScalaKryoInstantiator;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.EOFException;
-import java.io.IOException;
-import java.lang.reflect.Modifier;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * 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 = 3L;
-
- // ------------------------------------------------------------------------
-
- private final Map<Class<?>, Serializer<?>> registeredSerializers;
- private final Map<Class<?>, Class<? extends Serializer<?>>> registeredSerializersClasses;
- private final Set<Class<?>> registeredTypes;
-
- private final Class<T> type;
-
- // ------------------------------------------------------------------------
- // The fields below are lazily initialized after duplication or deserialization.
-
- private transient Kryo kryo;
- private transient T copyInstance;
-
- private transient DataOutputView previousOut;
- private transient DataInputView previousIn;
-
- private transient Input input;
- private transient Output output;
-
- // ------------------------------------------------------------------------
-
- public KryoSerializer(Class<T> type, ExecutionConfig executionConfig){
- if(type == null){
- throw new NullPointerException("Type class cannot be null.");
- }
- this.type = type;
-
- this.registeredSerializers = executionConfig.getRegisteredKryoSerializers();
- this.registeredSerializersClasses = executionConfig.getRegisteredKryoSerializersClasses();
- this.registeredTypes = executionConfig.getRegisteredKryoTypes();
- }
-
- /**
- * Copy-constructor that does not copy transient fields. They will be initialized once required.
- */
- protected KryoSerializer(KryoSerializer<T> toCopy) {
- registeredSerializers = toCopy.registeredSerializers;
- registeredSerializersClasses = toCopy.registeredSerializersClasses;
- registeredTypes = toCopy.registeredTypes;
-
- type = toCopy.type;
- if(type == null){
- throw new NullPointerException("Type class cannot be null.");
- }
- }
-
- // ------------------------------------------------------------------------
-
- @Override
- public boolean isImmutableType() {
- return false;
- }
-
- @Override
- public KryoSerializer<T> duplicate() {
- return new KryoSerializer<T>(this);
- }
-
- @Override
- public T createInstance() {
- if(Modifier.isAbstract(type.getModifiers()) || Modifier.isInterface(type.getModifiers()) ) {
- return null;
- } else {
- checkKryoInitialized();
- try {
- return kryo.newInstance(type);
- } catch(Throwable e) {
- return null;
- }
- }
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public T copy(T from) {
- if (from == null) {
- return null;
- }
- checkKryoInitialized();
- try {
- return kryo.copy(from);
- }
- catch(KryoException ke) {
- // kryo was unable to copy it, so we do it through serialization:
- ByteArrayOutputStream baout = new ByteArrayOutputStream();
- Output output = new Output(baout);
-
- kryo.writeObject(output, from);
-
- output.close();
-
- ByteArrayInputStream bain = new ByteArrayInputStream(baout.toByteArray());
- Input input = new Input(bain);
-
- return (T)kryo.readObject(input, from.getClass());
- }
- }
-
- @Override
- public T copy(T from, T reuse) {
- return copy(from);
- }
-
- @Override
- public int getLength() {
- return -1;
- }
-
- @Override
- public void serialize(T record, DataOutputView target) throws IOException {
- checkKryoInitialized();
- if (target != previousOut) {
- DataOutputViewStream outputStream = new DataOutputViewStream(target);
- output = new Output(outputStream);
- previousOut = target;
- }
-
- try {
- kryo.writeClassAndObject(output, record);
- output.flush();
- }
- catch (KryoException ke) {
- Throwable cause = ke.getCause();
- if (cause instanceof EOFException) {
- throw (EOFException) cause;
- }
- else {
- throw ke;
- }
- }
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public T deserialize(DataInputView source) throws IOException {
- checkKryoInitialized();
- if (source != previousIn) {
- DataInputViewStream inputStream = new DataInputViewStream(source);
- input = new NoFetchingInput(inputStream);
- previousIn = source;
- }
-
- try {
- return (T) kryo.readClassAndObject(input);
- } catch (KryoException ke) {
- Throwable cause = ke.getCause();
-
- if(cause instanceof EOFException) {
- throw (EOFException) cause;
- } else {
- throw ke;
- }
- }
- }
-
- @Override
- public T deserialize(T reuse, DataInputView source) throws IOException {
- return deserialize(source);
- }
-
- @Override
- public void copy(DataInputView source, DataOutputView target) throws IOException {
- checkKryoInitialized();
- if(this.copyInstance == null){
- this.copyInstance = createInstance();
- }
-
- T tmp = deserialize(copyInstance, source);
- serialize(tmp, target);
- }
-
- // --------------------------------------------------------------------------------------------
-
- @Override
- public int hashCode() {
- return type.hashCode();
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj != null && obj instanceof KryoSerializer) {
- KryoSerializer<?> other = (KryoSerializer<?>) obj;
- return other.type == this.type;
- } else {
- return false;
- }
- }
-
- // --------------------------------------------------------------------------------------------
-
- private void checkKryoInitialized() {
- if (this.kryo == null) {
- this.kryo = new ScalaKryoInstantiator().newKryo();
-
- // Throwable and all subclasses should be serialized via java serialization
- kryo.addDefaultSerializer(Throwable.class, new JavaSerializer());
-
- // 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);
- }
-
- // register given serializer classes
- for (Map.Entry<Class<?>, Class<? extends Serializer<?>>> e : registeredSerializersClasses.entrySet()) {
- Class<?> typeClass = e.getKey();
- Class<? extends Serializer<?>> serializerClass = e.getValue();
-
- Serializer<?> serializer =
- ReflectionSerializerFactory.makeSerializer(kryo, serializerClass, typeClass);
- kryo.register(typeClass, serializer);
- }
-
- // register given serializers
- for (Map.Entry<Class<?>, Serializer<?>> e : registeredSerializers.entrySet()) {
- kryo.register(e.getKey(), e.getValue());
- }
-
- kryo.setRegistrationRequired(false);
- kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
- }
- }
-
- // --------------------------------------------------------------------------------------------
- // For testing
- // --------------------------------------------------------------------------------------------
-
- Kryo getKryo() {
- checkKryoInitialized();
- return this.kryo;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
index f7742ca..8b95296 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
@@ -29,7 +29,6 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -79,7 +78,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
this.numFields = fieldSerializers.length;
this.executionConfig = executionConfig;
- Set<Class<?>> registeredPojoTypes = executionConfig.getRegisteredPojoTypes();
+ List<Class<?>> registeredPojoTypes = executionConfig.getRegisteredPojoTypes();
for (int i = 0; i < numFields; i++) {
this.fields[i].setAccessible(true);
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
new file mode 100644
index 0000000..e14546e
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java
@@ -0,0 +1,312 @@
+/*
+ * 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.runtime.kryo;
+
+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.JavaSerializer;
+import com.twitter.chill.ScalaKryoInstantiator;
+
+import org.apache.avro.generic.GenericData;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.NoFetchingInput;
+import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers.SpecificInstanceCollectionSerializerForArrayList;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.lang.reflect.Modifier;
+import java.util.List;
+
+/**
+ * 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 = 3L;
+
+ // ------------------------------------------------------------------------
+
+ private final List<ExecutionConfig.Entry<Class<?>, Serializer<?>>> registeredTypesWithSerializers;
+ private final List<ExecutionConfig.Entry<Class<?>, Class<? extends Serializer<?>>>> registeredTypesWithSerializerClasses;
+ private final List<ExecutionConfig.Entry<Class<?>, Serializer<?>>> defaultSerializers;
+ private final List<ExecutionConfig.Entry<Class<?>, Class<? extends Serializer<?>>>> defaultSerializerClasses;
+ private final List<Class<?>> registeredTypes;
+
+ private final Class<T> type;
+
+ // ------------------------------------------------------------------------
+ // The fields below are lazily initialized after duplication or deserialization.
+
+ private transient Kryo kryo;
+ private transient T copyInstance;
+
+ private transient DataOutputView previousOut;
+ private transient DataInputView previousIn;
+
+ private transient Input input;
+ private transient Output output;
+
+ // ------------------------------------------------------------------------
+
+ public KryoSerializer(Class<T> type, ExecutionConfig executionConfig){
+ if(type == null){
+ throw new NullPointerException("Type class cannot be null.");
+ }
+ this.type = type;
+
+ this.defaultSerializers = executionConfig.getDefaultKryoSerializers();
+ this.defaultSerializerClasses = executionConfig.getDefaultKryoSerializerClasses();
+ this.registeredTypesWithSerializers = executionConfig.getRegisteredTypesWithKryoSerializers();
+ this.registeredTypesWithSerializerClasses = executionConfig.getRegisteredTypesWithKryoSerializerClasses();
+ this.registeredTypes = executionConfig.getRegisteredKryoTypes();
+ }
+
+ /**
+ * Copy-constructor that does not copy transient fields. They will be initialized once required.
+ */
+ protected KryoSerializer(KryoSerializer<T> toCopy) {
+ registeredTypesWithSerializers = toCopy.registeredTypesWithSerializers;
+ registeredTypesWithSerializerClasses = toCopy.registeredTypesWithSerializerClasses;
+ defaultSerializers = toCopy.defaultSerializers;
+ defaultSerializerClasses = toCopy.defaultSerializerClasses;
+ registeredTypes = toCopy.registeredTypes;
+
+ type = toCopy.type;
+ if(type == null){
+ throw new NullPointerException("Type class cannot be null.");
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public KryoSerializer<T> duplicate() {
+ return new KryoSerializer<T>(this);
+ }
+
+ @Override
+ public T createInstance() {
+ if(Modifier.isAbstract(type.getModifiers()) || Modifier.isInterface(type.getModifiers()) ) {
+ return null;
+ } else {
+ checkKryoInitialized();
+ try {
+ return kryo.newInstance(type);
+ } catch(Throwable e) {
+ return null;
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public T copy(T from) {
+ if (from == null) {
+ return null;
+ }
+ checkKryoInitialized();
+ try {
+ return kryo.copy(from);
+ }
+ catch(KryoException ke) {
+ // kryo was unable to copy it, so we do it through serialization:
+ ByteArrayOutputStream baout = new ByteArrayOutputStream();
+ Output output = new Output(baout);
+
+ kryo.writeObject(output, from);
+
+ output.close();
+
+ ByteArrayInputStream bain = new ByteArrayInputStream(baout.toByteArray());
+ Input input = new Input(bain);
+
+ return (T)kryo.readObject(input, from.getClass());
+ }
+ }
+
+ @Override
+ public T copy(T from, T reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(T record, DataOutputView target) throws IOException {
+ checkKryoInitialized();
+ if (target != previousOut) {
+ DataOutputViewStream outputStream = new DataOutputViewStream(target);
+ output = new Output(outputStream);
+ previousOut = target;
+ }
+
+ try {
+ kryo.writeClassAndObject(output, record);
+ output.flush();
+ }
+ catch (KryoException ke) {
+ Throwable cause = ke.getCause();
+ if (cause instanceof EOFException) {
+ throw (EOFException) cause;
+ }
+ else {
+ throw ke;
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public T deserialize(DataInputView source) throws IOException {
+ checkKryoInitialized();
+ if (source != previousIn) {
+ DataInputViewStream inputStream = new DataInputViewStream(source);
+ input = new NoFetchingInput(inputStream);
+ previousIn = source;
+ }
+
+ try {
+ return (T) kryo.readClassAndObject(input);
+ } catch (KryoException ke) {
+ Throwable cause = ke.getCause();
+
+ if(cause instanceof EOFException) {
+ throw (EOFException) cause;
+ } else {
+ throw ke;
+ }
+ }
+ }
+
+ @Override
+ public T deserialize(T reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ checkKryoInitialized();
+ if(this.copyInstance == null){
+ this.copyInstance = createInstance();
+ }
+
+ T tmp = deserialize(copyInstance, source);
+ serialize(tmp, target);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public int hashCode() {
+ return type.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj != null && obj instanceof KryoSerializer) {
+ KryoSerializer<?> other = (KryoSerializer<?>) obj;
+ return other.type == this.type;
+ } else {
+ return false;
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private void checkKryoInitialized() {
+ if (this.kryo == null) {
+ this.kryo = new ScalaKryoInstantiator().newKryo();
+
+ // Throwable and all subclasses should be serialized via java serialization
+ kryo.addDefaultSerializer(Throwable.class, new JavaSerializer());
+
+ // Add default serializers first, so that they type registrations without a serializer
+ // are registered with a default serializer
+ for(ExecutionConfig.Entry<Class<?>, Serializer<?>> serializer : defaultSerializers) {
+ kryo.addDefaultSerializer(serializer.getKey(), serializer.getValue());
+ }
+ for(ExecutionConfig.Entry<Class<?>, Class<? extends Serializer<?>>> serializer : defaultSerializerClasses) {
+ kryo.addDefaultSerializer(serializer.getKey(), serializer.getValue());
+ }
+
+ // 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);
+ }
+
+ // register given serializer classes
+ for (ExecutionConfig.Entry<Class<?>, Class<? extends Serializer<?>>> e : registeredTypesWithSerializerClasses) {
+ Class<?> typeClass = e.getKey();
+ Class<? extends Serializer<?>> serializerClass = e.getValue();
+
+ Serializer<?> serializer =
+ ReflectionSerializerFactory.makeSerializer(kryo, serializerClass, typeClass);
+ kryo.register(typeClass, serializer);
+ }
+
+ // register given serializers
+ for (ExecutionConfig.Entry<Class<?>, Serializer<?>> e : registeredTypesWithSerializers) {
+ kryo.register(e.getKey(), e.getValue());
+ }
+ // this is needed for Avro but can not be added on demand.
+ kryo.register(GenericData.Array.class, new SpecificInstanceCollectionSerializerForArrayList());
+
+ kryo.setRegistrationRequired(false);
+ kryo.setClassLoader(Thread.currentThread().getContextClassLoader());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // For testing
+ // --------------------------------------------------------------------------------------------
+
+ Kryo getKryo() {
+ checkKryoInitialized();
+ return this.kryo;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
new file mode 100644
index 0000000..3e355aa
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
@@ -0,0 +1,242 @@
+/*
+ * 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.runtime.kryo;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.serializers.CollectionSerializer;
+import com.google.protobuf.Message;
+import com.twitter.chill.protobuf.ProtobufSerializer;
+import com.twitter.chill.thrift.TBaseSerializer;
+import de.javakaffee.kryoserializers.jodatime.JodaDateTimeSerializer;
+import de.javakaffee.kryoserializers.jodatime.JodaIntervalSerializer;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.thrift.protocol.TMessage;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import scala.reflect.ClassTag;
+
+import java.io.Serializable;
+import java.lang.reflect.Field;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+
+/**
+ * Class containing utilities for the serializers of the Flink Runtime.
+ *
+ * Most of the serializers are automatically added to the system.
+ *
+ * Note that users can also implement the {@link com.esotericsoftware.kryo.KryoSerializable} interface
+ * to provide custom serialization for their classes.
+ * Also, there is a Java Annotation for adding a default serializer (@DefaultSerializer) to classes.
+ */
+public class Serializers {
+ /**
+ * NOTE: This method is not a public Flink API.
+ *
+ * This method walks the entire hierarchy of the given type and registers all types it encounters
+ * to Kryo.
+ * It also watches for types which need special serializers.
+ */
+ private static Set<Class<?>> alreadySeen = new HashSet<Class<?>>();
+
+ public static void recursivelyRegisterType(Class<?> type, ExecutionConfig config) {
+ alreadySeen.add(type);
+ if(type.isPrimitive()) {
+ return;
+ }
+ config.registerKryoType(type);
+ addSerializerForType(config, type);
+
+ Field[] fields = type.getDeclaredFields();
+ for(Field field : fields) {
+ Type fieldType = field.getGenericType();
+ if(fieldType instanceof ParameterizedType) { // field has generics
+ ParameterizedType parameterizedFieldType = (ParameterizedType) fieldType;
+ for(Type t: parameterizedFieldType.getActualTypeArguments()) {
+ if(TypeExtractor.isClassType(t) ) {
+ Class clazz = TypeExtractor.typeToClass(t);
+ if(!alreadySeen.contains(clazz)) {
+ recursivelyRegisterType(TypeExtractor.typeToClass(t), config);
+ }
+ }
+ }
+ }
+ Class<?> clazz = field.getType();
+ if(!alreadySeen.contains(clazz)) {
+ recursivelyRegisterType(clazz, config);
+ }
+ }
+ }
+
+ public static void addSerializerForType(ExecutionConfig reg, Class<?> type) {
+ if(Message.class.isAssignableFrom(type)) {
+ registerProtoBuf(reg);
+ }
+ if(TMessage.class.isAssignableFrom(type)) {
+ registerThrift(reg);
+ }
+ if(GenericData.Record.class.isAssignableFrom(type)) {
+ registerGenericAvro(reg);
+ }
+ if(SpecificRecordBase.class.isAssignableFrom(type)) {
+ registerSpecificAvro(reg, (Class<? extends SpecificRecordBase>) type);
+ }
+ if(DateTime.class.isAssignableFrom(type) || Interval.class.isAssignableFrom(type)) {
+ registerJodaTime(reg);
+ }
+ }
+
+ /**
+ * Register serializers required for Google Protocol Buffers
+ * with Flink runtime.
+ */
+ public static void registerProtoBuf(ExecutionConfig reg) {
+ // Google Protobuf (FLINK-1392)
+ reg.registerTypeWithKryoSerializer(Message.class, ProtobufSerializer.class);
+ }
+
+ /**
+ * Register Apache Thrift messages
+ */
+ public static void registerThrift(ExecutionConfig reg) {
+ // TBaseSerializer states it should be initalized as a default Kryo serializer
+ reg.addDefaultKryoSerializer(TMessage.class, TBaseSerializer.class);
+ reg.registerKryoType(TMessage.class);
+ }
+
+ /**
+ * Register these serializers for using Avro's {@see GenericData.Record} and classes
+ * implementing {@see org.apache.avro.specific.SpecificRecordBase}
+ */
+ public static void registerGenericAvro(ExecutionConfig reg) {
+ // 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
+ reg.registerTypeWithKryoSerializer(GenericData.Array.class, SpecificInstanceCollectionSerializerForArrayList.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.
+ reg.addDefaultKryoSerializer(Schema.class, AvroSchemaSerializer.class);
+ }
+
+
+ public static void registerSpecificAvro(ExecutionConfig reg, Class<? extends SpecificRecordBase> avroType) {
+ registerGenericAvro(reg);
+ // 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.
+ // (However only if the GenericData.Array type is registered!)
+
+ ClassTag<SpecificRecordBase> tag = scala.reflect.ClassTag$.MODULE$.apply(avroType);
+ reg.registerTypeWithKryoSerializer(avroType, com.twitter.chill.avro.AvroSerializer.SpecificRecordSerializer(tag));
+ }
+
+
+ /**
+ * Currently, the following classes of JodaTime are supported:
+ * - DateTime
+ * - Interval
+ *
+ * The following chronologies are supported: (see {@link de.javakaffee.kryoserializers.jodatime.JodaDateTimeSerializer})
+ * <ul>
+ * <li>{@link org.joda.time.chrono.ISOChronology}</li>
+ * <li>{@link org.joda.time.chrono.CopticChronology}</li>
+ * <li>{@link org.joda.time.chrono.EthiopicChronology}</li>
+ * <li>{@link org.joda.time.chrono.GregorianChronology}</li>
+ * <li>{@link org.joda.time.chrono.JulianChronology}</li>
+ * <li>{@link org.joda.time.chrono.IslamicChronology}</li>
+ * <li>{@link org.joda.time.chrono.BuddhistChronology}</li>
+ * <li>{@link org.joda.time.chrono.GJChronology}</li>
+ * </ul>
+ */
+ public static void registerJodaTime(ExecutionConfig reg) {
+ reg.registerTypeWithKryoSerializer(DateTime.class, JodaDateTimeSerializer.class);
+ reg.registerTypeWithKryoSerializer(Interval.class, JodaIntervalSerializer.class);
+ }
+
+ /**
+ * Register less frequently used serializers
+ */
+ public static void registerJavaUtils(ExecutionConfig reg) {
+ // BitSet, Regex is already present through twitter-chill.
+ }
+
+
+ // --------------------------------------------------------------------------------------------
+ // Custom Serializers
+ // --------------------------------------------------------------------------------------------
+
+ public static class SpecificInstanceCollectionSerializerForArrayList extends SpecificInstanceCollectionSerializer<ArrayList> {
+ public SpecificInstanceCollectionSerializerForArrayList() {
+ super(ArrayList.class);
+ }
+ }
+ /**
+ * 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 implements Serializable {
+ private 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> implements Serializable {
+ @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);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/AbstractGenericTypeSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/AbstractGenericTypeSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/AbstractGenericTypeSerializerTest.java
index 2a11bd7..059c78d 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/AbstractGenericTypeSerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/AbstractGenericTypeSerializerTest.java
@@ -21,13 +21,10 @@ package org.apache.flink.api.java.typeutils.runtime;
import org.apache.flink.api.common.typeutils.SerializerTestInstance;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.util.StringUtils;
-import org.joda.time.DateTime;
import org.junit.Test;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java
deleted file mode 100644
index d1163d5..0000000
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java
+++ /dev/null
@@ -1,29 +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.
- */
-
-package org.apache.flink.api.java.typeutils.runtime;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-
-public class KryoGenericArraySerializerTest extends AbstractGenericArraySerializerTest {
- @Override
- protected <T> TypeSerializer<T> createComponentSerializer(Class<T> type) {
- return new KryoSerializer<T>(type, new ExecutionConfig());
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java
deleted file mode 100644
index 01c76d9..0000000
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java
+++ /dev/null
@@ -1,29 +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.
- */
-
-package org.apache.flink.api.java.typeutils.runtime;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-
-public class KryoGenericTypeComparatorTest extends AbstractGenericTypeComparatorTest {
- @Override
- protected <T> TypeSerializer<T> createSerializer(Class<T> type) {
- return new KryoSerializer<T>(type, new ExecutionConfig());
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/354efec0/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java
deleted file mode 100644
index 2f47e51..0000000
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java
+++ /dev/null
@@ -1,165 +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.
- */
-
-package org.apache.flink.api.java.typeutils.runtime;
-
-import static org.junit.Assert.*;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.junit.Test;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.Random;
-
-@SuppressWarnings("unchecked")
-public class KryoGenericTypeSerializerTest extends AbstractGenericTypeSerializerTest {
-
- @Test
- public void testJavaList(){
- Collection<Integer> a = new ArrayList<Integer>();
-
- fillCollection(a);
-
- runTests(a);
- }
-
- @Test
- public void testJavaSet(){
- Collection<Integer> b = new HashSet<Integer>();
-
- fillCollection(b);
-
- runTests(b);
- }
-
-
-
- @Test
- public void testJavaDequeue(){
- Collection<Integer> c = new LinkedList<Integer>();
-
- fillCollection(c);
-
- runTests(c);
- }
-
- @Test
- public void testJodaTime(){
- Collection<DateTime> b = new HashSet<DateTime>();
-
- b.add(new DateTime(1));
- b.add(new DateTime(2));
-
- runTests(b);
- }
-
- private void fillCollection(Collection<Integer> coll){
- coll.add(42);
- coll.add(1337);
- coll.add(49);
- coll.add(1);
- }
-
- @Override
- protected <T> TypeSerializer<T> createSerializer(Class<T> type) {
- return new KryoSerializer<T>(type, new ExecutionConfig());
- }
-
- /**
- * Make sure that the kryo serializer forwards EOF exceptions properly when serializing
- */
- @Test
- public void testForwardEOFExceptionWhileSerializing() {
- try {
- // construct a long string
- String str;
- {
- char[] charData = new char[40000];
- Random rnd = new Random();
-
- for (int i = 0; i < charData.length; i++) {
- charData[i] = (char) rnd.nextInt(10000);
- }
-
- str = new String(charData);
- }
-
- // construct a memory target that is too small for the string
- TestDataOutputSerializer target = new TestDataOutputSerializer(10000, 30000);
- KryoSerializer<String> serializer = new KryoSerializer<String>(String.class, new ExecutionConfig());
-
- try {
- serializer.serialize(str, target);
- fail("should throw a java.io.EOFException");
- }
- catch (java.io.EOFException e) {
- // that is how we like it
- }
- catch (Exception e) {
- fail("throws wrong exception: should throw a java.io.EOFException, has thrown a " + e.getClass().getName());
- }
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
-
- /**
- * Make sure that the kryo serializer forwards EOF exceptions properly when serializing
- */
- @Test
- public void testForwardEOFExceptionWhileDeserializing() {
- try {
- int numElements = 100;
- // construct a memory target that is too small for the string
- TestDataOutputSerializer target = new TestDataOutputSerializer(5*numElements, 5*numElements);
- KryoSerializer<Integer> serializer = new KryoSerializer<Integer>(Integer.class, new ExecutionConfig());
-
- for(int i = 0; i < numElements; i++){
- serializer.serialize(i, target);
- }
-
- TestInputView source = new TestInputView(target.copyByteBuffer());
-
- for(int i = 0; i < numElements; i++){
- int value = serializer.deserialize(source);
- assertEquals(i, value);
- }
-
- try {
- serializer.deserialize(source);
- fail("should throw a java.io.EOFException");
- }
- catch (java.io.EOFException e) {
- // that is how we like it :-)
- }
- catch (Exception e) {
- fail("throws wrong exception: should throw a java.io.EOFException, has thrown a " + e.getClass().getName());
- }
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
- }
-}
\ No newline at end of file