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/03/05 00:21:48 UTC
[1/6] flink git commit: [FLINK-3566] [FLINK-3563] [core]
TypeExtraction input type validation fixes
Repository: flink
Updated Branches:
refs/heads/release-1.0 fd0830ead -> b3765408b
[FLINK-3566] [FLINK-3563] [core] TypeExtraction input type validation fixes
This closes #1759
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b3765408
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b3765408
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b3765408
Branch: refs/heads/release-1.0
Commit: b3765408bfd928ce31defbed8f83f3a08e548332
Parents: 2cde19b
Author: twalthr <tw...@apache.org>
Authored: Thu Mar 3 15:44:28 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Mar 4 20:53:44 2016 +0100
----------------------------------------------------------------------
.../flink/api/java/typeutils/TypeExtractor.java | 14 +--
.../api/java/typeutils/TypeExtractorTest.java | 112 +++++++++++++++++++
2 files changed, 119 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b3765408/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
index 01afe14..dd4b132 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
@@ -890,8 +890,8 @@ public class TypeExtractor {
}
if (!(type instanceof TypeVariable<?>)) {
- // check for basic type
- if (typeInfo.isBasicType()) {
+ // check for Java Basic Types
+ if (typeInfo instanceof BasicTypeInfo) {
TypeInformation<?> actual;
// check if basic type at all
@@ -904,8 +904,8 @@ public class TypeExtractor {
}
}
- // check for tuple
- else if (typeInfo.isTupleType()) {
+ // check for Java Tuples
+ else if (typeInfo instanceof TupleTypeInfo) {
// check if tuple at all
if (!(isClassType(type) && Tuple.class.isAssignableFrom(typeToClass(type)))) {
throw new InvalidTypesException("Tuple type expected.");
@@ -1079,9 +1079,9 @@ public class TypeExtractor {
// check for generic object
else if (typeInfo instanceof GenericTypeInfo<?>) {
Class<?> clazz = null;
- if (!(isClassType(type) && ((GenericTypeInfo<?>) typeInfo).getTypeClass() == (clazz = typeToClass(type)))) {
- throw new InvalidTypesException("Generic object type '"
- + ((GenericTypeInfo<?>) typeInfo).getTypeClass().getCanonicalName() + "' expected but was '"
+ if (!(isClassType(type) && (clazz = typeToClass(type)).isAssignableFrom(((GenericTypeInfo<?>) typeInfo).getTypeClass()))) {
+ throw new InvalidTypesException("Generic type '"
+ + ((GenericTypeInfo<?>) typeInfo).getTypeClass().getCanonicalName() + "' or a subclass of it expected but was '"
+ clazz.getCanonicalName() + "'.");
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b3765408/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
index cdafa6a..a8f5ded 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/TypeExtractorTest.java
@@ -22,8 +22,11 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichCoGroupFunction;
@@ -38,6 +41,7 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple0;
@@ -1904,4 +1908,112 @@ public class TypeExtractorTest {
Either<String, Tuple1<Integer>> either = Either.Left("test");
TypeExtractor.getForObject(either);
}
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Test
+ public void testGenericTypeWithSubclassInput() {
+ Map<String, Object> inputMap = new HashMap<>();
+ inputMap.put("a", "b");
+ TypeInformation<?> inputType = TypeExtractor.createTypeInfo(inputMap.getClass());
+
+ MapFunction<?, ?> function = new MapFunction<Map<String, Object>,Map<String, Object>>(){
+
+ @Override
+ public Map<String, Object> map(Map<String, Object> stringObjectMap) throws Exception {
+ return stringObjectMap;
+ }
+ };
+
+ TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, (TypeInformation) inputType);
+ TypeInformation<?> expected = TypeExtractor.createTypeInfo(Map.class);
+ Assert.assertEquals(expected, ti);
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Test(expected=InvalidTypesException.class)
+ public void testGenericTypeWithSuperclassInput() {
+ TypeInformation<?> inputType = TypeExtractor.createTypeInfo(Map.class);
+
+ MapFunction<?, ?> function = new MapFunction<HashMap<String, Object>,Map<String, Object>>(){
+
+ @Override
+ public Map<String, Object> map(HashMap<String, Object> stringObjectMap) throws Exception {
+ return stringObjectMap;
+ }
+ };
+
+ TypeExtractor.getMapReturnTypes(function, (TypeInformation) inputType);
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Test
+ public void testInputWithCustomTypeInfo() {
+ TypeInformation<?> customTypeInfo = new TypeInformation<Object>() {
+
+ @Override
+ public boolean isBasicType() {
+ return true;
+ }
+
+ @Override
+ public boolean isTupleType() {
+ return true;
+ }
+
+ @Override
+ public int getArity() {
+ return 0;
+ }
+
+ @Override
+ public int getTotalFields() {
+ return 0;
+ }
+
+ @Override
+ public Class getTypeClass() {
+ return null;
+ }
+
+ @Override
+ public boolean isKeyType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<Object> createSerializer(ExecutionConfig config) {
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ return null;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object obj) {
+ return false;
+ }
+ };
+
+ MapFunction<?, ?> function = new MapFunction<Tuple1<String>, Tuple1<Object>>() {
+ @Override
+ public Tuple1<Object> map(Tuple1<String> value) throws Exception {
+ return null;
+ }
+ };
+
+ TypeExtractor.getMapReturnTypes(function,
+ (TypeInformation) new TupleTypeInfo<Tuple1<Object>>(customTypeInfo));
+ }
}
[3/6] flink git commit: [FLINK-3578] [streaming scala] Support
RichFunctions for Scala WindowFunction and AllWindowFunction
Posted by se...@apache.org.
[FLINK-3578] [streaming scala] Support RichFunctions for Scala WindowFunction and AllWindowFunction
This also consolidates the various wrapper classes for fold() and reduce() functions.
This closes #1765
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/91587892
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/91587892
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/91587892
Branch: refs/heads/release-1.0
Commit: 91587892b527e7d56661b14bfe4f425a7ba403a4
Parents: fd0830e
Author: Stephan Ewen <se...@apache.org>
Authored: Fri Mar 4 12:23:45 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Mar 4 20:53:44 2016 +0100
----------------------------------------------------------------------
.../functions/windowing/RichWindowFunction.java | 17 +-
.../InternalIterableWindowFunction.java | 2 +-
.../streaming/api/scala/AllWindowedStream.scala | 111 +++-----
.../streaming/api/scala/WindowedStream.scala | 95 +++----
.../api/scala/function/AllWindowFunction.scala | 11 +-
.../scala/function/RichAllWindowFunction.scala | 37 +++
.../api/scala/function/RichWindowFunction.scala | 38 +++
.../function/util/ScalaAllWindowFunction.scala | 38 +++
.../util/ScalaAllWindowFunctionWrapper.scala | 68 +++++
.../scala/function/util/ScalaFoldFunction.scala | 33 +++
.../function/util/ScalaReduceFunction.scala | 33 +++
.../function/util/ScalaWindowFunction.scala | 38 +++
.../util/ScalaWindowFunctionWrapper.scala | 68 +++++
.../streaming/api/scala/WindowFoldITCase.scala | 123 ++++++++-
.../api/scala/WindowFunctionITCase.scala | 157 +++++++++++
.../api/scala/WindowReduceITCase.scala | 276 +++++++++++++++++++
.../CheckingIdentityRichAllWindowFunction.scala | 84 ++++++
.../CheckingIdentityRichWindowFunction.scala | 81 ++++++
18 files changed, 1168 insertions(+), 142 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/RichWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/RichWindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/RichWindowFunction.java
index 18ad19a..2ed271f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/RichWindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/RichWindowFunction.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -7,7 +7,7 @@
* "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
+ * 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,
@@ -15,12 +15,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.api.functions.windowing;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.streaming.api.windowing.windows.Window;
+/**
+ * Rich variant of the {@link WindowFunction}. As a {@link RichFunction}, it gives access to the
+ * {@link org.apache.flink.api.common.functions.RuntimeContext} and provides setup and tear-down methods:
+ * {@link RichFunction#open(org.apache.flink.configuration.Configuration)} and
+ * {@link RichFunction#close()}.
+ *
+ * @param <IN> The type of the input value.
+ * @param <OUT> The type of the output value.
+ * @param <KEY> The type of the key.
+ * @param <W> The type of {@code Window} that this window function can be applied on.
+ */
@Public
public abstract class RichWindowFunction<IN, OUT, KEY, W extends Window> extends AbstractRichFunction implements WindowFunction<IN, OUT, KEY, W> {
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalIterableWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalIterableWindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalIterableWindowFunction.java
index 32318ea..7b441fb 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalIterableWindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalIterableWindowFunction.java
@@ -33,7 +33,7 @@ import org.apache.flink.util.Collector;
public final class InternalIterableWindowFunction<IN, OUT, KEY, W extends Window> extends InternalWindowFunction<Iterable<IN>, OUT, KEY, W> implements RichFunction {
private static final long serialVersionUID = 1L;
- protected WindowFunction<IN, OUT, KEY, W> wrappedFunction;
+ protected final WindowFunction<IN, OUT, KEY, W> wrappedFunction;
public InternalIterableWindowFunction(WindowFunction<IN, OUT, KEY, W> wrappedFunction) {
this.wrappedFunction = wrappedFunction;
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/AllWindowedStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/AllWindowedStream.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/AllWindowedStream.scala
index dcb9822..020c619 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/AllWindowedStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/AllWindowedStream.scala
@@ -24,15 +24,13 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.datastream.{AllWindowedStream => JavaAllWStream}
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction.AggregationType
import org.apache.flink.streaming.api.functions.aggregation.{ComparableAggregator, SumAggregator}
-import org.apache.flink.streaming.api.functions.windowing.{AllWindowFunction => JAllWindowFunction}
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
+import org.apache.flink.streaming.api.scala.function.util.{ScalaAllWindowFunction, ScalaAllWindowFunctionWrapper, ScalaReduceFunction, ScalaFoldFunction}
import org.apache.flink.streaming.api.windowing.evictors.Evictor
import org.apache.flink.streaming.api.windowing.triggers.Trigger
import org.apache.flink.streaming.api.windowing.windows.Window
import org.apache.flink.util.Collector
-import scala.collection.JavaConverters._
-
/**
* A [[AllWindowedStream]] represents a data stream where the stream of
* elements is split into windows based on a
@@ -56,7 +54,7 @@ import scala.collection.JavaConverters._
*/
@Public
class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
-
+
/**
* Sets the [[Trigger]] that should be used to trigger window emission.
*/
@@ -121,9 +119,8 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
throw new NullPointerException("Reduce function must not be null.")
}
val cleanFun = clean(function)
- val reducer = new ReduceFunction[T] {
- def reduce(v1: T, v2: T) = { cleanFun(v1, v2) }
- }
+ val reducer = new ScalaReduceFunction[T](cleanFun)
+
reduce(reducer)
}
@@ -160,11 +157,8 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
throw new NullPointerException("Fold function must not be null.")
}
val cleanFun = clean(function)
- val folder = new FoldFunction[T,R] {
- def fold(acc: R, v: T) = {
- cleanFun(acc, v)
- }
- }
+ val folder = new ScalaFoldFunction[T,R](cleanFun)
+
fold(initialValue, folder)
}
@@ -183,11 +177,8 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
function: AllWindowFunction[T, R, W]): DataStream[R] = {
val cleanedFunction = clean(function)
- val javaFunction = new JAllWindowFunction[T, R, W] {
- def apply(window: W, elements: java.lang.Iterable[T], out: Collector[R]): Unit = {
- cleanedFunction(window, elements.asScala, out)
- }
- }
+ val javaFunction = new ScalaAllWindowFunctionWrapper[T, R, W](cleanedFunction)
+
asScalaStream(javaStream.apply(javaFunction, implicitly[TypeInformation[R]]))
}
@@ -206,11 +197,8 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
function: (W, Iterable[T], Collector[R]) => Unit): DataStream[R] = {
val cleanedFunction = clean(function)
- val applyFunction = new JAllWindowFunction[T, R, W] {
- def apply(window: W, elements: java.lang.Iterable[T], out: Collector[R]): Unit = {
- cleanedFunction(window, elements.asScala, out)
- }
- }
+ val applyFunction = new ScalaAllWindowFunction[T, R, W](cleanedFunction)
+
asScalaStream(javaStream.apply(applyFunction, implicitly[TypeInformation[R]]))
}
@@ -222,22 +210,20 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
* Arriving data is pre-aggregated using the given pre-aggregation reducer.
*
* @param preAggregator The reduce function that is used for pre-aggregation
- * @param function The window function.
+ * @param windowFunction The window function.
* @return The data stream that is the result of applying the window function to the window.
*/
def apply[R: TypeInformation](
preAggregator: ReduceFunction[T],
- function: AllWindowFunction[T, R, W]): DataStream[R] = {
+ windowFunction: AllWindowFunction[T, R, W]): DataStream[R] = {
- val cleanedFunction = clean(function)
- val applyFunction = new JAllWindowFunction[T, R, W] {
- def apply(window: W, elements: java.lang.Iterable[T], out: Collector[R]): Unit = {
- cleanedFunction(window, elements.asScala, out)
- }
- }
+ val cleanedReducer = clean(preAggregator)
+ val cleanedWindowFunction = clean(windowFunction)
+
+ val applyFunction = new ScalaAllWindowFunctionWrapper[T, R, W](cleanedWindowFunction)
val returnType: TypeInformation[R] = implicitly[TypeInformation[R]]
- asScalaStream(javaStream.apply(clean(preAggregator), applyFunction, returnType))
+ asScalaStream(javaStream.apply(cleanedReducer, applyFunction, returnType))
}
/**
@@ -248,30 +234,25 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
* Arriving data is pre-aggregated using the given pre-aggregation reducer.
*
* @param preAggregator The reduce function that is used for pre-aggregation
- * @param function The window function.
+ * @param windowFunction The window function.
* @return The data stream that is the result of applying the window function to the window.
*/
def apply[R: TypeInformation](
preAggregator: (T, T) => T,
- function: (W, Iterable[T], Collector[R]) => Unit): DataStream[R] = {
- if (function == null) {
+ windowFunction: (W, Iterable[T], Collector[R]) => Unit): DataStream[R] = {
+
+ if (preAggregator == null) {
throw new NullPointerException("Reduce function must not be null.")
}
- if (function == null) {
+ if (windowFunction == null) {
throw new NullPointerException("WindowApply function must not be null.")
}
val cleanReducer = clean(preAggregator)
- val reducer = new ReduceFunction[T] {
- def reduce(v1: T, v2: T) = { cleanReducer(v1, v2) }
- }
-
- val cleanApply = clean(function)
- val applyFunction = new JAllWindowFunction[T, R, W] {
- def apply(window: W, input: java.lang.Iterable[T], out: Collector[R]): Unit = {
- cleanApply(window, input.asScala, out)
- }
- }
+ val cleanWindowFunction = clean(windowFunction)
+
+ val reducer = new ScalaReduceFunction[T](cleanReducer)
+ val applyFunction = new ScalaAllWindowFunction[T, R, W](cleanWindowFunction)
val returnType: TypeInformation[R] = implicitly[TypeInformation[R]]
asScalaStream(javaStream.apply(reducer, applyFunction, returnType))
@@ -286,24 +267,22 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
*
* @param initialValue Initial value of the fold
* @param preAggregator The reduce function that is used for pre-aggregation
- * @param function The window function.
+ * @param windowFunction The window function.
* @return The data stream that is the result of applying the window function to the window.
*/
def apply[R: TypeInformation](
initialValue: R,
preAggregator: FoldFunction[T, R],
- function: AllWindowFunction[R, R, W]): DataStream[R] = {
+ windowFunction: AllWindowFunction[R, R, W]): DataStream[R] = {
- val cleanedFunction = clean(function)
- val applyFunction = new JAllWindowFunction[R, R, W] {
- def apply(window: W, elements: java.lang.Iterable[R], out: Collector[R]): Unit = {
- cleanedFunction(window, elements.asScala, out)
- }
- }
+ val cleanFolder = clean(preAggregator)
+ val cleanWindowFunction = clean(windowFunction)
+
+ val applyFunction = new ScalaAllWindowFunctionWrapper[R, R, W](cleanWindowFunction)
asScalaStream(javaStream.apply(
initialValue,
- clean(preAggregator),
+ cleanFolder,
applyFunction,
implicitly[TypeInformation[R]]))
}
@@ -317,31 +296,27 @@ class AllWindowedStream[T, W <: Window](javaStream: JavaAllWStream[T, W]) {
*
* @param initialValue Initial value of the fold
* @param preAggregator The reduce function that is used for pre-aggregation
- * @param function The window function.
+ * @param windowFunction The window function.
* @return The data stream that is the result of applying the window function to the window.
*/
def apply[R: TypeInformation](
initialValue: R,
preAggregator: (R, T) => R,
- function: (W, Iterable[R], Collector[R]) => Unit): DataStream[R] = {
- if (function == null) {
+ windowFunction: (W, Iterable[R], Collector[R]) => Unit): DataStream[R] = {
+
+ if (preAggregator == null) {
throw new NullPointerException("Reduce function must not be null.")
}
- if (function == null) {
+ if (windowFunction == null) {
throw new NullPointerException("WindowApply function must not be null.")
}
val cleanFolder = clean(preAggregator)
- val folder = new FoldFunction[T, R] {
- def fold(v1: R, v2: T) = { cleanFolder(v1, v2) }
- }
-
- val cleanApply = clean(function)
- val applyFunction = new JAllWindowFunction[R, R, W] {
- def apply(window: W, input: java.lang.Iterable[R], out: Collector[R]): Unit = {
- cleanApply(window, input.asScala, out)
- }
- }
+ val cleanWindowFunction = clean(windowFunction)
+
+ val folder = new ScalaFoldFunction[T, R](cleanFolder)
+ val applyFunction = new ScalaAllWindowFunction[R, R, W](cleanWindowFunction)
+
val returnType: TypeInformation[R] = implicitly[TypeInformation[R]]
asScalaStream(javaStream.apply(initialValue, folder, applyFunction, returnType))
}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
index bea578e..773829e 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
@@ -25,14 +25,12 @@ import org.apache.flink.streaming.api.datastream.{WindowedStream => JavaWStream}
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction.AggregationType
import org.apache.flink.streaming.api.functions.aggregation.{ComparableAggregator, SumAggregator}
import org.apache.flink.streaming.api.scala.function.WindowFunction
-import org.apache.flink.streaming.api.functions.windowing.{WindowFunction => JWindowFunction}
+import org.apache.flink.streaming.api.scala.function.util.{ScalaFoldFunction, ScalaReduceFunction, ScalaWindowFunction, ScalaWindowFunctionWrapper}
import org.apache.flink.streaming.api.windowing.evictors.Evictor
import org.apache.flink.streaming.api.windowing.triggers.Trigger
import org.apache.flink.streaming.api.windowing.windows.Window
import org.apache.flink.util.Collector
-import scala.collection.JavaConverters._
-
/**
* A [[WindowedStream]] represents a data stream where elements are grouped by
* key, and for each key, the stream of elements is split into windows based on a
@@ -124,9 +122,7 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
throw new NullPointerException("Reduce function must not be null.")
}
val cleanFun = clean(function)
- val reducer = new ReduceFunction[T] {
- def reduce(v1: T, v2: T) = { cleanFun(v1, v2) }
- }
+ val reducer = new ScalaReduceFunction[T](cleanFun)
reduce(reducer)
}
@@ -163,11 +159,7 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
throw new NullPointerException("Fold function must not be null.")
}
val cleanFun = clean(function)
- val folder = new FoldFunction[T,R] {
- def fold(acc: R, v: T) = {
- cleanFun(acc, v)
- }
- }
+ val folder = new ScalaFoldFunction[T, R](cleanFun)
fold(initialValue, folder)
}
@@ -186,13 +178,8 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
function: WindowFunction[T, R, K, W]): DataStream[R] = {
val cleanFunction = clean(function)
- val javaFunction = new JWindowFunction[T, R, K, W] {
- def apply(key: K, window: W, input: java.lang.Iterable[T], out: Collector[R]) = {
- cleanFunction.apply(key, window, input.asScala, out)
- }
- }
-
- asScalaStream(javaStream.apply(javaFunction, implicitly[TypeInformation[R]]))
+ val applyFunction = new ScalaWindowFunctionWrapper[T, R, K, W](cleanFunction)
+ asScalaStream(javaStream.apply(applyFunction, implicitly[TypeInformation[R]]))
}
/**
@@ -213,11 +200,8 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
}
val cleanedFunction = clean(function)
- val applyFunction = new JWindowFunction[T, R, K, W] {
- def apply(key: K, window: W, elements: java.lang.Iterable[T], out: Collector[R]): Unit = {
- cleanedFunction(key, window, elements.asScala, out)
- }
- }
+ val applyFunction = new ScalaWindowFunction[T, R, K, W](cleanedFunction)
+
asScalaStream(javaStream.apply(applyFunction, implicitly[TypeInformation[R]]))
}
@@ -236,16 +220,13 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
preAggregator: ReduceFunction[T],
function: WindowFunction[T, R, K, W]): DataStream[R] = {
- val cleanedFunction = clean(function)
+ val cleanedPreAggregator = clean(preAggregator)
+ val cleanedWindowFunction = clean(function)
- val applyFunction = new JWindowFunction[T, R, K, W] {
- def apply(key: K, window: W, elements: java.lang.Iterable[T], out: Collector[R]): Unit = {
- cleanedFunction.apply(key, window, elements.asScala, out)
- }
- }
+ val applyFunction = new ScalaWindowFunctionWrapper[T, R, K, W](cleanedWindowFunction)
val resultType: TypeInformation[R] = implicitly[TypeInformation[R]]
- asScalaStream(javaStream.apply(clean(preAggregator), applyFunction, resultType))
+ asScalaStream(javaStream.apply(cleanedPreAggregator, applyFunction, resultType))
}
/**
@@ -256,31 +237,25 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
* Arriving data is pre-aggregated using the given pre-aggregation reducer.
*
* @param preAggregator The reduce function that is used for pre-aggregation
- * @param function The window function.
+ * @param windowFunction The window function.
* @return The data stream that is the result of applying the window function to the window.
*/
def apply[R: TypeInformation](
preAggregator: (T, T) => T,
- function: (K, W, Iterable[T], Collector[R]) => Unit): DataStream[R] = {
+ windowFunction: (K, W, Iterable[T], Collector[R]) => Unit): DataStream[R] = {
- if (function == null) {
+ if (preAggregator == null) {
throw new NullPointerException("Reduce function must not be null.")
}
- if (function == null) {
+ if (windowFunction == null) {
throw new NullPointerException("WindowApply function must not be null.")
}
val cleanReducer = clean(preAggregator)
- val reducer = new ReduceFunction[T] {
- def reduce(v1: T, v2: T) = { cleanReducer(v1, v2) }
- }
-
- val cleanApply = clean(function)
- val applyFunction = new JWindowFunction[T, R, K, W] {
- def apply(key: K, window: W, input: java.lang.Iterable[T], out: Collector[R]): Unit = {
- cleanApply(key, window, input.asScala, out)
- }
- }
+ val cleanWindowFunction = clean(windowFunction)
+
+ val reducer = new ScalaReduceFunction[T](cleanReducer)
+ val applyFunction = new ScalaWindowFunction[T, R, K, W](cleanWindowFunction)
asScalaStream(javaStream.apply(reducer, applyFunction, implicitly[TypeInformation[R]]))
}
@@ -303,16 +278,13 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
function: WindowFunction[R, R, K, W]): DataStream[R] = {
val cleanedFunction = clean(function)
+ val cleanedFoldFunction = clean(foldFunction)
- val applyFunction = new JWindowFunction[R, R, K, W] {
- def apply(key: K, window: W, elements: java.lang.Iterable[R], out: Collector[R]): Unit = {
- cleanedFunction.apply(key, window, elements.asScala, out)
- }
- }
+ val applyFunction = new ScalaWindowFunctionWrapper[R, R, K, W](cleanedFunction)
asScalaStream(javaStream.apply(
initialValue,
- clean(foldFunction),
+ cleanedFoldFunction,
applyFunction,
implicitly[TypeInformation[R]]))
}
@@ -325,32 +297,27 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K, W]) {
* Arriving data is incrementally aggregated using the given fold function.
*
* @param foldFunction The fold function that is used for incremental aggregation
- * @param function The window function.
+ * @param windowFunction The window function.
* @return The data stream that is the result of applying the window function to the window.
*/
def apply[R: TypeInformation](
initialValue: R,
foldFunction: (R, T) => R,
- function: (K, W, Iterable[R], Collector[R]) => Unit): DataStream[R] = {
+ windowFunction: (K, W, Iterable[R], Collector[R]) => Unit): DataStream[R] = {
- if (function == null) {
+ if (foldFunction == null) {
throw new NullPointerException("Fold function must not be null.")
}
- if (function == null) {
+ if (windowFunction == null) {
throw new NullPointerException("WindowApply function must not be null.")
}
val cleanFolder = clean(foldFunction)
- val folder = new FoldFunction[T, R] {
- def fold(acc: R, v: T) = { cleanFolder(acc, v) }
- }
-
- val cleanApply = clean(function)
- val applyFunction = new JWindowFunction[R, R, K, W] {
- def apply(key: K, window: W, input: java.lang.Iterable[R], out: Collector[R]): Unit = {
- cleanApply(key, window, input.asScala, out)
- }
- }
+ val cleanWindowFunction = clean(windowFunction)
+
+ val folder = new ScalaFoldFunction[T, R](cleanFolder)
+ val applyFunction = new ScalaWindowFunction[R, R, K, W](cleanWindowFunction)
+
val resultType: TypeInformation[R] = implicitly[TypeInformation[R]]
asScalaStream(javaStream.apply(initialValue, folder, applyFunction, resultType))
}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/AllWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/AllWindowFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/AllWindowFunction.scala
index 4e77d83..7677459 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/AllWindowFunction.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/AllWindowFunction.scala
@@ -25,11 +25,12 @@ import org.apache.flink.streaming.api.windowing.windows.Window
import org.apache.flink.util.Collector
/**
- * Base interface for functions that are evaluated over keyed (grouped) windows.
- *
- * @tparam IN The type of the input value.
- * @tparam OUT The type of the output value.
- */
+ * Base interface for functions that are evaluated over non-grouped windows,
+ * i.e., windows over all stream partitions.
+ *
+ * @tparam IN The type of the input value.
+ * @tparam OUT The type of the output value.
+ */
@Public
trait AllWindowFunction[IN, OUT, W <: Window] extends Function with Serializable {
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichAllWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichAllWindowFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichAllWindowFunction.scala
new file mode 100644
index 0000000..4ff1fe9
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichAllWindowFunction.scala
@@ -0,0 +1,37 @@
+/*
+ * 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.scala.function
+
+import org.apache.flink.api.common.functions.AbstractRichFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+
+/**
+ * Rich variant of the [[org.apache.flink.streaming.api.scala.function.AllWindowFunction]].
+ *
+ * As a [[org.apache.flink.api.common.functions.RichFunction]], it gives access to the
+ * [[org.apache.flink.api.common.functions.RuntimeContext]] and provides setup
+ * and tear-down methods.
+ *
+ * @tparam IN The type of the input value.
+ * @tparam OUT The type of the output value.
+ * @tparam W The type of Window that this window function can be applied on.
+ */
+abstract class RichAllWindowFunction[IN, OUT, W <: Window]
+ extends AbstractRichFunction
+ with AllWindowFunction[IN, OUT, W] {}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichWindowFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichWindowFunction.scala
new file mode 100644
index 0000000..72d44e7
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/RichWindowFunction.scala
@@ -0,0 +1,38 @@
+/*
+ * 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.scala.function
+
+import org.apache.flink.api.common.functions.AbstractRichFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+
+/**
+ * Rich variant of the [[org.apache.flink.streaming.api.scala.function.WindowFunction]].
+ *
+ * As a [[org.apache.flink.api.common.functions.RichFunction]], it gives access to the
+ * [[org.apache.flink.api.common.functions.RuntimeContext]] and provides setup
+ * and tear-down methods.
+ *
+ * @tparam IN The type of the input value.
+ * @tparam OUT The type of the output value.
+ * @tparam KEY The type of the key.
+ * @tparam W The type of Window that this window function can be applied on.
+ */
+abstract class RichWindowFunction[IN, OUT, KEY, W <: Window]
+ extends AbstractRichFunction
+ with WindowFunction[IN, OUT, KEY, W] {}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunction.scala
new file mode 100644
index 0000000..943e83f
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunction.scala
@@ -0,0 +1,38 @@
+/*
+ * 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.scala.function.util
+
+import org.apache.flink.streaming.api.functions.windowing.{AllWindowFunction => JAllWindowFunction}
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+
+import scala.collection.JavaConverters._
+
+/**
+ * A wrapper function that exposes a Scala Function3 as a Java AllWindowFunction.
+ */
+final class ScalaAllWindowFunction[IN, OUT, W <: Window](
+ private[this] val function: (W, Iterable[IN], Collector[OUT]) => Unit)
+ extends JAllWindowFunction[IN, OUT, W] {
+
+ @throws(classOf[Exception])
+ override def apply(window: W, input: java.lang.Iterable[IN], out: Collector[OUT]) {
+ function.apply(window, input.asScala, out)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunctionWrapper.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunctionWrapper.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunctionWrapper.scala
new file mode 100644
index 0000000..39142c2
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaAllWindowFunctionWrapper.scala
@@ -0,0 +1,68 @@
+/*
+ * 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.scala.function.util
+
+import org.apache.flink.api.common.functions.util.FunctionUtils
+import org.apache.flink.api.common.functions.{IterationRuntimeContext, RichFunction, RuntimeContext}
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.{AllWindowFunction => JAllWindowFunction}
+import org.apache.flink.streaming.api.scala.function.AllWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+
+import scala.collection.JavaConverters._
+
+/**
+ * A wrapper function that exposes a Scala WindowFunction as a JavaWindow function.
+ *
+ * The Scala and Java Window functions differ in their type of "Iterable":
+ * - Scala WindowFunction: scala.Iterable
+ * - Java WindowFunction: java.lang.Iterable
+ */
+final class ScalaAllWindowFunctionWrapper[IN, OUT, W <: Window](
+ private[this] val func: AllWindowFunction[IN, OUT, W])
+ extends JAllWindowFunction[IN, OUT, W] with RichFunction {
+
+ @throws(classOf[Exception])
+ override def apply(window: W, input: java.lang.Iterable[IN], out: Collector[OUT]) {
+ func.apply(window, input.asScala, out)
+ }
+
+ @throws(classOf[Exception])
+ override def open(parameters: Configuration) {
+ FunctionUtils.openFunction(func, parameters)
+ }
+
+ @throws(classOf[Exception])
+ override def close() {
+ FunctionUtils.closeFunction(func)
+ }
+
+ override def setRuntimeContext(t: RuntimeContext) {
+ FunctionUtils.setFunctionRuntimeContext(func, t)
+ }
+
+ override def getRuntimeContext(): RuntimeContext = {
+ throw new RuntimeException("This should never be called")
+ }
+
+ override def getIterationRuntimeContext(): IterationRuntimeContext = {
+ throw new RuntimeException("This should never be called")
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaFoldFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaFoldFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaFoldFunction.scala
new file mode 100644
index 0000000..f85ddd0
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaFoldFunction.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.scala.function.util
+
+import org.apache.flink.api.common.functions.FoldFunction
+
+/**
+ * A wrapper function that exposes a Scala Function2 as a [[FoldFunction]].
+ */
+final class ScalaFoldFunction[T, R](private[this] val foldFunction: (R, T) => R)
+ extends FoldFunction[T, R] {
+
+ @throws(classOf[Exception])
+ override def fold(accumulator: R, value: T): R = {
+ foldFunction(accumulator, value)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaReduceFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaReduceFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaReduceFunction.scala
new file mode 100644
index 0000000..027ccb2
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaReduceFunction.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.scala.function.util
+
+import org.apache.flink.api.common.functions.ReduceFunction
+
+/**
+ * A wrapper function that exposes a Scala Function2 as a [[ReduceFunction]].
+ */
+final class ScalaReduceFunction[T](private[this] val function: (T, T) => T)
+ extends ReduceFunction[T] {
+
+ @throws(classOf[Exception])
+ override def reduce(a: T, b: T): T = {
+ function(a, b)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunction.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunction.scala
new file mode 100644
index 0000000..3b4cb9d
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunction.scala
@@ -0,0 +1,38 @@
+/*
+ * 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.scala.function.util
+
+import org.apache.flink.streaming.api.functions.windowing.{WindowFunction => JWindowFunction}
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+
+import scala.collection.JavaConverters._
+
+/**
+ * A wrapper function that exposes a Scala Function4 as a Java WindowFunction.
+ */
+final class ScalaWindowFunction[IN, OUT, KEY, W <: Window](
+ private[this] val function: (KEY, W, Iterable[IN], Collector[OUT]) => Unit)
+ extends JWindowFunction[IN, OUT, KEY, W] {
+
+ @throws(classOf[Exception])
+ override def apply(key: KEY, window: W, input: java.lang.Iterable[IN], out: Collector[OUT]) {
+ function.apply(key, window, input.asScala, out)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunctionWrapper.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunctionWrapper.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunctionWrapper.scala
new file mode 100644
index 0000000..1d74b6c
--- /dev/null
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/function/util/ScalaWindowFunctionWrapper.scala
@@ -0,0 +1,68 @@
+/*
+ * 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.scala.function.util
+
+import org.apache.flink.api.common.functions.{IterationRuntimeContext, RuntimeContext, RichFunction}
+import org.apache.flink.api.common.functions.util.FunctionUtils
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.{ WindowFunction => JWindowFunction }
+import org.apache.flink.streaming.api.scala.function.WindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+
+import scala.collection.JavaConverters._
+
+/**
+ * A wrapper function that exposes a Scala WindowFunction as a JavaWindow function.
+ *
+ * The Scala and Java Window functions differ in their type of "Iterable":
+ * - Scala WindowFunction: scala.Iterable
+ * - Java WindowFunction: java.lang.Iterable
+ */
+final class ScalaWindowFunctionWrapper[IN, OUT, KEY, W <: Window](
+ private[this] val func: WindowFunction[IN, OUT, KEY, W])
+ extends JWindowFunction[IN, OUT, KEY, W] with RichFunction {
+
+ @throws(classOf[Exception])
+ override def apply(key: KEY, window: W, input: java.lang.Iterable[IN], out: Collector[OUT]) {
+ func.apply(key, window, input.asScala, out)
+ }
+
+ @throws(classOf[Exception])
+ override def open(parameters: Configuration) {
+ FunctionUtils.openFunction(func, parameters)
+ }
+
+ @throws(classOf[Exception])
+ override def close() {
+ FunctionUtils.closeFunction(func)
+ }
+
+ override def setRuntimeContext(t: RuntimeContext) {
+ FunctionUtils.setFunctionRuntimeContext(func, t)
+ }
+
+ override def getRuntimeContext(): RuntimeContext = {
+ throw new RuntimeException("This should never be called")
+ }
+
+ override def getIterationRuntimeContext(): IterationRuntimeContext = {
+ throw new RuntimeException("This should never be called")
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
index 7833651..6a6a956 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFoldITCase.scala
@@ -20,13 +20,17 @@ package org.apache.flink.streaming.api.scala
import java.util.concurrent.TimeUnit
+import org.apache.flink.api.common.functions.FoldFunction
+import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
-import org.apache.flink.streaming.api.functions.{AssignerWithPunctuatedWatermarks}
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction
+import org.apache.flink.streaming.api.scala.testutils.{CheckingIdentityRichWindowFunction, CheckingIdentityRichAllWindowFunction}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase
import org.junit.Test
import org.junit.Assert._
@@ -87,9 +91,68 @@ class WindowFoldITCase extends StreamingMultipleProgramsTestBase {
}
@Test
- def testFoldAllWindow(): Unit = {
+ def testFoldWithWindowFunction(): Unit = {
WindowFoldITCase.testResults = mutable.MutableList()
+ CheckingIdentityRichWindowFunction.reset()
+ val foldFunc = new FoldFunction[(String, Int), (String, Int)] {
+ override def fold(accumulator: (String, Int), value: (String, Int)): (String, Int) = {
+ (accumulator._1 + value._1, accumulator._2 + value._2)
+ }
+ }
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 6))
+ ctx.collect(("a", 7))
+ ctx.collect(("a", 8))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {
+ }
+ }).assignTimestampsAndWatermarks(new WindowFoldITCase.Tuple2TimestampExtractor)
+
+ source1
+ .keyBy(0)
+ .window(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .apply(
+ ("R:", 0),
+ foldFunc,
+ new CheckingIdentityRichWindowFunction[(String, Int), Tuple, TimeWindow]())
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowFoldITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("Fold Window Test")
+
+ val expectedResult = mutable.MutableList(
+ "(R:aaa,3)",
+ "(R:aaa,21)",
+ "(R:bbb,12)")
+
+ assertEquals(expectedResult.sorted, WindowFoldITCase.testResults.sorted)
+
+ CheckingIdentityRichWindowFunction.checkRichMethodCalls()
+ }
+
+ @Test
+ def testFoldAllWindow(): Unit = {
+ WindowFoldITCase.testResults = mutable.MutableList()
+
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.setParallelism(1)
@@ -131,6 +194,62 @@ class WindowFoldITCase extends StreamingMultipleProgramsTestBase {
assertEquals(expectedResult.sorted, WindowFoldITCase.testResults.sorted)
}
+ @Test
+ def testFoldAllWithWindowFunction(): Unit = {
+ WindowFoldITCase.testResults = mutable.MutableList()
+ CheckingIdentityRichAllWindowFunction.reset()
+
+ val foldFunc = new FoldFunction[(String, Int), (String, Int)] {
+ override def fold(accumulator: (String, Int), value: (String, Int)): (String, Int) = {
+ (accumulator._1 + value._1, accumulator._2 + value._2)
+ }
+ }
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("a", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("a", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 5))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {
+ }
+ }).assignTimestampsAndWatermarks(new WindowFoldITCase.Tuple2TimestampExtractor)
+
+ source1
+ .windowAll(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .apply(
+ ("R:", 0),
+ foldFunc,
+ new CheckingIdentityRichAllWindowFunction[(String, Int), TimeWindow]())
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowFoldITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("Fold All-Window Test")
+
+ val expectedResult = mutable.MutableList(
+ "(R:aaa,3)",
+ "(R:bababa,24)")
+
+ assertEquals(expectedResult.sorted, WindowFoldITCase.testResults.sorted)
+
+ CheckingIdentityRichAllWindowFunction.checkRichMethodCalls()
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFunctionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFunctionITCase.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFunctionITCase.scala
new file mode 100644
index 0000000..c38f422
--- /dev/null
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowFunctionITCase.scala
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.scala
+
+import java.util.concurrent.TimeUnit
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.streaming.api.TimeCharacteristic
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
+import org.apache.flink.streaming.api.functions.sink.SinkFunction
+import org.apache.flink.streaming.api.functions.source.SourceFunction
+import org.apache.flink.streaming.api.scala.testutils.{CheckingIdentityRichAllWindowFunction, CheckingIdentityRichWindowFunction}
+import org.apache.flink.streaming.api.watermark.Watermark
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
+import org.apache.flink.streaming.api.windowing.time.Time
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow
+import org.junit.Assert._
+import org.junit.Test
+
+import scala.collection.mutable
+
+class WindowFunctionITCase {
+
+ @Test
+ def testRichWindowFunction(): Unit = {
+ WindowFunctionITCase.testResults = mutable.MutableList()
+ CheckingIdentityRichWindowFunction.reset()
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 6))
+ ctx.collect(("a", 7))
+ ctx.collect(("a", 8))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {}
+
+ }).assignTimestampsAndWatermarks(new WindowFunctionITCase.Tuple2TimestampExtractor)
+
+ source1
+ .keyBy(0)
+ .window(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .apply(new CheckingIdentityRichWindowFunction[(String, Int), Tuple, TimeWindow]())
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowFunctionITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("RichWindowFunction Test")
+
+ val expectedResult = mutable.MutableList(
+ "(a,0)", "(a,1)", "(a,2)", "(a,6)", "(a,7)", "(a,8)",
+ "(b,3)", "(b,4)", "(b,5)")
+
+ assertEquals(expectedResult.sorted, WindowFunctionITCase.testResults.sorted)
+
+ CheckingIdentityRichWindowFunction.checkRichMethodCalls()
+ }
+
+ @Test
+ def testRichAllWindowFunction(): Unit = {
+ WindowFunctionITCase.testResults = mutable.MutableList()
+ CheckingIdentityRichAllWindowFunction.reset()
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 6))
+ ctx.collect(("a", 7))
+ ctx.collect(("a", 8))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {}
+
+ }).assignTimestampsAndWatermarks(new WindowFunctionITCase.Tuple2TimestampExtractor)
+
+ source1
+ .windowAll(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .apply(new CheckingIdentityRichAllWindowFunction[(String, Int), TimeWindow]())
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowFunctionITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("RichAllWindowFunction Test")
+
+ val expectedResult = mutable.MutableList(
+ "(a,0)", "(a,1)", "(a,2)", "(a,6)", "(a,7)", "(a,8)",
+ "(b,3)", "(b,4)", "(b,5)")
+
+ assertEquals(expectedResult.sorted, WindowFunctionITCase.testResults.sorted)
+
+ CheckingIdentityRichAllWindowFunction.checkRichMethodCalls()
+ }
+}
+
+object WindowFunctionITCase {
+
+ private var testResults: mutable.MutableList[String] = null
+
+ private class Tuple2TimestampExtractor extends AssignerWithPunctuatedWatermarks[(String, Int)] {
+
+ private var currentTimestamp = -1L
+
+ override def extractTimestamp(element: (String, Int), previousTimestamp: Long): Long = {
+ currentTimestamp = element._2
+ currentTimestamp
+ }
+
+ def checkAndGetNextWatermark(
+ lastElement: (String, Int),
+ extractedTimestamp: Long): Watermark = {
+ new Watermark(lastElement._2 - 1)
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowReduceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowReduceITCase.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowReduceITCase.scala
new file mode 100644
index 0000000..ffd94fc
--- /dev/null
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/WindowReduceITCase.scala
@@ -0,0 +1,276 @@
+/*
+ * 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.scala
+
+import java.util.concurrent.TimeUnit
+
+import org.apache.flink.api.common.functions.{ReduceFunction, FoldFunction}
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.streaming.api.TimeCharacteristic
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
+import org.apache.flink.streaming.api.functions.sink.SinkFunction
+import org.apache.flink.streaming.api.functions.source.SourceFunction
+import org.apache.flink.streaming.api.scala.testutils.{CheckingIdentityRichAllWindowFunction, CheckingIdentityRichWindowFunction}
+import org.apache.flink.streaming.api.watermark.Watermark
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
+import org.apache.flink.streaming.api.windowing.time.Time
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow
+import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase
+
+import org.junit.Assert._
+import org.junit.Test
+
+import scala.collection.mutable
+
+/**
+ * Tests for Folds over windows. These also test whether OutputTypeConfigurable functions
+ * work for windows, because FoldWindowFunction is OutputTypeConfigurable.
+ */
+class WindowReduceITCase extends StreamingMultipleProgramsTestBase {
+
+ @Test
+ def testReduceWindow(): Unit = {
+ WindowReduceITCase.testResults = mutable.MutableList()
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 6))
+ ctx.collect(("a", 7))
+ ctx.collect(("a", 8))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {
+ }
+ }).assignTimestampsAndWatermarks(new WindowReduceITCase.Tuple2TimestampExtractor)
+
+ source1
+ .keyBy(0)
+ .window(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .reduce( (a, b) => (a._1 + b._1, a._2 + b._2) )
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowReduceITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("Reduce Window Test")
+
+ val expectedResult = mutable.MutableList(
+ "(aaa,3)",
+ "(aaa,21)",
+ "(bbb,12)")
+
+ assertEquals(expectedResult.sorted, WindowReduceITCase.testResults.sorted)
+ }
+
+ @Test
+ def testReduceWithWindowFunction(): Unit = {
+ WindowReduceITCase.testResults = mutable.MutableList()
+ CheckingIdentityRichWindowFunction.reset()
+
+ val reduceFunc = new ReduceFunction[(String, Int)] {
+ override def reduce(a: (String, Int), b: (String, Int)): (String, Int) = {
+ (a._1 + b._1, a._2 + b._2)
+ }
+ }
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 6))
+ ctx.collect(("a", 7))
+ ctx.collect(("a", 8))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {
+ }
+ }).assignTimestampsAndWatermarks(new WindowReduceITCase.Tuple2TimestampExtractor)
+
+ source1
+ .keyBy(0)
+ .window(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .apply(
+ reduceFunc,
+ new CheckingIdentityRichWindowFunction[(String, Int), Tuple, TimeWindow]())
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowReduceITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("Reduce Window Test")
+
+ val expectedResult = mutable.MutableList(
+ "(aaa,3)",
+ "(aaa,21)",
+ "(bbb,12)")
+
+ assertEquals(expectedResult.sorted, WindowReduceITCase.testResults.sorted)
+
+ CheckingIdentityRichWindowFunction.checkRichMethodCalls()
+ }
+
+ @Test
+ def testReduceAllWindow(): Unit = {
+ WindowReduceITCase.testResults = mutable.MutableList()
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("a", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("a", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 5))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {
+ }
+ }).assignTimestampsAndWatermarks(new WindowReduceITCase.Tuple2TimestampExtractor)
+
+ source1
+ .windowAll(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .reduce( (a, b) => (a._1 + b._1, a._2 + b._2) )
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowReduceITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("Fold All-Window Test")
+
+ val expectedResult = mutable.MutableList(
+ "(aaa,3)",
+ "(bababa,24)")
+
+ assertEquals(expectedResult.sorted, WindowReduceITCase.testResults.sorted)
+ }
+
+ @Test
+ def testReduceAllWithWindowFunction(): Unit = {
+ WindowReduceITCase.testResults = mutable.MutableList()
+ CheckingIdentityRichAllWindowFunction.reset()
+
+ val reduceFunc = new ReduceFunction[(String, Int)] {
+ override def reduce(a: (String, Int), b: (String, Int)): (String, Int) = {
+ (a._1 + b._1, a._2 + b._2)
+ }
+ }
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ env.setParallelism(1)
+
+ val source1 = env.addSource(new SourceFunction[(String, Int)]() {
+ def run(ctx: SourceFunction.SourceContext[(String, Int)]) {
+ ctx.collect(("a", 0))
+ ctx.collect(("a", 1))
+ ctx.collect(("a", 2))
+ ctx.collect(("b", 3))
+ ctx.collect(("a", 3))
+ ctx.collect(("b", 4))
+ ctx.collect(("a", 4))
+ ctx.collect(("b", 5))
+ ctx.collect(("a", 5))
+
+ // source is finite, so it will have an implicit MAX watermark when it finishes
+ }
+
+ def cancel() {
+ }
+ }).assignTimestampsAndWatermarks(new WindowReduceITCase.Tuple2TimestampExtractor)
+
+ source1
+ .windowAll(TumblingEventTimeWindows.of(Time.of(3, TimeUnit.MILLISECONDS)))
+ .apply(
+ reduceFunc,
+ new CheckingIdentityRichAllWindowFunction[(String, Int), TimeWindow]())
+ .addSink(new SinkFunction[(String, Int)]() {
+ def invoke(value: (String, Int)) {
+ WindowReduceITCase.testResults += value.toString
+ }
+ })
+
+ env.execute("Fold All-Window Test")
+
+ val expectedResult = mutable.MutableList(
+ "(aaa,3)",
+ "(bababa,24)")
+
+ assertEquals(expectedResult.sorted, WindowReduceITCase.testResults.sorted)
+
+ CheckingIdentityRichAllWindowFunction.checkRichMethodCalls()
+ }
+}
+
+object WindowReduceITCase {
+
+ private var testResults: mutable.MutableList[String] = null
+
+ private class Tuple2TimestampExtractor extends AssignerWithPunctuatedWatermarks[(String, Int)] {
+
+ private var currentTimestamp = -1L
+
+ override def extractTimestamp(element: (String, Int), previousTimestamp: Long): Long = {
+ currentTimestamp = element._2
+ currentTimestamp
+ }
+
+ def checkAndGetNextWatermark(
+ lastElement: (String, Int),
+ extractedTimestamp: Long): Watermark = {
+ new Watermark(lastElement._2 - 1)
+ }
+ }
+}
+
+
+
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichAllWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichAllWindowFunction.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichAllWindowFunction.scala
new file mode 100644
index 0000000..acd8b12
--- /dev/null
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichAllWindowFunction.scala
@@ -0,0 +1,84 @@
+/*
+ * 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.scala.testutils
+
+import org.apache.flink.api.common.functions.RuntimeContext
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.scala.function.RichAllWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+
+
+class CheckingIdentityRichAllWindowFunction[T, W <: Window]
+ extends RichAllWindowFunction[T, T, W] {
+
+ override def apply(window: W, input: scala.Iterable[T], out: Collector[T]): Unit = {
+ for (value <- input) {
+ out.collect(value)
+ }
+ }
+
+ override def open(conf: Configuration): Unit = {
+ super.open(conf)
+ CheckingIdentityRichAllWindowFunction.openCalled = true
+ }
+
+ override def close(): Unit = {
+ super.close()
+ CheckingIdentityRichAllWindowFunction.closeCalled = true
+ }
+
+ override def setRuntimeContext(context: RuntimeContext): Unit = {
+ super.setRuntimeContext(context)
+ CheckingIdentityRichAllWindowFunction.contextSet = true
+ }
+}
+
+
+object CheckingIdentityRichAllWindowFunction {
+
+ @volatile
+ private[CheckingIdentityRichAllWindowFunction] var closeCalled = false
+
+ @volatile
+ private[CheckingIdentityRichAllWindowFunction] var openCalled = false
+
+ @volatile
+ private[CheckingIdentityRichAllWindowFunction] var contextSet = false
+
+ def reset(): Unit = {
+ closeCalled = false
+ openCalled = false
+ contextSet = false
+ }
+
+ def checkRichMethodCalls(): Unit = {
+ if (!contextSet) {
+ throw new AssertionError("context not set")
+ }
+ if (!openCalled) {
+ throw new AssertionError("open() not called")
+ }
+ if (!closeCalled) {
+ throw new AssertionError("close() not called")
+ }
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/flink/blob/91587892/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichWindowFunction.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichWindowFunction.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichWindowFunction.scala
new file mode 100644
index 0000000..e0de0de
--- /dev/null
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/testutils/CheckingIdentityRichWindowFunction.scala
@@ -0,0 +1,81 @@
+/*
+ * 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.scala.testutils
+
+import org.apache.flink.api.common.functions.RuntimeContext
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.scala.function.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+
+
+class CheckingIdentityRichWindowFunction[T, K, W <: Window]
+ extends RichWindowFunction[T, T, K, W] {
+
+ override def apply(key: K, window: W, input: scala.Iterable[T], out: Collector[T]): Unit = {
+ for (value <- input) {
+ out.collect(value)
+ }
+ }
+
+ override def open(conf: Configuration): Unit = {
+ super.open(conf)
+ CheckingIdentityRichWindowFunction.openCalled = true
+ }
+
+ override def close(): Unit = {
+ super.close()
+ CheckingIdentityRichWindowFunction.closeCalled = true
+ }
+
+ override def setRuntimeContext(context: RuntimeContext): Unit = {
+ super.setRuntimeContext(context)
+ CheckingIdentityRichWindowFunction.contextSet = true
+ }
+}
+
+object CheckingIdentityRichWindowFunction {
+
+ @volatile
+ private[CheckingIdentityRichWindowFunction] var closeCalled = false
+
+ @volatile
+ private[CheckingIdentityRichWindowFunction] var openCalled = false
+
+ @volatile
+ private[CheckingIdentityRichWindowFunction] var contextSet = false
+
+ def reset(): Unit = {
+ closeCalled = false
+ openCalled = false
+ contextSet = false
+ }
+
+ def checkRichMethodCalls(): Unit = {
+ if (!contextSet) {
+ throw new AssertionError("context not set")
+ }
+ if (!openCalled) {
+ throw new AssertionError("open() not called")
+ }
+ if (!closeCalled) {
+ throw new AssertionError("close() not called")
+ }
+ }
+}
[4/6] flink git commit: [FLINK-3575] [docs] Update 'Working With
State' section in docs
Posted by se...@apache.org.
[FLINK-3575] [docs] Update 'Working With State' section in docs
This closes #1760
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5c3c7345
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5c3c7345
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5c3c7345
Branch: refs/heads/release-1.0
Commit: 5c3c7345d1711989883a787a62ef91a72bf4c4f7
Parents: 91587892
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Mar 3 16:53:53 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Mar 4 20:53:44 2016 +0100
----------------------------------------------------------------------
docs/apis/streaming/state.md | 219 +++++++++++++++++++++++++++-----------
1 file changed, 157 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5c3c7345/docs/apis/streaming/state.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/state.md b/docs/apis/streaming/state.md
index df68a1b..f32d504 100644
--- a/docs/apis/streaming/state.md
+++ b/docs/apis/streaming/state.md
@@ -26,7 +26,8 @@ under the License.
All transformations in Flink may look like functions (in the functional processing terminology), but
are in fact stateful operators. You can make *every* transformation (`map`, `filter`, etc) stateful
-by declaring local variables or using Flink's state interface. You can register any local variable
+by using Flink's state interface or checkpointing instance fields of your function. You can register
+any instance field
as ***managed*** state by implementing an interface. In this case, and also in the case of using
Flink's native state interface, Flink will automatically take consistent snapshots of your state
periodically, and restore its value in the case of a failure.
@@ -34,102 +35,130 @@ periodically, and restore its value in the case of a failure.
The end effect is that updates to any form of state are the same under failure-free execution and
execution under failures.
-First, we look at how to make local variables consistent under failures, and then we look at
+First, we look at how to make instance fields consistent under failures, and then we look at
Flink's state interface.
By default state checkpoints will be stored in-memory at the JobManager. For proper persistence of large
state, Flink supports storing the checkpoints on file systems (HDFS, S3, or any mounted POSIX file system),
which can be configured in the `flink-conf.yaml` or via `StreamExecutionEnvironment.setStateBackend(…)`.
+See [state backends]({{ site.baseurl }}/apis/streaming/state_backends.html) for information
+about the available state backends and how to configure them.
* ToC
{:toc}
+## Using the Key/Value State Interface
-## Checkpointing Local Variables
+The Key/Value state interface provides access to different types of state that are all scoped to
+the key of the current input element. This means that this type of state can only be used
+on a `KeyedStream`, which can be created via `stream.keyBy(…)`.
-Local variables can be checkpointed by using the `Checkpointed` interface.
+Now, we will first look at the different types of state available and then we will see
+how they can be used in a program. The available state primitives are:
-When the user-defined function implements the `Checkpointed` interface, the `snapshotState(…)` and `restoreState(…)`
-methods will be executed to draw and restore function state.
+* `ValueState<T>`: This keeps a value that can be updated and
+retrieved (scoped to key of the input element, mentioned above, so there will possibly be one value
+for each key that the operation sees). The value can be set using `update(T)` and retrieved using
+`T value()`.
-In addition to that, user functions can also implement the `CheckpointNotifier` interface to receive notifications on
-completed checkpoints via the `notifyCheckpointComplete(long checkpointId)` method.
-Note that there is no guarantee for the user function to receive a notification if a failure happens between
-checkpoint completion and notification. The notifications should hence be treated in a way that notifications from
-later checkpoints can subsume missing notifications.
+* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `Iterable`
+over all currently stored elements. Elements are added using `add(T)`, the Iterable can
+be retrieved using `Iterable<T> get()`.
-For example the same counting, reduce function shown for `OperatorState`s by using the `Checkpointed` interface instead:
+* `ReducingState<T>`: This keeps a single value that represents the aggregation of all values
+added to the state. The interface is the same as for `ListState` but elements added using
+`add(T)` are reduced to an aggregate using a specified `ReduceFunction`.
-{% highlight java %}
-public class CounterSum extends ReduceFunction<Long>, Checkpointed<Long> {
+All types of state also have a method `clear()` that clears the state for the currently
+active key (i.e. the key of the input element).
- // persistent counter
- private long counter = 0;
+It is important to keep in mind that these state objects are only used for interfacing
+with state. The state is not necessarily stored inside but might reside on disk or somewhere else.
+The second thing to keep in mind is that the value you get from the state
+depend on the key of the input element. So the value you get in one invocation of your
+user function can be different from the one you get in another invocation if the key of
+the element is different.
- @Override
- public Long reduce(Long value1, Long value2) {
- counter++;
- return value1 + value2;
- }
+To get a state handle you have to create a `StateDescriptor` this holds the name of the state
+(as we will later see you can create several states, and they have to have unique names so
+that you can reference them), the type of the values that the state holds and possibly
+a user-specified function, such as a `ReduceFunction`. Depending on what type of state you
+want to retrieve you create one of `ValueStateDescriptor`, `ListStateDescriptor` or
+`ReducingStateDescriptor`.
- // regularly persists state during normal operation
- @Override
- public Serializable snapshotState(long checkpointId, long checkpointTimestamp) {
- return counter;
- }
+State is accessed using the `RuntimeContext`, so it is only possible in *rich functions*.
+Please see [here]({{ site.baseurl }}/apis/common/#specifying-transformation-functions) for
+information about that but we will also see an example shortly. The `RuntimeContext` that
+is available in a `RichFunction` has these methods for accessing state:
- // restores state on recovery from failure
- @Override
- public void restoreState(Long state) {
- counter = state;
- }
-}
-{% endhighlight %}
+* `ValueState<T> getState(ValueStateDescriptor<T>)`
+* `ReducingState<T> getReducingState(ReducingStateDescriptor<T>)`
+* `ListState<T> getListState(ListStateDescriptor<T>)`
-## Using the Key/Value State Interface
+This is an example `FlatMapFunction` that shows how all of the parts fit together:
-The state interface gives access to key/value states, which are a collection of key/value pairs.
-Because the state is partitioned by the keys (distributed accross workers), it can only be used
-on the `KeyedStream`, created via `stream.keyBy(…)` (which means also that it is usable in all
-types of functions on keyed windows).
+{% highlight java %}
+public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
-The handle to the state can be obtained from the function's `RuntimeContext`. The state handle will
-then give access to the value mapped under the key of the current record or window - each key consequently
-has its own value.
+ /**
+ * The ValueState handle. The first field is the count, the second field a running sum.
+ */
+ private transient ValueState<Tuple2<Long, Long>> sum;
-The following code sample shows how to use the key/value state inside a reduce function.
-When creating the state handle, one needs to supply a name for that state (a function can have multiple states
-of different types), the type of the state (used to create efficient serializers), and the default value (returned
-as a value for keys that do not yet have a value associated).
+ @Override
+ public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
-{% highlight java %}
-public class CounterSum extends RichReduceFunction<Long> {
+ // access the state value
+ Tuple2<Long, Long> currentSum = sum.value();
- /** The state handle */
- private OperatorState<Long> counter;
+ // update the count
+ currentSum.f0 += 1;
- @Override
- public Long reduce(Long value1, Long value2) {
- counter.update(counter.value() + 1);
- return value1 + value2;
+ // add the second field of the input value
+ currentSum.f1 += input.f1;
+
+ // update the state
+ sum.update(currentSum);
+
+ // if the count reaches 2, emit the average and clear the state
+ if (currentSum.f0 >= 2) {
+ out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
+ sum.clear();
+ }
}
@Override
public void open(Configuration config) {
- counter = getRuntimeContext().getKeyValueState("myCounter", Long.class, 0L);
+ ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
+ new ValueStateDescriptor<>(
+ "average", // the state name
+ TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // type information
+ Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
+ sum = getRuntimeContext().getState(descriptor);
}
}
+
+// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)
+env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
+ .keyBy(0)
+ .flatMap(new CountWindowAverage())
+ .print();
+
+// the printed output will be (1,4) and (1,5)
{% endhighlight %}
-State updated by this is usually kept locally inside the flink process (unless one configures explicitly
-an external state backend). This means that lookups and updates are process local and this very fast.
+This example implements a poor man's counting window. We key the tuples by the first field
+(in the example all have the same key `1`). The function stores the count and a running sum in
+a `ValueState`, once the count reaches 2 it will emit the average and clear the state so that
+we start over from `0`. Note that this would keep a different state value for each different input
+key if we had tuples with different values in the first field.
-The important implication of having the keys set implicitly is that it forces programs to group the stream
-by key (via the `keyBy()` function), making the key partitioning transparent to Flink. That allows the system
-to efficiently restore and redistribute keys and state.
+### State in the Scala DataStream API
-The Scala API has shortcuts that for stateful `map()` or `flatMap()` functions on `KeyedStream`, which give the
-state of the current key as an option directly into the function, and return the result with a state update:
+In addition to the interface described above, the Scala API has shortcuts for stateful
+`map()` or `flatMap()` functions with a single `ValueState` on `KeyedStream`. The user function
+gets the current value of the `ValueState` in an `Option` and must return an updated value that
+will be used to update the state.
{% highlight scala %}
val stream: DataStream[(String, Int)] = ...
@@ -143,6 +172,70 @@ val counts: DataStream[(String, Int)] = stream
})
{% endhighlight %}
+## Checkpointing Instance Fields
+
+Instance fields can be checkpointed by using the `Checkpointed` interface.
+
+When the user-defined function implements the `Checkpointed` interface, the `snapshotState(…)` and `restoreState(…)`
+methods will be executed to draw and restore function state.
+
+In addition to that, user functions can also implement the `CheckpointNotifier` interface to receive notifications on
+completed checkpoints via the `notifyCheckpointComplete(long checkpointId)` method.
+Note that there is no guarantee for the user function to receive a notification if a failure happens between
+checkpoint completion and notification. The notifications should hence be treated in a way that notifications from
+later checkpoints can subsume missing notifications.
+
+The above example for `ValueState` can be implemented using instance fields like this:
+
+{% highlight java %}
+
+public class CountWindowAverage
+ extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
+ implements Checkpointed<Tuple2<Long, Long>> {
+
+ private Tuple2<Long, Long> sum = null;
+
+ @Override
+ public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
+
+ // update the count
+ sum.f0 += 1;
+
+ // add the second field of the input value
+ sum.f1 += input.f1;
+
+
+ // if the count reaches 2, emit the average and clear the state
+ if (sum.f0 >= 2) {
+ out.collect(new Tuple2<>(input.f0, sum.f1 / sum.f0));
+ sum = Tuple2.of(0L, 0L);
+ }
+ }
+
+ @Override
+ public void open(Configuration config) {
+ if (sum == null) {
+ // only recreate if null
+ // restoreState will be called before open()
+ // so this will already set the sum to the restored value
+ sum = Tuple2.of(0L, 0L);
+ }
+ }
+
+ // regularly persists state during normal operation
+ @Override
+ public Serializable snapshotState(long checkpointId, long checkpointTimestamp) {
+ return sum;
+ }
+
+ // restores state on recovery from failure
+ @Override
+ public void restoreState(Tuple2<Long, Long> state) {
+ sum = state;
+ }
+}
+{% endhighlight %}
+
## Stateful Source Functions
Stateful sources require a bit more care as opposed to other operators.
@@ -150,7 +243,9 @@ In order to make the updates to the state and output collection atomic (required
on failure/recovery), the user is required to get a lock from the source's context.
{% highlight java %}
-public static class CounterSource extends RichParallelSourceFunction<Long>, Checkpointed<Long> {
+public static class CounterSource
+ extends RichParallelSourceFunction<Long>
+ implements Checkpointed<Long> {
/** current offset for exactly once semantics */
private long offset;
[6/6] flink git commit: [docs] Add documentation about Event Time,
Timestamp Assigners, Watermark Generation
Posted by se...@apache.org.
[docs] Add documentation about Event Time, Timestamp Assigners, Watermark Generation
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0e61ffc5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0e61ffc5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0e61ffc5
Branch: refs/heads/release-1.0
Commit: 0e61ffc54deb6dfcb34aa80e4c792518e2c2305b
Parents: 5c3c734
Author: Stephan Ewen <se...@apache.org>
Authored: Thu Mar 3 21:21:05 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Mar 4 20:53:44 2016 +0100
----------------------------------------------------------------------
docs/apis/streaming/event_time.md | 148 ++++++++
.../streaming/event_timestamps_watermarks.md | 358 ++++++++++++++++++
docs/apis/streaming/fig/times_clocks.svg | 368 +++++++++++++++++++
docs/apis/streaming/windows.md | 2 +-
4 files changed, 875 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0e61ffc5/docs/apis/streaming/event_time.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/event_time.md b/docs/apis/streaming/event_time.md
new file mode 100644
index 0000000..ad01f04
--- /dev/null
+++ b/docs/apis/streaming/event_time.md
@@ -0,0 +1,148 @@
+---
+title: "Event Time"
+
+sub-nav-id: eventtime
+sub-nav-group: streaming
+sub-nav-pos: 2
+---
+<!--
+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.
+-->
+
+* toc
+{:toc}
+
+# Event Time / Processing Time / Ingestion Time
+
+Flink supports different notions of *time* in streaming programs.
+
+- **Processing time:** Processing time refers to the system time of the machine that is executing the
+ respective operation.
+
+ When a streaming program runs on processing time, all time-based operations (like time windows) will
+ use the system clock of the machines that run the respective operator. For example, an hourly
+ processing time window will include all records that arrived at a specific operator between the
+ times when the system clock indicated the full hour.
+
+ Processing time is the simplest notion of time and requires no coordination between streams and machines.
+ It provides the best performance and the lowest latency. However, in distributed and asynchronous
+ environments processing time does not provide determinism, because it is susceptible to the speed at which
+ records arrive in the system (for example from the message queue), and to the speed at which the
+ records flow between operators inside the system.
+
+- **Event time:** Event time is the time that each individual event occurred on its producing device.
+ This time is typically embedded within the records before they enter Flink and that *event timestamp*
+ can be extracted from the record. An hourly event time window will contain all records that carry an
+ event timestamp that falls into that hour, regardless of when the records arrive, and in what order
+ they arrive.
+
+ Event time gives correct results even on out-of-order events, late events, or on replays
+ of data from backups or persistent logs. In event time, the progress of time depends on the data,
+ not on any wall clocks. Event time programs must specify how to generate *Event Time Watermarks*,
+ which is the mechanism that signals time progress in event time. The mechanism is
+ described below.
+
+ Event time processing often incurs a certain latency, due to it nature of waiting a certain time for
+ late events and out-of-order events. Because of that, event time programs are often combined with
+ *processing time* operations.
+
+- **Ingestion time:** Ingestion time is the time that events enter Flink. At the source operator, each
+ records gets the source's current time as a timestamp, and time-based operations (like time windows)
+ refer to that timestamp.
+
+ *Ingestion Time* sits conceptually in between *Event Time* and *Processing Time*. Compared to
+ *Processing Time*, it is slightly more expensive, but gives more predictable results: Because
+ *Ingestion Time* uses stable timestamps (assigned once at the source), different window operations
+ over the records will refer to the same timestamp, whereas in *Processing Time* each window operator
+ may assign the record to a different window (based on the local system clock and any transport delay).
+
+ Compered to *Event Time*, *Ingestion Time* programs cannot handle any out-of-order events or late data,
+ but the programs don't have to specify how to generate *Watermarks*.
+
+ Internally, *Ingestion Time* is treated much like event time, with automatic timestamp assignment and
+ automatic Watermark generation.
+
+<img src="fig/times_clocks.svg" class="center" width="80%" />
+
+
+### Setting a Time Characteristic
+
+The first part of a Flink DataStream program is usually to set the base *time characteristic*. That setting
+defines how data stream sources behave (for example whether to assign timestamps), and what notion of
+time the window operations like `KeyedStream.timeWindow(Time.secondss(30))` refer to.
+
+The following example shows a Flink program that aggregates events in hourly time windows. The behavior of the
+windows adapts with the time characteristic.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
+
+// alternatively:
+// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));
+
+stream
+ .keyBy( (event) -> event.getUser() )
+ .timeWindow(Time.hours(1))
+ .reduce( (a, b) -> a.add(b) )
+ .addSink(...);
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+val env = StreamExecutionEnvironment.getExecutionEnvironment
+env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
+
+// alternatively:
+// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
+// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+
+val stream: DataStream[MyEvent] = env.addSource(new FlinkKafkaConsumer09[MyEvent](topic, schema, props))
+
+stream
+ .keyBy( _.getUser )
+ .timeWindow(Time.hours(1))
+ .reduce( (a, b) => a.add(b) )
+ .addSink(...)
+{% endhighlight %}
+</div>
+</div>
+
+
+Note that in order to run this example in *Event Time*, the program needs to use either an event time
+source, or inject a *Timestamp Assigner & Watermark Generator*. Those functions describe how to access
+the event timestamps, and what timely out-of-orderness the event stream exhibits.
+
+The section below describes the general mechanism behind *Timestamps* and *Watermarks*. For a guide how
+to use timestamp assignment and watermark generation in the Flink DataStream API, please refer to
+[Generating Timestamps / Watermarks]({{ site.baseurl }}/apis/streaming/event_timestamps_watermarks.html)
+
+
+# Event Time and Watermarks
+
+*Note: For a deep introduction to Event Time, please refer also to the paper on the [Dataflow Model](https://static.googleusercontent.com/media/research.google.com/en//pubs/archive/43864.pdf)*
+
+
+
+
http://git-wip-us.apache.org/repos/asf/flink/blob/0e61ffc5/docs/apis/streaming/event_timestamps_watermarks.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/event_timestamps_watermarks.md b/docs/apis/streaming/event_timestamps_watermarks.md
new file mode 100644
index 0000000..7718d47
--- /dev/null
+++ b/docs/apis/streaming/event_timestamps_watermarks.md
@@ -0,0 +1,358 @@
+---
+title: "Generating Timestamps / Watermarks"
+
+sub-nav-group: streaming
+sub-nav-pos: 1
+sub-nav-parent: eventtime
+---
+<!--
+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.
+-->
+
+* toc
+{:toc}
+
+
+This section is relevant for program running on **Event Time**. For an introduction to *Event Time*,
+*Processing Time*, and *Ingestion Time*, please refer to the [event time introduction]({{ site.baseurl }}/apis/streaming/event_time.html)
+
+To work with *Event Time*, streaming programs need to set the *time characteristic* accordingly.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+val env = StreamExecutionEnvironment.getExecutionEnvironment
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+{% endhighlight %}
+</div>
+</div>
+
+
+## Assigning Timestamps
+
+In order to work with *Event Time*, Flink needs to know the events' *timestamps*, meaning each element in the
+stream needs to get its event timestamp *assigned*. That happens usually by accessing/extracting the
+timestamp from some field in the element.
+
+Timestamp assignment goes hand-in-hand with generating watermarks, which tell the system about
+the progress in event time.
+
+There are two ways to assign timestamps and generate Watermarks:
+
+ 1. Directly in the data stream source
+ 2. Via a *TimestampAssigner / WatermarkGenerator*
+
+
+### Source Functions with Timestamps and Watermarks
+
+Stream sources can also directly assign timestamps to the elements they produce and emit Watermarks. In that case,
+no Timestamp Assigner is needed.
+
+To assign a timestamp to an element in the soruce directly, the source must use the `collectWithTimestamp(...)`
+method on the `SourceContext`. To generate Watermarks, the source must call the `emitWatermark(Watermark)` function.
+
+Below is a simple example of a source *(non-checkpointed)* that assigns timestamps and generates Watermarks
+depending on special events:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+@Override
+public void run(SourceContext<MyType> ctx) throws Exception {
+ while (/* condition */) {
+ MyType next = getNext();
+ ctx.collectWithTimestamp(next, next.getEventTimestamp());
+
+ if (next.hasWatermarkTime()) {
+ ctx.emitWatermark(new Watermark(next.getWatermarkTime()));
+ }
+ }
+}
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+override def run(ctx: SourceContext[MyType]): Unit = {
+ while (/* condition */) {
+ val next: MyType = getNext()
+ ctx.collectWithTimestamp(next, next.eventTimestamp)
+
+ if (next.hasWatermarkTime) {
+ ctx.emitWatermark(new Watermark(next.getWatermarkTime))
+ }
+ }
+}
+{% endhighlight %}
+</div>
+</div>
+
+*Note:* If the streaming program uses a TimestampAssigner on a stream where elements have a timestamp already,
+those timestamps will be overwritten by the TimestampAssigner. Similarly, Watermarks will be overwritten as well.
+
+
+### Timestamp Assigners / Watermark Generators
+
+Timestamp Assigners take a stream and produce a new stream with timestamped elements and watermarks. If the
+original stream had timestamps or watermarks already, the timestamp assigner overwrites those.
+
+The timestamp assigners occur usually direct after the data source, but it is not strictly required to. A
+common pattern is for example to parse (*MapFunction*) and filter (*FilterFunction*) before the timestamp assigner.
+In any case, the timestamp assigner needs to occur before the first operation on event time
+(such as the first window operation).
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));
+
+DataStream<MyEvent> withTimestampsAndWatermarks = stream
+ .filter( event -> event.severity() == WARNING )
+ .assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks());
+
+withTimestampsAndWatermarks
+ .keyBy( (event) -> event.getGroup() )
+ .timeWindow(Time.seconds(10))
+ .reduce( (a, b) -> a.add(b) )
+ .addSink(...);
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+val env = StreamExecutionEnvironment.getExecutionEnvironment
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+
+val stream: DataStream[MyEvent] = env.addSource(new FlinkKafkaConsumer09[MyEvent](topic, schema, props))
+
+val withTimestampsAndWatermarks: DataStream[MyEvent] = stream
+ .filter( _.severity == WARNING )
+ .assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks())
+
+withTimestampsAndWatermarks
+ .keyBy( _.getGroup )
+ .timeWindow(Time.seconds(10))
+ .reduce( (a, b) => a.add(b) )
+ .addSink(...)
+{% endhighlight %}
+</div>
+</div>
+
+
+#### **With Ascending timestamps**
+
+The simplest case for generating watermarks is the case where timestamps within one source occur in ascending order.
+In that case, the current timestamp can always act as a watermark, because no lower timestamps will occur any more.
+
+Note that it is only necessary that timestamps are ascending *per parallel data source instance*. For example, if
+in a specific setup one Kafka partition is read by one parallel data source instance, then it is only necessary that
+timestamps are ascending within each Kafka partition. Flink's Watermark merging mechanism will generate correct
+whenever parallel streams are shuffled, unioned, connected, or merged.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+DataStream<MyEvent> stream = ...
+
+DataStream<MyEvent> withTimestampsAndWatermarks =
+ stream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<MyEvent>() {
+
+ @Override
+ public long extractAscendingTimestamp(MyEvent element) {
+ return element.getCreationTime();
+ }
+});
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+val stream: DataStream[MyEvent] = ...
+
+val withTimestampsAndWatermarks = stream.assignAscendingTimestamps( _.getCreationTime )
+{% endhighlight %}
+</div>
+</div>
+
+
+*Note:* The generation of watermarks on ascending timestamps is a special case of the periodic watermark
+generation described in the next section.
+
+
+#### **With Periodic Watermarks**
+
+The `AssignerWithPeriodicWatermarks` assigns timestamps and generate watermarks periodically (possibly depending
+the stream elements, or purely based on processing time).
+
+The interval (every *n* milliseconds) in which the watermark will be generated is defined via
+`ExecutionConfig.setAutoWatermarkInterval(...)`. Each time, the assigner's `getCurrentWatermark()` method will be
+called, and a new Watermark will be emitted, if the returned Watermark is non-null and larger than the previous
+Watermark.
+
+Two simple examples of timestamp assigners with periodic watermark generation are below.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+/**
+ * This generator generates watermarks assuming that elements come out of order to a certain degree only.
+ * The latest elements for a certain timestamp t will arrive at most n milliseconds after the earliest
+ * elements for timestamp t.
+ */
+public class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks<MyEvent> {
+
+ private final long maxOutOfOrderness = 3500; // 3.5 seconds
+
+ private long currentMaxTimestamp;
+
+ @Override
+ public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
+ long timestamp = element.getCreationTime();
+ currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
+ return timestamp;
+ }
+
+ @Override
+ public Watermark getCurrentWatermark() {
+ // return the watermark as current highest timestamp minus the out-of-orderness bound
+ return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
+ }
+}
+
+/**
+ * This generator generates watermarks that are lagging behind processing time by a certain amount.
+ * It assumes that elements arrive in Flink after at most a certain time.
+ */
+public class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks<MyEvent> {
+
+ private final long maxTimeLag = 5000; // 5 seconds
+
+ @Override
+ public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
+ return element.getCreationTime();
+ }
+
+ @Override
+ public Watermark getCurrentWatermark() {
+ // return the watermark as current time minus the maximum time lag
+ return new Watermark(System.currentTimeMillis() - maxTimeLag);
+ }
+}
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+/**
+ * This generator generates watermarks assuming that elements come out of order to a certain degree only.
+ * The latest elements for a certain timestamp t will arrive at most n milliseconds after the earliest
+ * elements for timestamp t.
+ */
+class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {
+
+ val maxOutOfOrderness = 3500L; // 3.5 seconds
+
+ var currentMaxTimestamp: Long;
+
+ override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {
+ val timestamp = element.getCreationTime()
+ currentMaxTimestamp = max(timestamp, currentMaxTimestamp)
+ timestamp;
+ }
+
+ override def getCurrentWatermark(): Watermark = {
+ // return the watermark as current highest timestamp minus the out-of-orderness bound
+ new Watermark(currentMaxTimestamp - maxOutOfOrderness);
+ }
+}
+
+/**
+ * This generator generates watermarks that are lagging behind processing time by a certain amount.
+ * It assumes that elements arrive in Flink after at most a certain time.
+ */
+class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {
+
+ val maxTimeLag = 5000L; // 5 seconds
+
+ override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {
+ element.getCreationTime
+ }
+
+ override def getCurrentWatermark(): Watermark = {
+ // return the watermark as current time minus the maximum time lag
+ new Watermark(System.currentTimeMillis() - maxTimeLag)
+ }
+}
+{% endhighlight %}
+</div>
+</div>
+
+
+#### **With Punctuated Watermarks**
+
+To generate Watermarks whenever a certain event indicates that a new watermark can be generated, use the
+`AssignerWithPunctuatedWatermarks`. For this class, Flink will first call the `extractTimestamp(...)` method
+to assign the element a timestamp, and then immediately call for that element the
+`checkAndGetNextWatermark(...)` method.
+
+The `checkAndGetNextWatermark(...)` method gets the timestamp that was assigned in the `extractTimestamp(...)`
+method, and can decide whether it wants to generate a Watermark. Whenever the `checkAndGetNextWatermark(...)`
+method returns a non-null Watermark, and that Watermark is larger than the latest previous Watermark, that
+new Watermark will be emitted.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+public class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks<MyEvent> {
+
+ @Override
+ public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
+ return element.getCreationTime();
+ }
+
+ @Override
+ public Watermark checkAndGetNextWatermark(MyEvent lastElement, long extractedTimestamp) {
+ return element.hasWatermarkMarker() ? new Watermark(extractedTimestamp) : null;
+ }
+}
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[MyEvent] {
+
+ override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {
+ element.getCreationTime
+ }
+
+ override def checkAndGetNextWatermark(lastElement: MyEvent, extractedTimestamp: Long): Watermark = {
+ if (element.hasWatermarkMarker()) new Watermark(extractedTimestamp) else null
+ }
+}
+{% endhighlight %}
+</div>
+</div>
+
+*Note:* It is possible to generate a watermark on every single event. However, because each watermark causes some
+computation downstream, an excessive number of watermarks slows down performance.
+
http://git-wip-us.apache.org/repos/asf/flink/blob/0e61ffc5/docs/apis/streaming/fig/times_clocks.svg
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/fig/times_clocks.svg b/docs/apis/streaming/fig/times_clocks.svg
new file mode 100644
index 0000000..2dede77
--- /dev/null
+++ b/docs/apis/streaming/fig/times_clocks.svg
@@ -0,0 +1,368 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+
+<!--
+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.
+-->
+
+<svg
+ xmlns:dc="http://purl.org/dc/elements/1.1/"
+ xmlns:cc="http://creativecommons.org/ns#"
+ xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+ xmlns:svg="http://www.w3.org/2000/svg"
+ xmlns="http://www.w3.org/2000/svg"
+ version="1.1"
+ width="447.60001"
+ height="212.21001"
+ id="svg2">
+ <defs
+ id="defs4" />
+ <metadata
+ id="metadata7">
+ <rdf:RDF>
+ <cc:Work
+ rdf:about="">
+ <dc:format>image/svg+xml</dc:format>
+ <dc:type
+ rdf:resource="http://purl.org/dc/dcmitype/StillImage" />
+ <dc:title></dc:title>
+ </cc:Work>
+ </rdf:RDF>
+ </metadata>
+ <g
+ transform="translate(56.082351,-319.04344)"
+ id="layer1">
+ <g
+ transform="translate(-93.876994,199.60201)"
+ id="g2989">
+ <path
+ d="m 229.20098,184.79534 c -5.33505,0 -9.65748,-3.98488 -9.65748,-8.90738 0,-4.9225 4.32243,-8.90738 9.65748,-8.90738"
+ id="path2991"
+ style="fill:none;stroke:#000000;stroke-width:1.24703336px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 190.55232,166.98058 38.64866,0 c 5.33505,0 9.66685,3.98488 9.66685,8.90738 0,4.9225 -4.3318,8.90738 -9.66685,8.90738 l -38.64866,0 c -5.33505,0 -9.66685,-3.98488 -9.66685,-8.90738 0,-4.9225 4.3318,-8.90738 9.66685,-8.90738 z"
+ id="path2993"
+ style="fill:none;stroke:#000000;stroke-width:1.24703336px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 229.20098,257.30142 c -5.33505,0 -9.65748,-3.94737 -9.65748,-8.82299 0,-4.87562 4.32243,-8.83237 9.65748,-8.83237"
+ id="path2995"
+ style="fill:none;stroke:#000000;stroke-width:1.24703336px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 190.55232,239.64606 38.64866,0 c 5.33505,0 9.66685,3.95675 9.66685,8.83237 0,4.87562 -4.3318,8.82299 -9.66685,8.82299 l -38.64866,0 c -5.33505,0 -9.66685,-3.94737 -9.66685,-8.82299 0,-4.87562 4.3318,-8.83237 9.66685,-8.83237 z"
+ id="path2997"
+ style="fill:none;stroke:#000000;stroke-width:1.24703336px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 80.56023,234.02034 c 0,-1.98775 1.5752,-3.59108 3.516071,-3.59108 1.950248,0 3.516072,1.60333 3.516072,3.59108 0,1.98775 -1.565824,3.59108 -3.516072,3.59108 -1.940871,0 -3.516071,-1.60333 -3.516071,-3.59108"
+ id="path2999"
+ style="fill:#000000;fill-opacity:1;fill-rule:evenodd;stroke:none" />
+ <path
+ d="m 80.56023,234.02034 c 0,-1.98775 1.5752,-3.59108 3.516071,-3.59108 1.950248,0 3.516072,1.60333 3.516072,3.59108 0,1.98775 -1.565824,3.59108 -3.516072,3.59108 -1.940871,0 -3.516071,-1.60333 -3.516071,-3.59108"
+ id="path3001"
+ style="fill:none;stroke:#000000;stroke-width:1.24703336px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 84.160687,237.61142 0,10.90451"
+ id="path3003"
+ style="fill:none;stroke:#000000;stroke-width:1.87523818px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="M 84.160687,246.5188 70.405815,280.69502"
+ id="path3005"
+ style="fill:none;stroke:#000000;stroke-width:1.87523818px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 84.160687,246.5188 13.754871,34.17622"
+ id="path3007"
+ style="fill:none;stroke:#000000;stroke-width:1.87523818px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 78.684992,259.17666 10.548214,0"
+ id="path3009"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 76.809753,265.73999 15.001905,0"
+ id="path3011"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 74.465706,271.36571 19.221191,0"
+ id="path3013"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 72.121658,276.99142 24.6125,0"
+ id="path3015"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 90.030182,223.30336 c 5.475695,5.23191 5.672595,13.91426 0.440681,19.38996"
+ id="path3017"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 95.740282,223.30336 c 5.475698,5.23191 5.672598,13.91426 0.440681,19.38996"
+ id="path3019"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 101.44101,223.30336 c 5.47569,5.23191 5.67259,13.91426 0.44068,19.38996"
+ id="path3021"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="M 77.241058,242.65581 C 71.765363,237.4239 71.568463,228.74155 76.800377,223.26585"
+ id="path3023"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="M 71.540334,242.65581 C 66.064639,237.4239 65.867739,228.74155 71.099653,223.26585"
+ id="path3025"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="M 65.83961,242.65581 C 60.363915,237.4239 60.167015,228.74155 65.398929,223.26585"
+ id="path3027"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 66.186529,155.39161 c 0,-3.26292 2.644086,-5.91638 5.916377,-5.91638 l 23.646752,0 c 3.262915,0 5.907002,2.65346 5.907002,5.91638 l 0,40.21448 c 0,3.26291 -2.644087,5.907 -5.907002,5.907 l -23.646752,0 c -3.272291,0 -5.916377,-2.64409 -5.916377,-5.907 z"
+ id="path3029"
+ style="fill:none;stroke:#000000;stroke-width:1.24703336px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 70.24642,156.03856 27.350348,0 0,4.37869 -27.350348,0 z"
+ id="path3031"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 70.24642,162.45188 27.350348,0 0,4.5287 -27.350348,0 z"
+ id="path3033"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 70.24642,169.01521 27.350348,0 0,4.5287 -27.350348,0 z"
+ id="path3035"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 70.24642,175.57855 27.350348,0 0,4.3693 -27.350348,0 z"
+ id="path3037"
+ style="fill:none;stroke:#000000;stroke-width:0.62820476px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <text
+ x="42.695183"
+ y="136.73421"
+ id="text3039"
+ xml:space="preserve"
+ style="font-size:11.2514286px;font-style:normal;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Event Producer</text>
+ <text
+ x="164.95934"
+ y="136.73421"
+ id="text3041"
+ xml:space="preserve"
+ style="font-size:11.2514286px;font-style:normal;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Message Queue</text>
+ <path
+ d="m 314.49619,174.79095 c 0,-8.79487 7.14465,-15.93953 15.93952,-15.93953 8.81362,0 15.93952,7.14466 15.93952,15.93953 0,8.81361 -7.1259,15.93952 -15.93952,15.93952 -8.79487,0 -15.93952,-7.12591 -15.93952,-15.93952"
+ id="path3043"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <text
+ x="317.23706"
+ y="129.91739"
+ id="text3045"
+ xml:space="preserve"
+ style="font-size:11.2514286px;font-style:normal;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Flink</text>
+ <text
+ x="295.78436"
+ y="143.4191"
+ id="text3047"
+ xml:space="preserve"
+ style="font-size:11.2514286px;font-style:normal;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Data Source</text>
+ <text
+ x="415.78812"
+ y="129.91739"
+ id="text3049"
+ xml:space="preserve"
+ style="font-size:11.2514286px;font-style:normal;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Flink</text>
+ <text
+ x="379.48352"
+ y="143.4191"
+ id="text3051"
+ xml:space="preserve"
+ style="font-size:11.2514286px;font-style:normal;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Window Operator</text>
+ <path
+ d="m 314.49619,248.71283 c 0,-8.79486 7.14465,-15.93952 15.93952,-15.93952 8.81362,0 15.93952,7.14466 15.93952,15.93952 0,8.81362 -7.1259,15.93953 -15.93952,15.93953 -8.79487,0 -15.93952,-7.12591 -15.93952,-15.93953"
+ id="path3053"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 408.89567,174.79095 c 0,-8.79487 7.16341,-15.93953 16.01454,-15.93953 8.85112,0 16.01453,7.14466 16.01453,15.93953 0,8.81361 -7.16341,15.93952 -16.01453,15.93952 -8.85113,0 -16.01454,-7.12591 -16.01454,-15.93952"
+ id="path3055"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 408.89567,248.71283 c 0,-8.79486 7.16341,-15.93952 16.01454,-15.93952 8.85112,0 16.01453,7.14466 16.01453,15.93952 0,8.81362 -7.16341,15.93953 -16.01453,15.93953 -8.85113,0 -16.01454,-7.12591 -16.01454,-15.93953"
+ id="path3057"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 432.33615,169.63404 c 2.41906,0 4.3693,0.16877 4.3693,0.37505 l 0,9.43245 c 0,0.18752 -1.95024,0.35629 -4.3693,0.35629"
+ id="path3059"
+ style="fill:none;stroke:#000000;stroke-width:1.87523818px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 418.5719,169.63404 c -2.49407,0 -4.51932,0.16877 -4.51932,0.3938 l 0,9.39494 c 0,0.20628 2.02525,0.37505 4.51932,0.37505"
+ id="path3061"
+ style="fill:none;stroke:#000000;stroke-width:1.87523818px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 432.33615,244.02474 c 2.41906,0 4.3693,0.16877 4.3693,0.37504 l 0,9.4137 c 0,0.20628 -1.95024,0.37505 -4.3693,0.37505"
+ id="path3063"
+ style="fill:none;stroke:#000000;stroke-width:1.87523818px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 418.5719,244.02474 c -2.49407,0 -4.51932,0.16877 -4.51932,0.37504 l 0,9.4137 c 0,0.20628 2.02525,0.37505 4.51932,0.37505"
+ id="path3065"
+ style="fill:none;stroke:#000000;stroke-width:1.87523818px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 323.57234,170.53415 7.50095,0 0,-4.18178 7.50095,8.36357 -7.50095,8.36356 0,-4.18178 -7.50095,0 z"
+ id="path3067"
+ style="fill:none;stroke:#000000;stroke-width:0.61882859px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 323.57234,244.60606 7.50095,0 0,-4.18178 7.50095,8.36356 -7.50095,8.36356 0,-4.18178 -7.50095,0 z"
+ id="path3069"
+ style="fill:none;stroke:#000000;stroke-width:0.61882859px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 117.44616,174.17212 49.70319,0 0,1.24703 -49.70319,0 z m 42.52103,-4.2943 8.41982,4.91313 -8.41982,4.9225 c -0.30004,0.16877 -0.68446,0.075 -0.85324,-0.22503 -0.17814,-0.30004 -0.075,-0.68447 0.22503,-0.85324 l 7.50096,-4.37868 0,1.07826 -7.50096,-4.37868 c -0.30003,-0.16877 -0.40317,-0.55319 -0.22503,-0.85323 0.16878,-0.30004 0.5532,-0.3938 0.85324,-0.22503 z"
+ id="path3071"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.00937619px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 117.44616,247.92523 49.70319,0 0,1.25641 -49.70319,0 z m 42.52103,-4.28492 8.41982,4.91313 -8.41982,4.91312 c -0.30004,0.17815 -0.68446,0.075 -0.85324,-0.22503 -0.17814,-0.29066 -0.075,-0.67508 0.22503,-0.85323 l 7.50096,-4.36931 0,1.07827 -7.50096,-4.37869 c -0.30003,-0.16877 -0.40317,-0.55319 -0.22503,-0.85323 0.16878,-0.30004 0.5532,-0.40317 0.85324,-0.22503 z"
+ id="path3073"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.00937619px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 117.86809,240.81808 52.5348,-48.78432 -0.84386,-0.91887 -52.54417,48.78432 z m 50.19075,-40.74892 2.83161,-9.33869 -9.51683,2.13777 c -0.33755,0.075 -0.5532,0.40318 -0.47819,0.74072 0.075,0.33754 0.41255,0.5532 0.7501,0.47819 l 8.47607,-1.90337 -0.74072,-0.7876 -2.51282,8.3073 c -0.10313,0.33755 0.0844,0.68447 0.41256,0.77823 0.32816,0.10314 0.67508,-0.0844 0.77822,-0.41255 z"
+ id="path3075"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.00937619px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 252.14452,174.17212 49.71256,0 0,1.25641 -49.71256,0 z m 42.5304,-4.2943 8.41982,4.91313 -8.41982,4.93187 c -0.30004,0.16877 -0.67509,0.075 -0.86261,-0.22503 -0.16877,-0.30004 -0.075,-0.69384 0.22503,-0.86261 l 7.50095,-4.3693 0,1.06888 -7.50095,-4.3693 c -0.30004,-0.16877 -0.3938,-0.56257 -0.22503,-0.86261 0.18752,-0.30004 0.56257,-0.3938 0.86261,-0.22503 z"
+ id="path3077"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 252.14452,248.56281 49.71256,0 0,1.23766 -49.71256,0 z m 42.5304,-4.29429 8.41982,4.91312 -8.41982,4.91313 c -0.30004,0.18752 -0.67509,0.075 -0.86261,-0.22503 -0.16877,-0.30004 -0.075,-0.67509 0.22503,-0.84386 l 7.50095,-4.38806 0,1.08764 -7.50095,-4.3693 c -0.30004,-0.18753 -0.3938,-0.56258 -0.22503,-0.86261 0.18752,-0.30004 0.56257,-0.3938 0.86261,-0.22503 z"
+ id="path3079"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <text
+ x="189.87421"
+ y="177.52216"
+ id="text3081"
+ xml:space="preserve"
+ style="font-size:4.95062876px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">partition 1</text>
+ <text
+ x="189.87421"
+ y="251.06122"
+ id="text3083"
+ xml:space="preserve"
+ style="font-size:4.95062876px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">partition 2</text>
+ <path
+ d="m 352.46976,174.17212 49.71256,0 0,1.25641 -49.71256,0 z m 42.5304,-4.2943 8.41982,4.91313 -8.41982,4.93187 c -0.30004,0.16877 -0.67509,0.075 -0.86261,-0.22503 -0.16877,-0.30004 -0.075,-0.69384 0.22503,-0.86261 l 7.50095,-4.3693 0,1.06888 -7.50095,-4.3693 c -0.30004,-0.16877 -0.3938,-0.56257 -0.22503,-0.86261 0.18752,-0.30004 0.56257,-0.3938 0.86261,-0.22503 z"
+ id="path3085"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 352.46976,248.56281 49.71256,0 0,1.23766 -49.71256,0 z m 42.5304,-4.29429 8.41982,4.91312 -8.41982,4.91313 c -0.30004,0.18752 -0.67509,0.075 -0.86261,-0.22503 -0.16877,-0.30004 -0.075,-0.67509 0.22503,-0.84386 l 7.50095,-4.38806 0,1.08764 -7.50095,-4.3693 c -0.30004,-0.18753 -0.3938,-0.56258 -0.22503,-0.86261 0.18752,-0.30004 0.56257,-0.3938 0.86261,-0.22503 z"
+ id="path3087"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 352.93857,177.66006 55.60081,58.61994 -0.90012,0.86261 -55.61956,-58.63869 z m 53.76307,50.46266 2.23154,9.4887 -9.35744,-2.73785 c -0.33754,-0.0938 -0.52507,-0.4313 -0.43131,-0.76884 0.0938,-0.33755 0.45006,-0.52507 0.7876,-0.43131 l 8.32606,2.43781 -0.7876,0.7501 -1.98775,-8.45733 c -0.075,-0.33754 0.13127,-0.67508 0.46881,-0.75009 0.33754,-0.0938 0.67508,0.13126 0.75009,0.46881 z"
+ id="path3089"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 352.93857,246.8751 52.58167,-54.81321 -0.90011,-0.86261 -52.60043,54.79446 z m 50.70644,-46.65593 2.28779,-9.4887 -9.39495,2.68159 c -0.31879,0.0938 -0.52506,0.45006 -0.4313,0.76885 0.0938,0.33754 0.45006,0.52506 0.7876,0.4313 l 8.34481,-2.38155 -0.7876,-0.7501 -2.02526,8.43858 c -0.0938,0.33754 0.11251,0.67508 0.45006,0.76884 0.33754,0.075 0.67508,-0.13126 0.76885,-0.46881 z"
+ id="path3091"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 40.561401,302.94472 c 0,-5.00689 4.069266,-9.07615 9.0574,-9.07615 5.006885,0 9.076152,4.06926 9.076152,9.07615 0,5.00688 -4.069267,9.0574 -9.076152,9.0574 -4.988134,0 -9.0574,-4.05052 -9.0574,-9.0574"
+ id="path3093"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 49.468782,303.57292 0,-7.66972"
+ id="path3095"
+ style="fill:none;stroke:#000000;stroke-width:1.24703336px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 52.862963,306.02011 -3.394181,-2.45656"
+ id="path3097"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 417.33424,302.94472 c 0,-5.00689 4.08802,-9.07615 9.13241,-9.07615 5.06315,0 9.15117,4.06926 9.15117,9.07615 0,5.00688 -4.08802,9.0574 -9.15117,9.0574 -5.04439,0 -9.13241,-4.05052 -9.13241,-9.0574"
+ id="path3099"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 426.39164,303.5823 0,-7.66972"
+ id="path3101"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 429.78582,306.02011 -3.39418,-2.45656"
+ id="path3103"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 297.319,302.94472 c 0,-5.00689 4.05052,-9.07615 9.0574,-9.07615 5.00689,0 9.0574,4.06926 9.0574,9.07615 0,5.00688 -4.05051,9.0574 -9.0574,9.0574 -5.00688,0 -9.0574,-4.05052 -9.0574,-9.0574"
+ id="path3105"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 306.22639,303.5823 0,-7.66972"
+ id="path3107"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 309.62057,306.02011 -3.39418,-2.45656"
+ id="path3109"
+ style="fill:none;stroke:#000000;stroke-width:1.25640953px;stroke-linecap:butt;stroke-linejoin:miter;stroke-miterlimit:4;stroke-opacity:1;stroke-dasharray:none" />
+ <text
+ x="67.69368"
+ y="308.08017"
+ id="text3111"
+ xml:space="preserve"
+ style="font-size:8.70110512px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Event</text>
+ <text
+ x="69.493912"
+ y="318.58148"
+ id="text3113"
+ xml:space="preserve"
+ style="font-size:8.70110512px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Time</text>
+ <text
+ x="319.44556"
+ y="308.07742"
+ id="text3115"
+ xml:space="preserve"
+ style="font-size:8.70110512px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Ingestion</text>
+ <text
+ x="329.19681"
+ y="318.57874"
+ id="text3117"
+ xml:space="preserve"
+ style="font-size:8.70110512px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Time</text>
+ <text
+ x="443.55893"
+ y="309.46686"
+ id="text3119"
+ xml:space="preserve"
+ style="font-size:8.70110512px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Window</text>
+ <text
+ x="437.40817"
+ y="319.96817"
+ id="text3121"
+ xml:space="preserve"
+ style="font-size:8.70110512px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Processing</text>
+ <text
+ x="449.85974"
+ y="330.46951"
+ id="text3123"
+ xml:space="preserve"
+ style="font-size:8.70110512px;font-style:italic;font-weight:normal;text-align:start;text-anchor:start;fill:#000000;font-family:Verdana">Time</text>
+ <path
+ d="m 307.31402,271.36571 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23765 1.25641,0 0,1.23765 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.
49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23765 1.25641,0 0,1.23765 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23765 1.25641,0 0,1.23765 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0
,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23765 1.25641,0 0,1.23765 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.23766 1.25641,0 0,1.23766 -1.25641,0 z m 0,-2.49407 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.49406 0,-1.25641 1.25641,0 0,1.25641 -1.25641,0 z m 0,-2.51282 0,-1.40643 1.08764,0 0,1.25641 -0.46881,0 0.63758,-0.61883 0,0.76885 -1.25641,0 z m 2.34405,-1.40643 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23765,
0 0,1.25641 -1.23765,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23765,0 0,1.25641 -1.23765,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282
,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23765,0 0,1.25641 -1.23765,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641
z m 2.51282,0 1.23765,0 0,1.25641 -1.23765,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23765,0 0,1.25641 -1.23765,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.51282,0 1.23766,0 0,1.25641 -1.23766,0 0,-1.25641 z m 2.49407,0 1.25641,0 0,1.25641 -1.25641,0 0,-1.25641 z m 2.49406,0 1.25641,0 0,1.25641 -1.25641,0
0,-1.25641 z m 2.51282,0 0.93762,0 0,1.5752 -1.25641,0 0,-0.93762 0.61883,0.61883 -0.30004,0 0,-1.25641 z m 0.93762,2.81286 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23765 -1.25641,0 0,-1.23765 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23765 -1.25641,0 0,-1.23765 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1
.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23765 -1.25641,0 0,-1.23765 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.2564 -1.25641,0 0,-1.2564 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-
1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23765 -1.25641,0 0,-1.23765 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.2564 -1.25641,0 0,-1.2564 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.49406 0,1.25641 -1.25641,0 0,-1.25641 1.25641,0 z m 0,2.51282 0,1.23766 -1.25641,0 0,-1.23766 1.25641,0 z m 0,2.49407 0,1.72522 -0.7876,0 0,-1.25641 0.15002,0 -0.61883,0.61883 0,-1.08764 1.25641,0 z m -2.04401,1.72522 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.2
5641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.2564,0 0,-1.25641 1.2564,0 0,1.25641 z m -2.51281,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23765,0 0,-1.25641 1.23765,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,
0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23765,0 0,-1.25641 1.23765,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23765,0 0,-1.25641 1.23765,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25
641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23765,0 0,-1.25641 1.23765,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49406,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.51282,0 -1.23766,0 0,-1.25641 1.23766,0 0,1.25641 z m -2.49407
,0 -1.25641,0 0,-1.25641 1.25641,0 0,1.25641 z m -2.49407,0 -0.63758,0 0,-1.25641 0.63758,0 0,1.25641 z"
+ id="path3125"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 70.968387,294.47802 12.751619,-25.48449 -1.115767,-0.56257 -12.751619,25.49386 z m 14.429957,-24.65001 c 0.618829,-1.22828 0.112514,-2.72847 -1.115767,-3.3473 -1.237657,-0.61883 -2.737847,-0.12189 -3.356676,1.11577 -0.618828,1.23766 -0.12189,2.73785 1.115767,3.35668 1.237657,0.61882 2.737847,0.11251 3.356676,-1.12515 z"
+ id="path3127"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.00937619px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 325.74761,293.15598 6.37581,-31.97281 -1.23765,-0.24379 -6.37581,31.97281 z m 8.21355,-31.59777 c 0.26253,-1.36892 -0.61883,-2.68159 -1.969,-2.94412 -1.35017,-0.28129 -2.66284,0.60008 -2.94413,1.95025 -0.26253,1.36892 0.61883,2.68159 1.969,2.94412 1.35017,0.28129 2.68159,-0.60008 2.94413,-1.95025 z"
+ id="path3129"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ <path
+ d="m 440.38092,294.03734 -14.5706,-36.49214 1.16265,-0.46881 14.5706,36.49214 z m -16.31457,-35.7983 c -0.50632,-1.27516 0.11251,-2.73785 1.40643,-3.24416 1.27516,-0.52507 2.73784,0.11251 3.24416,1.38768 0.50631,1.27516 -0.11252,2.73784 -1.38768,3.24416 -1.29391,0.52506 -2.73784,-0.11252 -3.26291,-1.38768 z"
+ id="path3131"
+ style="fill:#000000;fill-opacity:1;fill-rule:nonzero;stroke:#000000;stroke-width:0.01875238px;stroke-linecap:butt;stroke-linejoin:round;stroke-opacity:1;stroke-dasharray:none" />
+ </g>
+ </g>
+</svg>
http://git-wip-us.apache.org/repos/asf/flink/blob/0e61ffc5/docs/apis/streaming/windows.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/windows.md b/docs/apis/streaming/windows.md
index 5f239fc..8af303b 100644
--- a/docs/apis/streaming/windows.md
+++ b/docs/apis/streaming/windows.md
@@ -3,7 +3,7 @@ title: "Windows"
sub-nav-id: windows
sub-nav-group: streaming
-sub-nav-pos: 2
+sub-nav-pos: 3
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
[5/6] flink git commit: [FLINK-3495] [tests] Disable RocksDB tests on
Windows
Posted by se...@apache.org.
[FLINK-3495] [tests] Disable RocksDB tests on Windows
This closes #1711
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2cde19b1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2cde19b1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2cde19b1
Branch: refs/heads/release-1.0
Commit: 2cde19b1ae6199f40d84915bea81c0f5049dcfe8
Parents: 5a63cbf
Author: zentol <s....@web.de>
Authored: Thu Feb 25 14:31:34 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Mar 4 20:53:44 2016 +0100
----------------------------------------------------------------------
.../contrib/streaming/state/RocksDBAsyncKVSnapshotTest.java | 8 ++++++++
.../streaming/state/RocksDBStateBackendConfigTest.java | 8 ++++++++
.../contrib/streaming/state/RocksDBStateBackendTest.java | 8 ++++++++
3 files changed, 24 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2cde19b1/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncKVSnapshotTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncKVSnapshotTest.java b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncKVSnapshotTest.java
index 8806c7f..8f3c59d 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncKVSnapshotTest.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBAsyncKVSnapshotTest.java
@@ -42,9 +42,12 @@ import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
import org.apache.flink.streaming.runtime.tasks.StreamTaskStateList;
+import org.apache.flink.util.OperatingSystem;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
+import org.junit.Assume;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.mockito.PowerMockito;
@@ -67,6 +70,11 @@ import static org.junit.Assert.assertTrue;
@SuppressWarnings("serial")
public class RocksDBAsyncKVSnapshotTest {
+ @Before
+ public void checkOperatingSystem() {
+ Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows());
+ }
+
/**
* This ensures that asynchronous state handles are actually materialized asynchonously.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/2cde19b1/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
index e62d39c..8e0993b 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
@@ -26,6 +26,9 @@ import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.util.OperatingSystem;
+import org.junit.Assume;
+import org.junit.Before;
import org.junit.Test;
import org.rocksdb.CompactionStyle;
@@ -45,6 +48,11 @@ public class RocksDBStateBackendConfigTest {
private static final String TEMP_URI = new File(System.getProperty("java.io.tmpdir")).toURI().toString();
+ @Before
+ public void checkOperatingSystem() {
+ Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows());
+ }
+
// ------------------------------------------------------------------------
// RocksDB local file directory
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2cde19b1/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
index 3c77eb9..57f906e 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendTest.java
@@ -22,6 +22,9 @@ import org.apache.commons.io.FileUtils;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.runtime.state.StateBackendTestBase;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.util.OperatingSystem;
+import org.junit.Assume;
+import org.junit.Before;
import java.io.File;
import java.io.IOException;
@@ -35,6 +38,11 @@ public class RocksDBStateBackendTest extends StateBackendTestBase<RocksDBStateBa
private File dbDir;
private File chkDir;
+ @Before
+ public void checkOperatingSystem() {
+ Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows());
+ }
+
@Override
protected RocksDBStateBackend getStateBackend() throws IOException {
dbDir = new File(new File(ConfigConstants.DEFAULT_TASK_MANAGER_TMP_PATH, UUID.randomUUID().toString()), "state");
[2/6] flink git commit: [FLINK-3491] [tests] Prevent URIException in
HDFSCopyTest and disable HDFSCopyUtilitiesTest on Windows
Posted by se...@apache.org.
[FLINK-3491] [tests] Prevent URIException in HDFSCopyTest and disable HDFSCopyUtilitiesTest on Windows
This closes #1703
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5a63cbfd
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5a63cbfd
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5a63cbfd
Branch: refs/heads/release-1.0
Commit: 5a63cbfdfc8cc1b5464b8e16306fa6feac16a9a0
Parents: 0e61ffc
Author: zentol <s....@web.de>
Authored: Wed Feb 24 11:14:23 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Mar 4 20:53:44 2016 +0100
----------------------------------------------------------------------
.../flink/streaming/util/HDFSCopyUtilitiesTest.java | 14 +++++++++++---
1 file changed, 11 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5a63cbfd/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/HDFSCopyUtilitiesTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/HDFSCopyUtilitiesTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/HDFSCopyUtilitiesTest.java
index 07e1e65..9022e18 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/HDFSCopyUtilitiesTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/HDFSCopyUtilitiesTest.java
@@ -17,6 +17,10 @@
*/
package org.apache.flink.streaming.util;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.OperatingSystem;
+import org.junit.Assume;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -26,7 +30,6 @@ import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
-import java.net.URI;
import static org.junit.Assert.assertTrue;
@@ -35,6 +38,11 @@ public class HDFSCopyUtilitiesTest {
@Rule
public TemporaryFolder tempFolder = new TemporaryFolder();
+ @Before
+ public void checkOperatingSystem() {
+ Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows());
+ }
+
/**
* This test verifies that a hadoop configuration is correctly read in the external
@@ -54,7 +62,7 @@ public class HDFSCopyUtilitiesTest {
HDFSCopyFromLocal.copyFromLocal(
originalFile,
- new URI(copyFile.getAbsolutePath()));
+ new Path(copyFile.getAbsolutePath()).toUri());
try (DataInputStream in = new DataInputStream(new FileInputStream(copyFile))) {
assertTrue(in.readUTF().equals("Hello there, 42!"));
@@ -79,7 +87,7 @@ public class HDFSCopyUtilitiesTest {
}
HDFSCopyToLocal.copyToLocal(
- new URI(originalFile.getAbsolutePath()),
+ new Path(originalFile.getAbsolutePath()).toUri(),
copyFile);
try (DataInputStream in = new DataInputStream(new FileInputStream(copyFile))) {