You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/09/21 15:54:14 UTC
[1/6] flink git commit: [FLINK-4645] [core] Adjust signatures of
'registerTypeWithKryoSerializer(...)' methods to allow simpler passing of
classes without generic casting.
Repository: flink
Updated Branches:
refs/heads/master 721220203 -> b51605686
[FLINK-4645] [core] Adjust signatures of 'registerTypeWithKryoSerializer(...)' methods to allow simpler passing of
classes without generic casting.
This is not API breaking due to generic type erasure in Java. The changes method still has the same
binary signature.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/82ef021c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/82ef021c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/82ef021c
Branch: refs/heads/master
Commit: 82ef021cb8f77635bcec61d49eedce1ddefd1e48
Parents: 349d06d
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Sep 20 19:47:49 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Sep 21 17:53:32 2016 +0200
----------------------------------------------------------------------
.../org/apache/flink/api/common/ExecutionConfig.java | 11 ++++++++---
.../api/environment/StreamExecutionEnvironment.java | 4 ++--
2 files changed, 10 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/82ef021c/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 81ee930..aadf867 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
@@ -311,6 +311,7 @@ public class ExecutionConfig implements Serializable {
* @return The specified restart configuration
*/
@PublicEvolving
+ @SuppressWarnings("deprecation")
public RestartStrategies.RestartStrategyConfiguration getRestartStrategy() {
if (restartStrategyConfiguration == null) {
// support the old API calls by creating a restart strategy from them
@@ -601,11 +602,15 @@ public class ExecutionConfig implements Serializable {
* @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) {
+ @SuppressWarnings("rawtypes")
+ public void registerTypeWithKryoSerializer(Class<?> type, Class<? extends Serializer> serializerClass) {
if (type == null || serializerClass == null) {
throw new NullPointerException("Cannot register null class or serializer.");
}
- registeredTypesWithKryoSerializerClasses.put(type, serializerClass);
+
+ @SuppressWarnings("unchecked")
+ Class<? extends Serializer<?>> castedSerializerClass = (Class<? extends Serializer<?>>) serializerClass;
+ registeredTypesWithKryoSerializerClasses.put(type, castedSerializerClass);
}
/**
@@ -620,7 +625,7 @@ public class ExecutionConfig implements Serializable {
if (type == null) {
throw new NullPointerException("Cannot register null type class.");
}
- if(!registeredPojoTypes.contains(type)) {
+ if (!registeredPojoTypes.contains(type)) {
registeredPojoTypes.add(type);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/82ef021c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 06823d5..8fc2872 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -577,8 +577,8 @@ public abstract class StreamExecutionEnvironment {
* @param serializerClass
* The class of the serializer to use.
*/
- public void registerTypeWithKryoSerializer(Class<?> type,
- Class<? extends Serializer<?>> serializerClass) {
+ @SuppressWarnings("rawtypes")
+ public void registerTypeWithKryoSerializer(Class<?> type, Class<? extends Serializer> serializerClass) {
config.registerTypeWithKryoSerializer(type, serializerClass);
}
[4/6] flink git commit: [FLINK-4640] [streaming api] Ensure that the
state descriptors properly initialize the serializers.
Posted by se...@apache.org.
[FLINK-4640] [streaming api] Ensure that the state descriptors properly initialize the serializers.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4d4eb64b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4d4eb64b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4d4eb64b
Branch: refs/heads/master
Commit: 4d4eb64be7490672771243147824a70d3d47c501
Parents: 82ef021
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Sep 20 20:49:40 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Sep 21 17:53:33 2016 +0200
----------------------------------------------------------------------
.../api/datastream/AllWindowedStream.java | 4 +-
.../api/datastream/WindowedStream.java | 4 +-
.../operators/windowing/WindowOperator.java | 43 ++--
.../operators/StateDescriptorPassingTest.java | 214 +++++++++++++++++++
.../operators/windowing/WindowOperatorTest.java | 2 +
5 files changed, 241 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4d4eb64b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
index 4b083c8..6b09f3c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
@@ -469,9 +469,7 @@ public class AllWindowedStream<T, W extends Window> {
} else {
FoldingStateDescriptor<T, R> stateDesc = new FoldingStateDescriptor<>("window-contents",
- initialValue,
- foldFunction,
- resultType);
+ initialValue, foldFunction, resultType.createSerializer(getExecutionEnvironment().getConfig()));
opName = "TriggerWindow(" + windowAssigner + ", " + stateDesc + ", " + trigger + ", " + udfName + ")";
http://git-wip-us.apache.org/repos/asf/flink/blob/4d4eb64b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
index e81d7af..ae98619 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
@@ -489,9 +489,7 @@ public class WindowedStream<T, K, W extends Window> {
} else {
FoldingStateDescriptor<T, R> stateDesc = new FoldingStateDescriptor<>("window-contents",
- initialValue,
- foldFunction,
- resultType);
+ initialValue, foldFunction, resultType.createSerializer(getExecutionEnvironment().getConfig()));
opName = "TriggerWindow(" + windowAssigner + ", " + stateDesc + ", " + trigger + ", " + udfName + ")";
http://git-wip-us.apache.org/repos/asf/flink/blob/4d4eb64b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
index 25ec519..dffa2a1 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
@@ -61,7 +61,6 @@ import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.Triggerable;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalWindowFunction;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.Preconditions;
import java.io.IOException;
import java.io.ObjectInputStream;
@@ -74,7 +73,8 @@ import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.ScheduledFuture;
-import static java.util.Objects.requireNonNull;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* An operator that implements the logic for windowing based on a {@link WindowAssigner} and
@@ -186,26 +186,29 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
/**
* Creates a new {@code WindowOperator} based on the given policies and user functions.
*/
- public WindowOperator(WindowAssigner<? super IN, W> windowAssigner,
- TypeSerializer<W> windowSerializer,
- KeySelector<IN, K> keySelector,
- TypeSerializer<K> keySerializer,
- StateDescriptor<? extends AppendingState<IN, ACC>, ?> windowStateDescriptor,
- InternalWindowFunction<ACC, OUT, K, W> windowFunction,
- Trigger<? super IN, ? super W> trigger,
- long allowedLateness) {
+ public WindowOperator(
+ WindowAssigner<? super IN, W> windowAssigner,
+ TypeSerializer<W> windowSerializer,
+ KeySelector<IN, K> keySelector,
+ TypeSerializer<K> keySerializer,
+ StateDescriptor<? extends AppendingState<IN, ACC>, ?> windowStateDescriptor,
+ InternalWindowFunction<ACC, OUT, K, W> windowFunction,
+ Trigger<? super IN, ? super W> trigger,
+ long allowedLateness) {
super(windowFunction);
- this.windowAssigner = requireNonNull(windowAssigner);
- this.windowSerializer = windowSerializer;
- this.keySelector = requireNonNull(keySelector);
- this.keySerializer = requireNonNull(keySerializer);
+ checkArgument(allowedLateness >= 0);
- this.windowStateDescriptor = windowStateDescriptor;
- this.trigger = requireNonNull(trigger);
+ checkArgument(windowStateDescriptor == null || windowStateDescriptor.isSerializerInitialized(),
+ "window state serializer is not properly initialized");
- Preconditions.checkArgument(allowedLateness >= 0);
+ this.windowAssigner = checkNotNull(windowAssigner);
+ this.windowSerializer = checkNotNull(windowSerializer);
+ this.keySelector = checkNotNull(keySelector);
+ this.keySerializer = checkNotNull(keySerializer);
+ this.windowStateDescriptor = windowStateDescriptor;
+ this.trigger = checkNotNull(trigger);
this.allowedLateness = allowedLateness;
setChainingStrategy(ChainingStrategy.ALWAYS);
@@ -666,7 +669,7 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
public <S extends Serializable> ValueState<S> getKeyValueState(String name,
Class<S> stateType,
S defaultState) {
- requireNonNull(stateType, "The state type class must not be null");
+ checkNotNull(stateType, "The state type class must not be null");
TypeInformation<S> typeInfo;
try {
@@ -686,8 +689,8 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
TypeInformation<S> stateType,
S defaultState) {
- requireNonNull(name, "The name of the state must not be null");
- requireNonNull(stateType, "The state type information must not be null");
+ checkNotNull(name, "The name of the state must not be null");
+ checkNotNull(stateType, "The state type information must not be null");
ValueStateDescriptor<S> stateDesc = new ValueStateDescriptor<>(name, stateType.createSerializer(getExecutionConfig()), defaultState);
return getPartitionedState(stateDesc);
http://git-wip-us.apache.org/repos/asf/flink/blob/4d4eb64b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java
new file mode 100644
index 0000000..c0ca6a0
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StateDescriptorPassingTest.java
@@ -0,0 +1,214 @@
+/*
+ * 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.streaming.api.operators;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.serializers.JavaSerializer;
+
+import org.apache.flink.api.common.functions.FoldFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
+import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator;
+import org.apache.flink.util.Collector;
+
+import org.junit.Test;
+
+import java.io.File;
+
+import static org.junit.Assert.*;
+
+/**
+ * Various tests around the proper passing of state descriptors to the operators
+ * and their serialization.
+ *
+ * The tests use an arbitrary generic type to validate the behavior.
+ */
+@SuppressWarnings("serial")
+public class StateDescriptorPassingTest {
+
+ @Test
+ public void testFoldWindowState() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+ env.registerTypeWithKryoSerializer(File.class, JavaSerializer.class);
+
+ DataStream<String> src = env.fromElements("abc");
+
+ SingleOutputStreamOperator<?> result = src
+ .keyBy(new KeySelector<String, String>() {
+ @Override
+ public String getKey(String value) {
+ return null;
+ }
+ })
+ .timeWindow(Time.milliseconds(1000))
+ .fold(new File("/"), new FoldFunction<String, File>() {
+
+ @Override
+ public File fold(File a, String e) {
+ return null;
+ }
+ });
+
+ validateStateDescriptorConfigured(result);
+ }
+
+ @Test
+ public void testReduceWindowState() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+ env.registerTypeWithKryoSerializer(File.class, JavaSerializer.class);
+
+ DataStream<File> src = env.fromElements(new File("/"));
+
+ SingleOutputStreamOperator<?> result = src
+ .keyBy(new KeySelector<File, String>() {
+ @Override
+ public String getKey(File value) {
+ return null;
+ }
+ })
+ .timeWindow(Time.milliseconds(1000))
+ .reduce(new ReduceFunction<File>() {
+
+ @Override
+ public File reduce(File value1, File value2) {
+ return null;
+ }
+ });
+
+ validateStateDescriptorConfigured(result);
+ }
+
+ @Test
+ public void testApplyWindowState() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+ env.registerTypeWithKryoSerializer(File.class, JavaSerializer.class);
+
+ DataStream<File> src = env.fromElements(new File("/"));
+
+ SingleOutputStreamOperator<?> result = src
+ .keyBy(new KeySelector<File, String>() {
+ @Override
+ public String getKey(File value) {
+ return null;
+ }
+ })
+ .timeWindow(Time.milliseconds(1000))
+ .apply(new WindowFunction<File, String, String, TimeWindow>() {
+ @Override
+ public void apply(String s, TimeWindow window,
+ Iterable<File> input, Collector<String> out) {}
+ });
+
+ validateStateDescriptorConfigured(result);
+ }
+
+ @Test
+ public void testFoldWindowAllState() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+ env.registerTypeWithKryoSerializer(File.class, JavaSerializer.class);
+
+ DataStream<String> src = env.fromElements("abc");
+
+ SingleOutputStreamOperator<?> result = src
+ .timeWindowAll(Time.milliseconds(1000))
+ .fold(new File("/"), new FoldFunction<String, File>() {
+
+ @Override
+ public File fold(File a, String e) {
+ return null;
+ }
+ });
+
+ validateStateDescriptorConfigured(result);
+ }
+
+ @Test
+ public void testReduceWindowAllState() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+ env.registerTypeWithKryoSerializer(File.class, JavaSerializer.class);
+
+ DataStream<File> src = env.fromElements(new File("/"));
+
+ SingleOutputStreamOperator<?> result = src
+ .timeWindowAll(Time.milliseconds(1000))
+ .reduce(new ReduceFunction<File>() {
+
+ @Override
+ public File reduce(File value1, File value2) {
+ return null;
+ }
+ });
+
+ validateStateDescriptorConfigured(result);
+ }
+
+ @Test
+ public void testApplyWindowAllState() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+ env.registerTypeWithKryoSerializer(File.class, JavaSerializer.class);
+
+ DataStream<File> src = env.fromElements(new File("/"));
+
+ SingleOutputStreamOperator<?> result = src
+ .timeWindowAll(Time.milliseconds(1000))
+ .apply(new AllWindowFunction<File, String, TimeWindow>() {
+ @Override
+ public void apply(TimeWindow window, Iterable<File> input, Collector<String> out) {}
+ });
+
+ validateStateDescriptorConfigured(result);
+ }
+
+ // ------------------------------------------------------------------------
+ // generic validation
+ // ------------------------------------------------------------------------
+
+ private void validateStateDescriptorConfigured(SingleOutputStreamOperator<?> result) {
+ OneInputTransformation<?, ?> transform = (OneInputTransformation<?, ?>) result.getTransformation();
+ WindowOperator<?, ?, ?, ?, ?> op = (WindowOperator<?, ?, ?, ?, ?>) transform.getOperator();
+ StateDescriptor<?, ?> descr = op.getStateDescriptor();
+
+ // this would be the first statement to fail if state descriptors were not properly initialized
+ TypeSerializer<?> serializer = descr.getSerializer();
+ assertTrue(serializer instanceof KryoSerializer);
+
+ Kryo kryo = ((KryoSerializer<?>) serializer).getKryo();
+
+ assertTrue("serializer registration was not properly passed on",
+ kryo.getSerializer(File.class) instanceof JavaSerializer);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/4d4eb64b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java
index 67a6f55..fd73bcc 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java
@@ -2102,6 +2102,7 @@ public class WindowOperatorTest extends TestLogger {
}
},
inputType);
+ windowStateDesc.initializeSerializerUnlessSet(new ExecutionConfig());
WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator =
new WindowOperator<>(
@@ -2246,6 +2247,7 @@ public class WindowOperatorTest extends TestLogger {
}
},
inputType);
+ windowStateDesc.initializeSerializerUnlessSet(new ExecutionConfig());
WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator =
new WindowOperator<>(
[3/6] flink git commit: [hotfix] [streaming api] Add proper
deprecation JavaDocs
Posted by se...@apache.org.
[hotfix] [streaming api] Add proper deprecation JavaDocs
Also includes minor style cleanup of a test.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/349d06d6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/349d06d6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/349d06d6
Branch: refs/heads/master
Commit: 349d06d6c67a4b48e8ec3b507a98616580a8d863
Parents: db45da7
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Sep 20 18:53:46 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Sep 21 17:53:32 2016 +0200
----------------------------------------------------------------------
.../api/environment/StreamExecutionEnvironment.java | 14 ++++++++++++--
.../org/apache/flink/streaming/api/TypeFillTest.java | 10 +++++-----
2 files changed, 17 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/349d06d6/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 78aab97..06823d5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -577,8 +577,8 @@ public abstract class StreamExecutionEnvironment {
* @param serializerClass
* The class of the serializer to use.
*/
- public void registerTypeWithKryoSerializer(Class<?> type,
- Class<? extends Serializer<?>> serializerClass) {
+ public void registerTypeWithKryoSerializer(Class<?> type,
+ Class<? extends Serializer<?>> serializerClass) {
config.registerTypeWithKryoSerializer(type, serializerClass);
}
@@ -1094,8 +1094,11 @@ public abstract class StreamExecutionEnvironment {
* contents
* of files.
* @return The DataStream containing the given directory.
+ *
+ * @deprecated Use {@link #readFile(FileInputFormat, String, FileProcessingMode, long)} instead.
*/
@Deprecated
+ @SuppressWarnings("deprecation")
public DataStream<String> readFileStream(String filePath, long intervalMillis,
FileMonitoringFunction.WatchType watchType) {
DataStream<Tuple3<String, Long, Long>> source = addSource(new FileMonitoringFunction(
@@ -1169,6 +1172,8 @@ public abstract class StreamExecutionEnvironment {
* while
* a negative value ensures retrying forever.
* @return A data stream containing the strings received from the socket
+ *
+ * @deprecated Use {@link #socketTextStream(String, int, String, long)} instead.
*/
@Deprecated
public DataStreamSource<String> socketTextStream(String hostname, int port, char delimiter, long maxRetry) {
@@ -1215,8 +1220,11 @@ public abstract class StreamExecutionEnvironment {
* @param delimiter
* A character which splits received strings into records
* @return A data stream containing the strings received from the socket
+ *
+ * @deprecated Use {@link #socketTextStream(String, int, String)} instead.
*/
@Deprecated
+ @SuppressWarnings("deprecation")
public DataStreamSource<String> socketTextStream(String hostname, int port, char delimiter) {
return socketTextStream(hostname, port, delimiter, 0);
}
@@ -1313,7 +1321,9 @@ public abstract class StreamExecutionEnvironment {
DataStreamSource<OUT> source;
if (inputFormat instanceof FileInputFormat) {
+ @SuppressWarnings("unchecked")
FileInputFormat<OUT> format = (FileInputFormat<OUT>) inputFormat;
+
source = createFileInput(format, typeInfo, "Custom File source",
FileProcessingMode.PROCESS_ONCE, -1);
} else {
http://git-wip-us.apache.org/repos/asf/flink/blob/349d06d6/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
index d931f7b..a4abbcb 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
@@ -96,7 +96,7 @@ public class TypeFillTest {
}
- private class TestSource<T> implements SourceFunction<T> {
+ private static class TestSource<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
@Override
@@ -106,19 +106,19 @@ public class TypeFillTest {
public void cancel() {}
}
- private class TestMap<T, O> implements MapFunction<T, O> {
+ private static class TestMap<T, O> implements MapFunction<T, O> {
@Override
public O map(T value) throws Exception {
return null;
}
}
- private class TestFlatMap<T, O> implements FlatMapFunction<T, O> {
+ private static class TestFlatMap<T, O> implements FlatMapFunction<T, O> {
@Override
public void flatMap(T value, Collector<O> out) throws Exception {}
}
- private class TestCoMap<IN1, IN2, OUT> implements CoMapFunction<IN1, IN2, OUT> {
+ private static class TestCoMap<IN1, IN2, OUT> implements CoMapFunction<IN1, IN2, OUT> {
@Override
public OUT map1(IN1 value) {
@@ -132,7 +132,7 @@ public class TypeFillTest {
}
- private class TestCoFlatMap<IN1, IN2, OUT> implements CoFlatMapFunction<IN1, IN2, OUT> {
+ private static class TestCoFlatMap<IN1, IN2, OUT> implements CoFlatMapFunction<IN1, IN2, OUT> {
@Override
public void flatMap1(IN1 value, Collector<OUT> out) throws Exception {}
[5/6] flink git commit: [hotfix] [tests] Fix race condition in
RescalingITCase that could make the test stuck in a blocking call until
timeout
Posted by se...@apache.org.
[hotfix] [tests] Fix race condition in RescalingITCase that could make the test stuck in a blocking call until timeout
This closes #2513
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5f67b54b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5f67b54b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5f67b54b
Branch: refs/heads/master
Commit: 5f67b54b2ca4f7ea79d184e65a99ef230dbdc660
Parents: 4d4eb64
Author: Stefan Richter <s....@data-artisans.com>
Authored: Mon Sep 19 17:54:23 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Sep 21 17:53:33 2016 +0200
----------------------------------------------------------------------
.../test/checkpointing/RescalingITCase.java | 51 ++++++++++++--------
1 file changed, 30 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5f67b54b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
index 7f1d7f3..263bf79 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
@@ -69,9 +69,9 @@ import static org.junit.Assert.fail;
public class RescalingITCase extends TestLogger {
- private static int numTaskManagers = 2;
- private static int slotsPerTaskManager = 2;
- private static int numSlots = numTaskManagers * slotsPerTaskManager;
+ private static final int numTaskManagers = 2;
+ private static final int slotsPerTaskManager = 2;
+ private static final int numSlots = numTaskManagers * slotsPerTaskManager;
private static TestingCluster cluster;
@@ -109,12 +109,12 @@ public class RescalingITCase extends TestLogger {
*/
@Test
public void testSavepointRescalingWithPartitionedState() throws Exception {
- int numberKeys = 42;
- int numberElements = 1000;
- int numberElements2 = 500;
- int parallelism = numSlots / 2;
- int parallelism2 = numSlots;
- int maxParallelism = 13;
+ final int numberKeys = 42;
+ final int numberElements = 1000;
+ final int numberElements2 = 500;
+ final int parallelism = numSlots / 2;
+ final int parallelism2 = numSlots;
+ final int maxParallelism = 13;
FiniteDuration timeout = new FiniteDuration(3, TimeUnit.MINUTES);
Deadline deadline = timeout.fromNow();
@@ -214,9 +214,9 @@ public class RescalingITCase extends TestLogger {
*/
@Test
public void testSavepointRescalingFailureWithNonPartitionedState() throws Exception {
- int parallelism = numSlots / 2;
- int parallelism2 = numSlots;
- int maxParallelism = 13;
+ final int parallelism = numSlots / 2;
+ final int parallelism2 = numSlots;
+ final int maxParallelism = 13;
FiniteDuration timeout = new FiniteDuration(3, TimeUnit.MINUTES);
Deadline deadline = timeout.fromNow();
@@ -235,12 +235,14 @@ public class RescalingITCase extends TestLogger {
Object savepointResponse = null;
- // we might be too early for taking a savepoint if the operators have not been started yet
+ // wait until the operator is started
+ NonPartitionedStateSource.workStartedLatch.await();
+
while (deadline.hasTimeLeft()) {
Future<Object> savepointPathFuture = jobManager.ask(new JobManagerMessages.TriggerSavepoint(jobID), deadline.timeLeft());
-
- savepointResponse = Await.result(savepointPathFuture, deadline.timeLeft());
+ FiniteDuration waitingTime = new FiniteDuration(10, TimeUnit.SECONDS);
+ savepointResponse = Await.result(savepointPathFuture, waitingTime);
if (savepointResponse instanceof JobManagerMessages.TriggerSavepointSuccess) {
break;
@@ -428,6 +430,8 @@ public class RescalingITCase extends TestLogger {
env.enableCheckpointing(checkpointInterval);
env.setRestartStrategy(RestartStrategies.noRestart());
+ NonPartitionedStateSource.workStartedLatch = new CountDownLatch(1);
+
DataStream<Integer> input = env.addSource(new NonPartitionedStateSource());
input.addSink(new DiscardingSink<Integer>());
@@ -466,7 +470,7 @@ public class RescalingITCase extends TestLogger {
DataStream<Tuple2<Integer, Integer>> result = input.flatMap(new SubtaskIndexFlatMapper(numberElements));
- result.addSink(new CollectionSink());
+ result.addSink(new CollectionSink<Tuple2<Integer, Integer>>());
return env.getStreamGraph().getJobGraph();
}
@@ -504,7 +508,7 @@ public class RescalingITCase extends TestLogger {
DataStream<Tuple2<Integer, Integer>> result = input.flatMap(new SubtaskIndexFlatMapper(numberElements));
- result.addSink(new CollectionSink());
+ result.addSink(new CollectionSink<Tuple2<Integer, Integer>>());
return env.getStreamGraph().getJobGraph();
}
@@ -645,8 +649,10 @@ public class RescalingITCase extends TestLogger {
private static final long serialVersionUID = -8108185918123186841L;
- private int counter = 0;
- private boolean running = true;
+ private static volatile CountDownLatch workStartedLatch = new CountDownLatch(1);
+
+ private volatile int counter = 0;
+ private volatile boolean running = true;
@Override
public Integer snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
@@ -669,13 +675,16 @@ public class RescalingITCase extends TestLogger {
ctx.collect(counter * getRuntimeContext().getIndexOfThisSubtask());
}
- Thread.sleep(100);
+ Thread.sleep(2);
+ if(counter == 10) {
+ workStartedLatch.countDown();
+ }
}
}
@Override
public void cancel() {
- running = true;
+ running = false;
}
}
}
[2/6] flink git commit: [hotfix] [docs] Parametrized version of
quickstart script and artifacts in the quickstart docs
Posted by se...@apache.org.
[hotfix] [docs] Parametrized version of quickstart script and artifacts in the quickstart docs
This closes #2522
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/db45da7d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/db45da7d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/db45da7d
Branch: refs/heads/master
Commit: db45da7d70aa1b84f7fbf0e74263e7e115f5ddd5
Parents: 7212202
Author: chobeat <si...@gmail.com>
Authored: Tue Sep 20 16:27:16 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Sep 21 17:53:32 2016 +0200
----------------------------------------------------------------------
docs/quickstart/java_api_quickstart.md | 2 +-
docs/quickstart/run_example_quickstart.md | 2 +-
docs/quickstart/scala_api_quickstart.md | 2 +-
3 files changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/db45da7d/docs/quickstart/java_api_quickstart.md
----------------------------------------------------------------------
diff --git a/docs/quickstart/java_api_quickstart.md b/docs/quickstart/java_api_quickstart.md
index 389c25a..1c7bd8a 100644
--- a/docs/quickstart/java_api_quickstart.md
+++ b/docs/quickstart/java_api_quickstart.md
@@ -71,7 +71,7 @@ A note to Mac OS X users: The default JVM heapsize for Java is too small for Fli
## Build Project
-If you want to __build your project__, go to your project directory and issue the `mvn clean install -Pbuild-jar` command. You will __find a jar__ that runs on every Flink cluster in __target/your-artifact-id-1.0-SNAPSHOT.jar__. There is also a fat-jar, __target/your-artifact-id-1.0-SNAPSHOT-flink-fat-jar.jar__. This
+If you want to __build your project__, go to your project directory and issue the `mvn clean install -Pbuild-jar` command. You will __find a jar__ that runs on every Flink cluster in __target/your-artifact-id-{{ site.version }}.jar__. There is also a fat-jar, __target/your-artifact-id-{{ site.version }}-flink-fat-jar.jar__. This
also contains all dependencies that get added to the maven project.
## Next Steps
http://git-wip-us.apache.org/repos/asf/flink/blob/db45da7d/docs/quickstart/run_example_quickstart.md
----------------------------------------------------------------------
diff --git a/docs/quickstart/run_example_quickstart.md b/docs/quickstart/run_example_quickstart.md
index e079280..a64e87f 100644
--- a/docs/quickstart/run_example_quickstart.md
+++ b/docs/quickstart/run_example_quickstart.md
@@ -44,7 +44,7 @@ about this. For our purposes, the command to run is this:
$ mvn archetype:generate\
-DarchetypeGroupId=org.apache.flink\
-DarchetypeArtifactId=flink-quickstart-java\
- -DarchetypeVersion=1.0.0\
+ -DarchetypeVersion={{ site.version }}\
-DgroupId=wiki-edits\
-DartifactId=wiki-edits\
-Dversion=0.1\
http://git-wip-us.apache.org/repos/asf/flink/blob/db45da7d/docs/quickstart/scala_api_quickstart.md
----------------------------------------------------------------------
diff --git a/docs/quickstart/scala_api_quickstart.md b/docs/quickstart/scala_api_quickstart.md
index 16101d5..22562f4 100644
--- a/docs/quickstart/scala_api_quickstart.md
+++ b/docs/quickstart/scala_api_quickstart.md
@@ -171,7 +171,7 @@ The IntelliJ IDE also supports Maven and offers a plugin for Scala development.
### Build Project
-If you want to __build your project__, go to your project directory and issue the `mvn clean package -Pbuild-jar` command. You will __find a jar__ that runs on every Flink cluster in __target/your-artifact-id-1.0-SNAPSHOT.jar__. There is also a fat-jar, __target/your-artifact-id-1.0-SNAPSHOT-flink-fat-jar.jar__. This
+If you want to __build your project__, go to your project directory and issue the `mvn clean package -Pbuild-jar` command. You will __find a jar__ that runs on every Flink cluster in __target/your-artifact-id-{{ site.version }}.jar__. There is also a fat-jar, __target/your-artifact-id-{{ site.version }}-flink-fat-jar.jar__. This
also contains all dependencies that get added to the maven project.
## Next Steps
[6/6] flink git commit: [docs] Update docs on data types and
serialization, to include type hints, type registration,
and serializer registration.
Posted by se...@apache.org.
[docs] Update docs on data types and serialization, to include type hints, type registration, and serializer registration.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b5160568
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b5160568
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b5160568
Branch: refs/heads/master
Commit: b516056869a825db0913ff852c071de198f4d390
Parents: 5f67b54
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Sep 20 23:10:44 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Sep 21 17:53:33 2016 +0200
----------------------------------------------------------------------
docs/dev/types_serialization.md | 153 +++++++++++++++++++++++------------
1 file changed, 102 insertions(+), 51 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b5160568/docs/dev/types_serialization.md
----------------------------------------------------------------------
diff --git a/docs/dev/types_serialization.md b/docs/dev/types_serialization.md
index 8a32491..4b8e25f 100644
--- a/docs/dev/types_serialization.md
+++ b/docs/dev/types_serialization.md
@@ -1,5 +1,5 @@
---
-title: "Data Types"
+title: "Data Types & Serialization"
nav-id: types
nav-parent_id: dev
nav-pos: 9
@@ -23,13 +23,8 @@ specific language governing permissions and limitations
under the License.
-->
-Flink handles types in a unique way, containing its own type descriptors,
-generic type extraction, and type serialization framework.
-This document describes the concepts and the rationale behind them.
-
-There are fundamental differences in the way that the Scala API and
-the Java API handle type information, so most of the issues described
-here relate only to one of the to APIs.
+Apache Flink handles data types and serialization in a unique way, containing its own type descriptors,
+generic type extraction, and type serialization framework. This document describes the concepts and the rationale behind them.
* This will be replaced by the TOC
{:toc}
@@ -37,21 +32,43 @@ here relate only to one of the to APIs.
## Type handling in Flink
-Flink tries to know as much information about what types enter and leave user functions as possible.
-This stands in contrast to the approach to just assuming nothing and letting the
-programming language and serialization framework handle all types dynamically.
+Flink tries to infer a lot of information about the data types that are exchanged and stored during the distributed computation.
+Think about it like a database that infers the schema of tables. In most cases, Flink infers all necessary information seamlessly
+by itself. Having the type information allows Flink to do some cool things:
+
+* Using POJOs types and grouping / joining / aggregating them by referring to field names (like `dataSet.keyBy("username")`).
+ The type information allows Flink to check (for typos and type compatibility) early rather than failing later ar runtime.
+
+* The more Flink knows about data types, the better the serialization and data layout schemes are.
+ That is quite important for the memory usage paradigm in Flink (work on serialized data inside/outside the heap where ever possible
+ and make serialization very cheap).
+
+* Finally, it also spares users in the majority of cases from worrying about serialization frameworks and having to register types.
+
+In general, the information about data types is needed during the *pre-flight phase* - that is, when the program's calls on `DataStream`
+and `DataSet` are made, and before any call to `execute()`, `print()`, `count()`, or `collect()`.
+
+
+## Most Frequent Issues
+
+The most frequent issues where users need to interact with Flink's data type handling are:
-* To allow using POJOs and grouping/joining them by referring to field names, Flink needs the type
- information to make checks (for typos and type compatibility) before the job is executed.
+* **Registering subtypes:** If the function signatures describe only the supertypes, but they actually use subtypes of those during execution,
+ it may increase performance a lot to make Flink aware of these subtypes.
+ For that, call `.registerType(clazz)` on the `StreamExecutionEnvironment` or `ExecutionEnvironment` for each subtype.
-* The more we know, the better serialization and data layout schemes the compiler/optimizer can develop.
- That is quite important for the memory usage paradigm in Flink (work on serialized data
- inside/outside the heap and make serialization very cheap).
+* **Registering custom serializers:** Flink falls back to [Kryo](https://github.com/EsotericSoftware/kryo) for the types that it does not handle transparently
+ by itself. Not all types are seamlessly handled by Kryo (and thus by Flink). For example, many Google Guava collection types do not work well
+ by default. The solution is to register additional serializers for the types that cause problems.
+ Call `.getConfig().addDefaultKryoSerializer(clazz, serializer)` on the `StreamExecutionEnvironment` or `ExecutionEnvironment`.
+ Additional Kryo serializers are available in many libraries.
-* For the upcoming logical programs (see roadmap draft) we need this to know the "schema" of functions.
+* **Adding Type Hints:** Sometimes, when Flink cannot infer the generic types despits all tricks, a user must pass a *type hint*. That is generally
+ only necessary in the Java API. The [Type Hints Section](#type-hints-in-the-java-api) describes that in more detail.
-* Finally, it also spares users having to worry about serialization frameworks and having to register
- types at those frameworks.
+* **Manually creating a `TypeInformation`:** This may be necessary for some API calls where it is not possible for Flink to infer
+ the data types due to Java's generic type erasure. See [Creating a TypeInformation or TypeSerializer](#creating-a-typeinformation-or-typeserializer)
+ for details.
## Flink's TypeInformation class
@@ -75,7 +92,7 @@ Internally, Flink makes the following distinctions between types:
* POJOs: classes that follow a certain bean-like pattern
-* Scala auxiliary types (Option, Either, Lists, Maps, ...)
+* Auxiliary types (Option, Either, Lists, Maps, ...)
* Generic types: These will not be serialized by Flink itself, but by Kryo.
@@ -84,18 +101,66 @@ names in the definition of keys: `dataSet.join(another).where("name").equalTo("p
They are also transparent to the runtime and can be handled very efficiently by Flink.
-**Rules for POJO types**
+#### Rules for POJO types
Flink recognizes a data type as a POJO type (and allows "by-name" field referencing) if the following
conditions are fulfilled:
* The class is public and standalone (no non-static inner class)
* The class has a public no-argument constructor
-* All fields in the class (and all superclasses) are either public or
- or have a public getter and a setter method that follows the Java beans
+* All fields in the class (and all superclasses) are either public (and non-final)
+ or have a public getter- and a setter- method that follows the Java beans
naming conventions for getters and setters.
+#### Creating a TypeInformation or TypeSerializer
+
+To create a TypeInformation object for a type, use the language specific way:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+Because Java generally erases generic type information, you need to pass the type to the TypeInformation
+construction:
+
+For non-generic types, you can pass the Class:
+{% highlight java %}
+TypeInformation<String> info = TypeInformation.of(String.class);
+{% endhighlight %}
+
+For generic types, you need to "capture" the generic type information via the `TypeHint`:
+{% highlight java %}
+TypeInformation<Tuple2<String, Double>> info = TypeInformation.of(new TypeHint<Tuple2<String, Double>>(){});
+{% endhighlight %}
+Internally, this creates an anonymous subclass of the TypeHint that captures the generic information to preserve it
+until runtime.
+</div>
+
+<div data-lang="scala" markdown="1">
+In Scala, Flink uses *macros* that runs at compile time and captures all generic type information while it is
+still available.
+{% highlight scala %}
+// important: this import is needed to access the 'createTypeInformation' macro function
+import org.apache.flink.streaming.api.scala._
+
+val stringInfo: TypeInformation[String] = createTypeInformation[String]
+
+val tupleInfo: TypeInformation[(String, Double)] = createTypeInformation[(String, Double)]
+{% endhighlight %}
+
+You can still use the same method as in Java as a fallback.
+</div>
+</div>
+
+To create a `TypeSerializer`, simply call `typeInfo.createSerializer(config)` on the `TypeInformation` object.
+
+The `config` parameter is of type `ExecutionConfig` and holds the information about the program's registered
+custom serializers. Where ever possibly, try to pass the programs proper ExecutionConfig. You can usually
+obtain it from `DataStream` or `DataSet` via calling `getExecutionConfig()`. Inside functions (like `MapFunction`), you can
+get it by making the function a [Rich Function]() and calling `getRuntimeContext().getExecutionConfig()`.
+
+--------
+--------
+
## Type Information in the Scala API
Scala has very elaborate concepts for runtime type information though *type manifests* and *class tags*. In
@@ -156,15 +221,15 @@ def selectFirst[T : TypeInformation](input: DataSet[(T, _)]) : DataSet[T] = {
{% endhighlight %}
+--------
+--------
-## Type Information in the Java API
-Java in general erases generic type information. Only for subclasses of generic classes, the subclass
-stores the type to which the generic type variables bind.
+## Type Information in the Java API
-Flink uses reflection on the (anonymous) classes that implement the user functions to figure out the types of
-the generic parameters of the function. This logic also contains some simple type inference for cases where
-the return types of functions are dependent on input types, such as in the generic utility method below:
+In the general case, Java erases generic type information. Flink tries to reconstruct as much type information
+as possible via reflection, using the few bits that Java preserves (mainly function signatures and subclass information).
+This logic also contains some simple type inference for cases where the return type of a function depends on its input type:
{% highlight java %}
public class AppendOne<T> extends MapFunction<T, Tuple2<T, Long>> {
@@ -175,16 +240,14 @@ public class AppendOne<T> extends MapFunction<T, Tuple2<T, Long>> {
}
{% endhighlight %}
-Not in all cases can Flink figure out the data types of functions reliably in Java.
-Some issues remain with generic lambdas (we are trying to solve this with the Java community,
-see below) and with generic type variables that we cannot infer.
+There are cases where Flink cannot reconstruct all generic type information. In that case, a user has to help out via *type hints*.
#### Type Hints in the Java API
-To help cases where Flink cannot reconstruct the erased generic type information, the Java API
-offers so called *type hints* from version 0.9 on. The type hints tell the system the type of
-the data set produced by a function. The following gives an example:
+In cases where Flink cannot reconstruct the erased generic type information, the Java API
+offers so called *type hints*. The type hints tell the system the type of
+the data stream or data set produced by a function:
{% highlight java %}
DataSet<SomeType> result = dataSet
@@ -193,12 +256,11 @@ DataSet<SomeType> result = dataSet
{% endhighlight %}
The `returns` statement specifies the produced type, in this case via a class. The hints support
-type definition through
+type definition via
* Classes, for non-parameterized types (no generics)
-* Strings in the form of `returns("Tuple2<Integer, my.SomeType>")`, which are parsed and converted
- to a TypeInformation.
-* A TypeInformation directly
+* TypeHints in the form of `returns(new TypeHint<Tuple2<Integer, SomeType>>(){})`. The `TypeHint` class
+ can capture generic type information and preserve it for the runtime (via an anonymous subclass).
#### Type extraction for Java 8 lambdas
@@ -208,18 +270,7 @@ with an implementing class that extends the function interface.
Currently, Flink tries to figure out which method implements the lambda and uses Java's generic signatures to
determine the parameter types and the return type. However, these signatures are not generated for lambdas
-by all compilers (as of writing this document only reliably by the Eclipse JDT compiler 4.5 from Milestone 2
-onwards)
-
-
-**Improving Type information for Java Lambdas**
-
-One of the Flink committers (Timo Walther) has actually become active in the Eclipse JDT compiler community and
-in the OpenJDK community and submitted patches to the compiler to improve availability of type information
-available for Java 8 lambdas.
-
-The Eclipse JDT compiler has added support for this as of version 4.5 M4. Discussion about the feature in the
-OpenJDK compiler is pending.
+by all compilers (as of writing this document only reliably by the Eclipse JDT compiler from 4.5 onwards).
#### Serialization of POJO types