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/05/12 10:34:35 UTC
flink git commit: [FLINK-1676] Rework
ExecutionConfig.enableForceKryo()
Repository: flink
Updated Branches:
refs/heads/master c1dba084a -> 4cc971da3
[FLINK-1676] Rework ExecutionConfig.enableForceKryo()
This closes #473
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4cc971da
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4cc971da
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4cc971da
Branch: refs/heads/master
Commit: 4cc971da320ffd0e90bd88b3402f24caf50f50d0
Parents: c1dba08
Author: Robert Metzger <rm...@apache.org>
Authored: Tue Mar 10 17:22:46 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Tue May 12 09:27:47 2015 +0200
----------------------------------------------------------------------
docs/internals/types_serialization.md | 27 ++++++++
.../flink/api/common/ExecutionConfig.java | 17 +++--
.../java/org/apache/flink/api/java/DataSet.java | 8 +--
.../flink/api/java/ExecutionEnvironment.java | 10 +++
.../api/java/typeutils/GenericTypeInfo.java | 7 --
.../flink/api/java/typeutils/PojoTypeInfo.java | 13 ++++
.../apache/flink/api/io/avro/AvroPojoTest.java | 69 ++++++++++++++++++--
7 files changed, 125 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4cc971da/docs/internals/types_serialization.md
----------------------------------------------------------------------
diff --git a/docs/internals/types_serialization.md b/docs/internals/types_serialization.md
index 187364b..19260f1 100644
--- a/docs/internals/types_serialization.md
+++ b/docs/internals/types_serialization.md
@@ -220,8 +220,35 @@ The Eclipse JDT compiler has added support for this as of version 4.5 M4. Discus
OpenJDK compiler is pending.
+#### Serialization of POJO types
+The PojoTypeInformation is creating serializers for all the fields inside the POJO. Standard types such as
+int, long, String etc. are handled by serializers we ship with Flink.
+For all other types, we fall back to Kryo.
+If Kryo is not able to handle the type, you can ask the PojoTypeInfo to serialize the POJO using Avro.
+To do so, you have to call
+
+{% highlight java %}
+final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+env.getConfig().enableForceAvro();
+{% endhighlight %}
+
+Note that Flink is automatically serializing POJOs generated by Avro with the Avro serializer.
+
+If you want your **entire** POJO Type to be treated by the Kryo serializer, set
+
+{% highlight java %}
+final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+env.getConfig().enableForceKryo();
+{% endhighlight %}
+
+If Kryo is not able to serialize your POJO, you can add a custom serializer to Kryo, using
+{% highlight java %}
+env.getConfig().addDefaultKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass)
+{% endhighlight %}
+
+There are different variants of these methods available.
http://git-wip-us.apache.org/repos/asf/flink/blob/4cc971da/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 3809329..8df6aab 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
@@ -74,7 +74,7 @@ public class ExecutionConfig implements Serializable {
private boolean disableAutoTypeRegistration = false;
- private boolean serializeGenericTypesWithAvro = false;
+ private boolean forceAvro = false;
/** If set to true, progress updates are printed to System.out during execution */
private boolean printProgressDuringExecution = true;
@@ -272,16 +272,19 @@ public class ExecutionConfig implements Serializable {
return forceKryo;
}
- public void enableGenericTypeSerializationWithAvro() {
- serializeGenericTypesWithAvro = true;
+ /**
+ * Force Flink to use the AvroSerializer for POJOs.
+ */
+ public void enableForceAvro() {
+ forceAvro = true;
}
- public void disableGenericTypeSerializationWithAvro() {
- serializeGenericTypesWithAvro = true;
+ public void disableForceAvro() {
+ forceAvro = false;
}
- public boolean serializeGenericTypesWithAvro() {
- return serializeGenericTypesWithAvro;
+ public boolean isForceAvroEnabled() {
+ return forceAvro;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/4cc971da/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
index 5d1ca4c..e99d4cf 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
@@ -80,10 +80,8 @@ import org.apache.flink.api.java.operators.UnionOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.typeutils.InputTypeConfigurable;
import org.apache.flink.api.java.typeutils.MissingTypeInfo;
-import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.core.fs.FileSystem.WriteMode;
@@ -123,11 +121,7 @@ public abstract class DataSet<T> {
}
this.context = context;
- if (typeInfo instanceof PojoTypeInfo && context.getConfig().isForceKryoEnabled()) {
- this.type = new GenericTypeInfo<T>(typeInfo.getTypeClass());
- } else {
- this.type = typeInfo;
- }
+ this.type = typeInfo;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/4cc971da/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 5e71f44..7e4604e 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
@@ -974,6 +974,16 @@ public abstract class ExecutionEnvironment {
config.getDefaultKryoSerializerClasses().size();
LOG.info("The job has {} registered types and {} default Kryo serializers", registeredTypes, defaultKryoSerializers);
+ if(config.isForceKryoEnabled() && config.isForceAvroEnabled()) {
+ LOG.warn("In the ExecutionConfig, both Avro and Kryo are enforced. Using Kryo serializer");
+ }
+ if(config.isForceKryoEnabled()) {
+ LOG.info("Using KryoSerializer for serializing POJOs");
+ }
+ if(config.isForceAvroEnabled()) {
+ LOG.info("Using AvroSerializer for serializing POJOs");
+ }
+
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()));
http://git-wip-us.apache.org/repos/asf/flink/blob/4cc971da/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 e25a5d9..5caf8f2 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
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.typeinfo.AtomicType;
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.AvroSerializer;
import org.apache.flink.api.java.typeutils.runtime.GenericTypeComparator;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
@@ -70,9 +69,6 @@ public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType
@Override
public TypeSerializer<T> createSerializer(ExecutionConfig config) {
- if(config.serializeGenericTypesWithAvro()) {
- return new AvroSerializer<T>(this.typeClass);
- }
return new KryoSerializer<T>(this.typeClass, config);
}
@@ -108,7 +104,4 @@ public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType
public String toString() {
return "GenericType<" + typeClass.getCanonicalName() + ">";
}
-
-
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4cc971da/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
index 2f3db7c..273a907 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
@@ -33,11 +33,15 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.runtime.AvroSerializer;
import org.apache.flink.api.java.typeutils.runtime.PojoComparator;
import org.apache.flink.api.java.typeutils.runtime.PojoSerializer;
import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
import com.google.common.base.Joiner;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* TypeInformation for "Java Beans"-style types. Flink refers to them as POJOs,
@@ -55,6 +59,8 @@ public class PojoTypeInfo<T> extends CompositeType<T> {
private static final long serialVersionUID = 1L;
+ private static final Logger LOG = LoggerFactory.getLogger(PojoTypeInfo.class);
+
private final static String REGEX_FIELD = "[\\p{L}_\\$][\\p{L}\\p{Digit}_\\$]*";
private final static String REGEX_NESTED_FIELDS = "("+REGEX_FIELD+")(\\.(.+))?";
private final static String REGEX_NESTED_FIELDS_WILDCARD = REGEX_NESTED_FIELDS
@@ -313,6 +319,13 @@ public class PojoTypeInfo<T> extends CompositeType<T> {
@Override
public TypeSerializer<T> createSerializer(ExecutionConfig config) {
+ if(config.isForceKryoEnabled()) {
+ return new KryoSerializer<T>(this.typeClass, config);
+ }
+ if(config.isForceAvroEnabled()) {
+ return new AvroSerializer<T>(this.typeClass);
+ }
+
TypeSerializer<?>[] fieldSerializers = new TypeSerializer<?>[fields.length ];
Field[] reflectiveFields = new Field[fields.length];
http://git-wip-us.apache.org/repos/asf/flink/blob/4cc971da/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
index bddb79c..7cebccb 100644
--- a/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
+++ b/flink-staging/flink-avro/src/test/java/org/apache/flink/api/io/avro/AvroPojoTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.io.AvroInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.PojoSerializer;
import org.apache.flink.core.fs.Path;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.util.Collector;
@@ -40,6 +41,8 @@ import org.junit.runners.Parameterized;
import java.io.File;
import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
@RunWith(Parameterized.class)
public class AvroPojoTest extends MultipleProgramsTestBase {
@@ -74,7 +77,7 @@ public class AvroPojoTest extends MultipleProgramsTestBase {
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>() {
+ .map(new MapFunction<User, User>() {
@Override
public User map(User value) throws Exception {
value.setTypeMap(null);
@@ -92,6 +95,34 @@ public class AvroPojoTest extends MultipleProgramsTestBase {
}
@Test
+ public void testSerializeWithAvro() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().enableForceAvro();
+ 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 {
+ Map<CharSequence, Long> ab = new HashMap<CharSequence, Long>(1);
+ ab.put("hehe", 12L);
+ value.setTypeMap(ab);
+ 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\": {\"hehe\": 12}, \"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\": {\"hehe\": 12}, \"type_fixed\": null, \"type_union\": null}\n";
+ }
+
+ @Test
public void testKeySelection() throws Exception {
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.getConfig().enableObjectReuse();
@@ -103,7 +134,7 @@ public class AvroPojoTest extends MultipleProgramsTestBase {
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) {
+ for (User u : values) {
out.collect(new Tuple2<String, Integer>(u.getName().toString(), 1));
}
}
@@ -115,12 +146,10 @@ public class AvroPojoTest extends MultipleProgramsTestBase {
expected = "(Alyssa,1)\n(Charlie,1)\n";
}
-
@Test
public void testWithAvroGenericSer() throws Exception {
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- env.getConfig().enableGenericTypeSerializationWithAvro();
- env.getConfig().enableForceKryo();
+ env.getConfig().enableForceAvro();
Path in = new Path(inFile.getAbsoluteFile().toURI());
AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
@@ -147,6 +176,36 @@ public class AvroPojoTest extends MultipleProgramsTestBase {
expected = "(Charlie,1)\n(Alyssa,1)\n";
}
+ @Test
+ public void testWithKryoGenericSer() throws Exception {
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().enableForceKryo();
+ 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(new KeySelector<User, String>() {
+ @Override
+ public String getKey(User value) throws Exception {
+ return String.valueOf(value.getName());
+ }
+ }).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 = "(Charlie,1)\n(Alyssa,1)\n";
+ }
+
/**
* Test some know fields for grouping on
*/