You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2015/02/19 17:41:22 UTC
[1/6] flink git commit: [streaming] Moved WindowingOptimizer to its
own class + several minor fixes
Repository: flink
Updated Branches:
refs/heads/master 6c01598a3 -> 8111177cf
[streaming] Moved WindowingOptimizer to its own class + several minor fixes
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8111177c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8111177c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8111177c
Branch: refs/heads/master
Commit: 8111177cf75c5252bf4e219a377cff5cc07640d8
Parents: 962db92
Author: Gyula Fora <gy...@apache.org>
Authored: Thu Feb 19 16:13:23 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Thu Feb 19 17:03:00 2015 +0100
----------------------------------------------------------------------
.../apache/flink/streaming/api/StreamGraph.java | 70 +++++++++++-
.../api/StreamingJobGraphGenerator.java | 75 -------------
.../flink/streaming/api/WindowingOptimzier.java | 110 +++++++++++++++++++
3 files changed, 174 insertions(+), 81 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8111177c/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
index c87abb5..8e2b49e 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -86,6 +87,7 @@ public class StreamGraph extends StreamingPlan {
private Map<String, Long> iterationTimeouts;
private Map<String, Map<String, OperatorState<?>>> operatorStates;
private Map<String, InputFormat<String, ?>> inputFormatLists;
+ private List<Map<String, ?>> containingMaps;
private Set<String> sources;
@@ -103,28 +105,50 @@ public class StreamGraph extends StreamingPlan {
}
public void initGraph() {
+ containingMaps = new ArrayList<Map<String, ?>>();
+
operatorParallelisms = new HashMap<String, Integer>();
+ containingMaps.add(operatorParallelisms);
bufferTimeouts = new HashMap<String, Long>();
+ containingMaps.add(bufferTimeouts);
outEdgeLists = new HashMap<String, List<String>>();
+ containingMaps.add(outEdgeLists);
outEdgeTypes = new HashMap<String, List<Integer>>();
+ containingMaps.add(outEdgeTypes);
selectedNames = new HashMap<String, List<List<String>>>();
+ containingMaps.add(selectedNames);
inEdgeLists = new HashMap<String, List<String>>();
+ containingMaps.add(inEdgeLists);
outputPartitioners = new HashMap<String, List<StreamPartitioner<?>>>();
+ containingMaps.add(outputPartitioners);
operatorNames = new HashMap<String, String>();
+ containingMaps.add(operatorNames);
invokableObjects = new HashMap<String, StreamInvokable<?, ?>>();
+ containingMaps.add(invokableObjects);
typeSerializersIn1 = new HashMap<String, StreamRecordSerializer<?>>();
+ containingMaps.add(typeSerializersIn1);
typeSerializersIn2 = new HashMap<String, StreamRecordSerializer<?>>();
+ containingMaps.add(typeSerializersIn2);
typeSerializersOut1 = new HashMap<String, StreamRecordSerializer<?>>();
+ containingMaps.add(typeSerializersOut1);
typeSerializersOut2 = new HashMap<String, StreamRecordSerializer<?>>();
+ containingMaps.add(typeSerializersOut1);
outputSelectors = new HashMap<String, List<OutputSelector<?>>>();
+ containingMaps.add(outputSelectors);
jobVertexClasses = new HashMap<String, Class<? extends AbstractInvokable>>();
+ containingMaps.add(jobVertexClasses);
iterationIds = new HashMap<String, Integer>();
+ containingMaps.add(jobVertexClasses);
iterationIDtoHeadName = new HashMap<Integer, String>();
iterationIDtoTailName = new HashMap<Integer, String>();
iterationTailCount = new HashMap<String, Integer>();
+ containingMaps.add(iterationTailCount);
iterationTimeouts = new HashMap<String, Long>();
+ containingMaps.add(iterationTailCount);
operatorStates = new HashMap<String, Map<String, OperatorState<?>>>();
+ containingMaps.add(operatorStates);
inputFormatLists = new HashMap<String, InputFormat<String, ?>>();
+ containingMaps.add(operatorStates);
sources = new HashSet<String>();
}
@@ -332,6 +356,24 @@ public class StreamGraph extends StreamingPlan {
outputPartitioners.get(upStream).remove(outputIndex);
}
+ public void removeVertex(String toRemove) {
+ List<String> outEdges = new ArrayList<String>(getOutEdges(toRemove));
+ List<String> inEdges = new ArrayList<String>(getInEdges(toRemove));
+
+ for (String output : outEdges) {
+ removeEdge(toRemove, output);
+ }
+
+ for (String input : inEdges) {
+ removeEdge(input, toRemove);
+ }
+
+ for (Map<String, ?> map : containingMaps) {
+ map.remove(toRemove);
+ }
+
+ }
+
private void addTypeSerializers(String vertexName, StreamRecordSerializer<?> in1,
StreamRecordSerializer<?> in2, StreamRecordSerializer<?> out1,
StreamRecordSerializer<?> out2) {
@@ -481,9 +523,12 @@ public class StreamGraph extends StreamingPlan {
public JobGraph getJobGraph(String jobGraphName) {
this.jobName = jobGraphName;
- StreamingJobGraphGenerator optimizer = new StreamingJobGraphGenerator(this);
- return optimizer.createJobGraph(jobGraphName);
+ WindowingOptimzier.optimizeGraph(this);
+
+ StreamingJobGraphGenerator jobgraphGenerator = new StreamingJobGraphGenerator(this);
+
+ return jobgraphGenerator.createJobGraph(jobGraphName);
}
public void setJobName(String jobName) {
@@ -569,17 +614,25 @@ public class StreamGraph extends StreamingPlan {
@Override
public String getStreamingPlanAsJSON() {
+ WindowingOptimzier.optimizeGraph(this);
+
try {
JSONObject json = new JSONObject();
JSONArray nodes = new JSONArray();
json.put("nodes", nodes);
-
+ List<Integer> operatorIDs = new ArrayList<Integer>();
for (String id : operatorNames.keySet()) {
+ operatorIDs.add(Integer.valueOf(id));
+ }
+ Collections.sort(operatorIDs);
+
+ for (Integer idInt : operatorIDs) {
JSONObject node = new JSONObject();
nodes.put(node);
- node.put("id", Integer.valueOf(id));
+ String id = idInt.toString();
+ node.put("id", idInt);
node.put("type", getOperatorName(id));
if (sources.contains(id)) {
@@ -588,8 +641,13 @@ public class StreamGraph extends StreamingPlan {
node.put("pact", "Data Stream");
}
- node.put("contents", getOperatorName(id) + " at "
- + getInvokable(id).getUserFunction().getClass().getSimpleName());
+ if (getInvokable(id) != null && getInvokable(id).getUserFunction() != null) {
+ node.put("contents", getOperatorName(id) + " at "
+ + getInvokable(id).getUserFunction().getClass().getSimpleName());
+ } else {
+ node.put("contents", getOperatorName(id));
+ }
+
node.put("parallelism", getParallelism(id));
int numIn = getInEdges(id).size();
http://git-wip-us.apache.org/repos/asf/flink/blob/8111177c/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
index f3edb78..9beea89 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
@@ -23,8 +23,6 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -38,7 +36,6 @@ import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.invokable.StreamInvokable.ChainingStrategy;
-import org.apache.flink.streaming.api.invokable.operator.windowing.StreamDiscretizer;
import org.apache.flink.streaming.api.streamvertex.StreamIterationHead;
import org.apache.flink.streaming.api.streamvertex.StreamIterationTail;
import org.apache.flink.streaming.partitioner.StreamPartitioner;
@@ -80,8 +77,6 @@ public class StreamingJobGraphGenerator {
init();
- applyWindowOptimizations();
-
setChaining();
setSlotSharing();
@@ -89,76 +84,6 @@ public class StreamingJobGraphGenerator {
return jobGraph;
}
- private void applyWindowOptimizations() {
- Set<Entry<String, StreamInvokable<?, ?>>> invokables = streamGraph.getInvokables();
- List<Tuple2<String, StreamDiscretizer<?>>> discretizers = new ArrayList<Tuple2<String, StreamDiscretizer<?>>>();
-
- // Get the discretizers
- for (Entry<String, StreamInvokable<?, ?>> entry : invokables) {
- if (entry.getValue() instanceof StreamDiscretizer) {
- discretizers.add(new Tuple2<String, StreamDiscretizer<?>>(entry.getKey(),
- (StreamDiscretizer<?>) entry.getValue()));
- }
- }
-
- // Share common discrtizers
- setDiscretizerReuse(discretizers);
-
- }
-
- private void setDiscretizerReuse(List<Tuple2<String, StreamDiscretizer<?>>> discretizers) {
- List<Tuple2<StreamDiscretizer<?>, List<String>>> matchingDiscretizers = new ArrayList<Tuple2<StreamDiscretizer<?>, List<String>>>();
-
- for (Tuple2<String, StreamDiscretizer<?>> discretizer : discretizers) {
- boolean inMatching = false;
- for (Tuple2<StreamDiscretizer<?>, List<String>> matching : matchingDiscretizers) {
- Set<String> discretizerInEdges = new HashSet<String>(
- streamGraph.getInEdges(discretizer.f0));
- Set<String> matchingInEdges = new HashSet<String>(
- streamGraph.getInEdges(matching.f1.get(0)));
-
- if (discretizer.f1.equals(matching.f0)
- && discretizerInEdges.equals(matchingInEdges)) {
- matching.f1.add(discretizer.f0);
- inMatching = true;
- break;
- }
- }
- if (!inMatching) {
- List<String> matchingNames = new ArrayList<String>();
- matchingNames.add(discretizer.f0);
- matchingDiscretizers.add(new Tuple2<StreamDiscretizer<?>, List<String>>(
- discretizer.f1, matchingNames));
- }
- }
-
- for (Tuple2<StreamDiscretizer<?>, List<String>> matching : matchingDiscretizers) {
- List<String> matchList = matching.f1;
- if (matchList.size() > 1) {
- String first = matchList.get(0);
- for (int i = 1; i < matchList.size(); i++) {
- replaceDiscretizer(matchList.get(i), first);
- }
- }
- }
- }
-
- private void replaceDiscretizer(String toReplace, String replaceWith) {
- // Convert to array to create a copy
- List<String> outEdges = new ArrayList<String>(streamGraph.getOutEdges(toReplace));
-
- int numOutputs = outEdges.size();
-
- // Reconnect outputs
- for (int i = 0; i < numOutputs; i++) {
- String outName = outEdges.get(i);
-
- streamGraph.setEdge(replaceWith, outName,
- streamGraph.getOutPartitioner(toReplace, outName), 0, new ArrayList<String>());
- streamGraph.removeEdge(toReplace, outName);
- }
- }
-
private void setChaining() {
for (String sourceName : streamGraph.getSources()) {
createChain(sourceName, sourceName);
http://git-wip-us.apache.org/repos/asf/flink/blob/8111177c/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimzier.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimzier.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimzier.java
new file mode 100644
index 0000000..f0fa772
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimzier.java
@@ -0,0 +1,110 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.invokable.StreamInvokable;
+import org.apache.flink.streaming.api.invokable.operator.windowing.StreamDiscretizer;
+
+public class WindowingOptimzier {
+
+ public static void optimizeGraph(StreamGraph streamGraph) {
+ Set<Entry<String, StreamInvokable<?, ?>>> invokables = streamGraph.getInvokables();
+ List<Tuple2<String, StreamDiscretizer<?>>> discretizers = new ArrayList<Tuple2<String, StreamDiscretizer<?>>>();
+
+ // Get the discretizers
+ for (Entry<String, StreamInvokable<?, ?>> entry : invokables) {
+ if (entry.getValue() instanceof StreamDiscretizer) {
+ discretizers.add(new Tuple2<String, StreamDiscretizer<?>>(entry.getKey(),
+ (StreamDiscretizer<?>) entry.getValue()));
+ }
+ }
+
+ // Share common discrtizers
+ setDiscretizerReuse(streamGraph, discretizers);
+ }
+
+ private static void setDiscretizerReuse(StreamGraph streamGraph,
+ List<Tuple2<String, StreamDiscretizer<?>>> discretizers) {
+ List<Tuple2<StreamDiscretizer<?>, List<String>>> matchingDiscretizers = new ArrayList<Tuple2<StreamDiscretizer<?>, List<String>>>();
+
+ for (Tuple2<String, StreamDiscretizer<?>> discretizer : discretizers) {
+ boolean inMatching = false;
+ for (Tuple2<StreamDiscretizer<?>, List<String>> matching : matchingDiscretizers) {
+ Set<String> discretizerInEdges = new HashSet<String>(
+ streamGraph.getInEdges(discretizer.f0));
+ Set<String> matchingInEdges = new HashSet<String>(
+ streamGraph.getInEdges(matching.f1.get(0)));
+
+ if (discretizer.f1.equals(matching.f0)
+ && discretizerInEdges.equals(matchingInEdges)) {
+ matching.f1.add(discretizer.f0);
+ inMatching = true;
+ break;
+ }
+ }
+ if (!inMatching) {
+ List<String> matchingNames = new ArrayList<String>();
+ matchingNames.add(discretizer.f0);
+ matchingDiscretizers.add(new Tuple2<StreamDiscretizer<?>, List<String>>(
+ discretizer.f1, matchingNames));
+ }
+ }
+
+ for (Tuple2<StreamDiscretizer<?>, List<String>> matching : matchingDiscretizers) {
+ List<String> matchList = matching.f1;
+ if (matchList.size() > 1) {
+ String first = matchList.get(0);
+ for (int i = 1; i < matchList.size(); i++) {
+ replaceDiscretizer(streamGraph, matchList.get(i), first);
+ }
+ }
+ }
+ }
+
+ private static void replaceDiscretizer(StreamGraph streamGraph, String toReplace,
+ String replaceWith) {
+ // Convert to array to create a copy
+ List<String> outEdges = new ArrayList<String>(streamGraph.getOutEdges(toReplace));
+
+ int numOutputs = outEdges.size();
+
+ // Reconnect outputs
+ for (int i = 0; i < numOutputs; i++) {
+ String outName = outEdges.get(i);
+
+ streamGraph.setEdge(replaceWith, outName,
+ streamGraph.getOutPartitioner(toReplace, outName), 0, new ArrayList<String>());
+ streamGraph.removeEdge(toReplace, outName);
+ }
+
+ List<String> inEdges = new ArrayList<String>(streamGraph.getInEdges(toReplace));
+ // Remove inputs
+ for (String inName : inEdges) {
+ streamGraph.removeEdge(inName, toReplace);
+ }
+
+ streamGraph.removeVertex(toReplace);
+ }
+}
[5/6] flink git commit: [streaming] Added cloneable versions of
MultiEvictionPolicy and MultiTriggerPolicy
Posted by gy...@apache.org.
[streaming] Added cloneable versions of MultiEvictionPolicy and MultiTriggerPolicy
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/962db92a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/962db92a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/962db92a
Branch: refs/heads/master
Commit: 962db92afee9eb8e5efce46b5c3f9f57e017cb1d
Parents: 09490c4
Author: Jonas Traub (powibol) <jo...@s-traub.com>
Authored: Thu Feb 19 14:54:15 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Thu Feb 19 17:03:00 2015 +0100
----------------------------------------------------------------------
.../policy/CloneableMultiEvictionPolicy.java | 84 ++++++++++++++++++++
.../policy/CloneableMultiTriggerPolicy.java | 63 +++++++++++++++
2 files changed, 147 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/962db92a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiEvictionPolicy.java
new file mode 100644
index 0000000..d86b174
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiEvictionPolicy.java
@@ -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.windowing.policy;
+
+import java.util.LinkedList;
+
+/**
+ * This policy does the same as {@link MultiEvictionPolicy}. Additionally it is
+ * cloneable and only cloneable policies can be passed to the constructor.
+ *
+ * @param <DATA>
+ * The type of data handled by this policy
+ */
+public class CloneableMultiEvictionPolicy<DATA> extends MultiEvictionPolicy<DATA> implements
+ CloneableEvictionPolicy<DATA> {
+
+ /**
+ * Default version id.
+ */
+ private static final long serialVersionUID = 1L;
+
+ private CloneableEvictionPolicy<DATA>[] allPolicies;
+ private EvictionStrategy strategy;
+
+ /**
+ * This policy does the same as {@link MultiTriggerPolicy}. Additionally it
+ * is cloneable and only cloneable policies can be passed to the
+ * constructor.
+ *
+ * When using this constructor the MAX strategy is used by default. You can
+ * select other strategies using
+ * {@link CloneableMultiEvictionPolicy#CloneableMultiEvictionPolicy(org.apache.flink.streaming.api.windowing.policy.MultiEvictionPolicy.EvictionStrategy, CloneableEvictionPolicy...)}
+ * .
+ *
+ * @param evictionPolicies
+ * some cloneable policies to be tied together.
+ */
+ public CloneableMultiEvictionPolicy(CloneableEvictionPolicy<DATA>... evictionPolicies) {
+ this(EvictionStrategy.MAX, evictionPolicies);
+ }
+
+ /**
+ * This policy does the same as {@link MultiTriggerPolicy}. Additionally it
+ * is cloneable and only cloneable policies can be passed to the
+ * constructor.
+ *
+ * @param strategy
+ * the strategy to be used. See {@link EvictionStrategy} for a
+ * list of possible options.
+ * @param evictionPolicies
+ * some cloneable policies to be tied together.
+ */
+ public CloneableMultiEvictionPolicy(EvictionStrategy strategy,
+ CloneableEvictionPolicy<DATA>... evictionPolicies) {
+ super(strategy, evictionPolicies);
+ this.allPolicies = evictionPolicies;
+ this.strategy = strategy;
+ }
+
+ @SuppressWarnings("unchecked")
+ public CloneableEvictionPolicy<DATA> clone() {
+ LinkedList<CloneableEvictionPolicy<DATA>> clonedPolicies = new LinkedList<CloneableEvictionPolicy<DATA>>();
+ for (int i = 0; i < allPolicies.length; i++) {
+ clonedPolicies.add(allPolicies[i].clone());
+ }
+ return new CloneableMultiEvictionPolicy<DATA>(strategy,
+ clonedPolicies.toArray(new CloneableEvictionPolicy[allPolicies.length]));
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/962db92a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiTriggerPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiTriggerPolicy.java
new file mode 100644
index 0000000..aaecefb
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CloneableMultiTriggerPolicy.java
@@ -0,0 +1,63 @@
+/*
+ * 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.windowing.policy;
+
+import java.util.LinkedList;
+
+/**
+ * This policy does the same as {@link MultiTriggerPolicy}. Additionally it is
+ * cloneable and only cloneable policies can be passed to the constructor.
+ *
+ * @param <DATA>
+ * The type of data handled by this policy
+ */
+public class CloneableMultiTriggerPolicy<DATA> extends MultiTriggerPolicy<DATA> implements
+ CloneableTriggerPolicy<DATA>, Cloneable {
+
+ /**
+ * Default version id.
+ */
+ private static final long serialVersionUID = 1L;
+
+ private CloneableTriggerPolicy<DATA>[] allPolicies;
+
+ /**
+ * This policy does the same as {@link MultiTriggerPolicy}. Additionally it
+ * is cloneable and only cloneable policies can be passed to the
+ * constructor.
+ *
+ * @param policies
+ * some cloneable policies to be tied together.
+ */
+ public CloneableMultiTriggerPolicy(CloneableTriggerPolicy<DATA>... policies) {
+ super(policies);
+ this.allPolicies = policies;
+ }
+
+ @SuppressWarnings("unchecked")
+ public CloneableTriggerPolicy<DATA> clone() {
+ LinkedList<CloneableTriggerPolicy<DATA>> clonedPolicies = new LinkedList<CloneableTriggerPolicy<DATA>>();
+ for (int i = 0; i < allPolicies.length; i++) {
+ clonedPolicies.add(allPolicies[i].clone());
+ }
+ return new CloneableMultiTriggerPolicy<DATA>(
+ clonedPolicies.toArray(new CloneableTriggerPolicy[allPolicies.length]));
+
+ }
+
+}
[4/6] flink git commit: [streaming] Discretizer sharing added with
further window optimzations for better chaining
Posted by gy...@apache.org.
[streaming] Discretizer sharing added with further window optimzations for better chaining
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ca7f36e1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ca7f36e1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ca7f36e1
Branch: refs/heads/master
Commit: ca7f36e1ee5791cdbda7efc5a01e329abfdcc5c2
Parents: e5528a2
Author: Gyula Fora <gy...@apache.org>
Authored: Thu Feb 19 14:11:25 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Thu Feb 19 17:03:00 2015 +0100
----------------------------------------------------------------------
.../apache/flink/streaming/api/StreamGraph.java | 26 +++++-
.../api/StreamingJobGraphGenerator.java | 86 +++++++++++++++++++-
.../api/datastream/DiscretizedStream.java | 16 +++-
.../windowing/GroupedStreamDiscretizer.java | 5 ++
.../operator/windowing/StreamDiscretizer.java | 6 ++
.../windowing/policy/CountEvictionPolicy.java | 5 ++
.../windowing/policy/CountTriggerPolicy.java | 5 ++
.../api/windowing/policy/DeltaPolicy.java | 5 ++
.../api/windowing/policy/PunctuationPolicy.java | 6 ++
.../windowing/policy/TimeEvictionPolicy.java | 6 ++
.../api/windowing/policy/TimeTriggerPolicy.java | 6 ++
.../policy/TumblingEvictionPolicy.java | 5 ++
.../windowing/WindowIntegrationTest.java | 70 ++++++++++++++--
13 files changed, 231 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
index 4800c96..c87abb5 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamGraph.java
@@ -27,6 +27,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import org.apache.flink.api.common.ExecutionConfig;
@@ -256,9 +257,10 @@ public class StreamGraph extends StreamingPlan {
addVertex(vertexName, CoStreamVertex.class, taskInvokableObject, operatorName, parallelism);
- addTypeSerializers(vertexName, new StreamRecordSerializer<IN1>(in1TypeInfo, executionConfig),
- new StreamRecordSerializer<IN2>(in2TypeInfo, executionConfig), new StreamRecordSerializer<OUT>(
- outTypeInfo, executionConfig), null);
+ addTypeSerializers(vertexName,
+ new StreamRecordSerializer<IN1>(in1TypeInfo, executionConfig),
+ new StreamRecordSerializer<IN2>(in2TypeInfo, executionConfig),
+ new StreamRecordSerializer<OUT>(outTypeInfo, executionConfig), null);
if (LOG.isDebugEnabled()) {
LOG.debug("CO-TASK: {}", vertexName);
@@ -319,6 +321,17 @@ public class StreamGraph extends StreamingPlan {
selectedNames.get(upStreamVertexName).add(outputNames);
}
+ public void removeEdge(String upStream, String downStream) {
+ int inputIndex = getInEdges(downStream).indexOf(upStream);
+ inEdgeLists.get(downStream).remove(inputIndex);
+
+ int outputIndex = getOutEdges(upStream).indexOf(downStream);
+ outEdgeLists.get(upStream).remove(outputIndex);
+ outEdgeTypes.get(upStream).remove(outputIndex);
+ selectedNames.get(upStream).remove(outputIndex);
+ outputPartitioners.get(upStream).remove(outputIndex);
+ }
+
private void addTypeSerializers(String vertexName, StreamRecordSerializer<?> in1,
StreamRecordSerializer<?> in2, StreamRecordSerializer<?> out1,
StreamRecordSerializer<?> out2) {
@@ -404,7 +417,8 @@ public class StreamGraph extends StreamingPlan {
}
public <OUT> void setOutType(String id, TypeInformation<OUT> outType) {
- StreamRecordSerializer<OUT> serializer = new StreamRecordSerializer<OUT>(outType, executionConfig);
+ StreamRecordSerializer<OUT> serializer = new StreamRecordSerializer<OUT>(outType,
+ executionConfig);
typeSerializersOut1.put(id, serializer);
}
@@ -480,6 +494,10 @@ public class StreamGraph extends StreamingPlan {
this.chaining = chaining;
}
+ public Set<Entry<String, StreamInvokable<?, ?>>> getInvokables() {
+ return invokableObjects.entrySet();
+ }
+
public Collection<String> getSources() {
return sources;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
index d57bfc0..f3edb78 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamingJobGraphGenerator.java
@@ -23,6 +23,8 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -36,6 +38,7 @@ import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.invokable.StreamInvokable.ChainingStrategy;
+import org.apache.flink.streaming.api.invokable.operator.windowing.StreamDiscretizer;
import org.apache.flink.streaming.api.streamvertex.StreamIterationHead;
import org.apache.flink.streaming.api.streamvertex.StreamIterationTail;
import org.apache.flink.streaming.partitioner.StreamPartitioner;
@@ -71,20 +74,97 @@ public class StreamingJobGraphGenerator {
public JobGraph createJobGraph(String jobName) {
jobGraph = new JobGraph(jobName);
+
// Turn lazy scheduling off
jobGraph.setScheduleMode(ScheduleMode.ALL);
init();
- for (String sourceName : streamGraph.getSources()) {
- createChain(sourceName, sourceName);
- }
+ applyWindowOptimizations();
+
+ setChaining();
setSlotSharing();
return jobGraph;
}
+ private void applyWindowOptimizations() {
+ Set<Entry<String, StreamInvokable<?, ?>>> invokables = streamGraph.getInvokables();
+ List<Tuple2<String, StreamDiscretizer<?>>> discretizers = new ArrayList<Tuple2<String, StreamDiscretizer<?>>>();
+
+ // Get the discretizers
+ for (Entry<String, StreamInvokable<?, ?>> entry : invokables) {
+ if (entry.getValue() instanceof StreamDiscretizer) {
+ discretizers.add(new Tuple2<String, StreamDiscretizer<?>>(entry.getKey(),
+ (StreamDiscretizer<?>) entry.getValue()));
+ }
+ }
+
+ // Share common discrtizers
+ setDiscretizerReuse(discretizers);
+
+ }
+
+ private void setDiscretizerReuse(List<Tuple2<String, StreamDiscretizer<?>>> discretizers) {
+ List<Tuple2<StreamDiscretizer<?>, List<String>>> matchingDiscretizers = new ArrayList<Tuple2<StreamDiscretizer<?>, List<String>>>();
+
+ for (Tuple2<String, StreamDiscretizer<?>> discretizer : discretizers) {
+ boolean inMatching = false;
+ for (Tuple2<StreamDiscretizer<?>, List<String>> matching : matchingDiscretizers) {
+ Set<String> discretizerInEdges = new HashSet<String>(
+ streamGraph.getInEdges(discretizer.f0));
+ Set<String> matchingInEdges = new HashSet<String>(
+ streamGraph.getInEdges(matching.f1.get(0)));
+
+ if (discretizer.f1.equals(matching.f0)
+ && discretizerInEdges.equals(matchingInEdges)) {
+ matching.f1.add(discretizer.f0);
+ inMatching = true;
+ break;
+ }
+ }
+ if (!inMatching) {
+ List<String> matchingNames = new ArrayList<String>();
+ matchingNames.add(discretizer.f0);
+ matchingDiscretizers.add(new Tuple2<StreamDiscretizer<?>, List<String>>(
+ discretizer.f1, matchingNames));
+ }
+ }
+
+ for (Tuple2<StreamDiscretizer<?>, List<String>> matching : matchingDiscretizers) {
+ List<String> matchList = matching.f1;
+ if (matchList.size() > 1) {
+ String first = matchList.get(0);
+ for (int i = 1; i < matchList.size(); i++) {
+ replaceDiscretizer(matchList.get(i), first);
+ }
+ }
+ }
+ }
+
+ private void replaceDiscretizer(String toReplace, String replaceWith) {
+ // Convert to array to create a copy
+ List<String> outEdges = new ArrayList<String>(streamGraph.getOutEdges(toReplace));
+
+ int numOutputs = outEdges.size();
+
+ // Reconnect outputs
+ for (int i = 0; i < numOutputs; i++) {
+ String outName = outEdges.get(i);
+
+ streamGraph.setEdge(replaceWith, outName,
+ streamGraph.getOutPartitioner(toReplace, outName), 0, new ArrayList<String>());
+ streamGraph.removeEdge(toReplace, outName);
+ }
+ }
+
+ private void setChaining() {
+ for (String sourceName : streamGraph.getSources()) {
+ createChain(sourceName, sourceName);
+ }
+ }
+
private List<Tuple2<String, String>> createChain(String startNode, String current) {
if (!builtNodes.contains(startNode)) {
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DiscretizedStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DiscretizedStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DiscretizedStream.java
index bc9f5b9..0c84d0a 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DiscretizedStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DiscretizedStream.java
@@ -73,7 +73,7 @@ public class DiscretizedStream<OUT> extends WindowedDataStream<OUT> {
WindowTransformation.REDUCEWINDOW, "Window Reduce", getType(),
new WindowReducer<OUT>(reduceFunction)).merge();
- if (!isGrouped()) {
+ if (!isGrouped() && out.discretizedStream.invokable instanceof WindowMerger) {
return out.transform(WindowTransformation.REDUCEWINDOW, "Window Reduce", out.getType(),
new WindowReducer<OUT>(discretizedStream.clean(reduceFunction)));
} else {
@@ -122,7 +122,9 @@ public class DiscretizedStream<OUT> extends WindowedDataStream<OUT> {
out.groupByKey = null;
return out;
- } else if (transformation == WindowTransformation.MAPWINDOW) {
+ } else if (transformation != WindowTransformation.MAPWINDOW
+ && parallelism != discretizedStream.getExecutionEnvironment()
+ .getDegreeOfParallelism()) {
return transform(transformation, "Window partitioner", getType(),
new WindowPartitioner<OUT>(parallelism)).setParallelism(parallelism);
} else {
@@ -137,8 +139,14 @@ public class DiscretizedStream<OUT> extends WindowedDataStream<OUT> {
private DiscretizedStream<OUT> merge() {
TypeInformation<StreamWindow<OUT>> type = discretizedStream.getType();
- return wrap(discretizedStream.groupBy(new WindowKey<OUT>()).transform("Window Merger",
- type, new WindowMerger<OUT>()));
+ // Only merge partitioned streams
+ if (discretizedStream.invokable instanceof WindowPartitioner) {
+ return wrap(discretizedStream.groupBy(new WindowKey<OUT>()).transform("Window Merger",
+ type, new WindowMerger<OUT>()));
+ } else {
+ return this;
+ }
+
}
@SuppressWarnings("rawtypes")
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
index ec01e28..5e21a31 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
@@ -125,4 +125,9 @@ public class GroupedStreamDiscretizer<IN> extends StreamDiscretizer<IN> {
}
}
+ @Override
+ public String toString() {
+ return "GroupedDiscretizer(Key: " + keySelector.getClass().getSimpleName() + ", Trigger: "
+ + triggerPolicy.toString() + ", Eviction: " + evictionPolicy.toString() + ")";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
index fb65283..104196e 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
@@ -213,4 +213,10 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, WindowEvent<IN>>
}
}
}
+
+ @Override
+ public String toString() {
+ return "Discretizer(Trigger: " + triggerPolicy.toString() + ", Eviction: " + evictionPolicy.toString()
+ + ")";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
index b4bdd43..3ede27b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
@@ -135,4 +135,9 @@ public class CountEvictionPolicy<IN> implements CloneableEvictionPolicy<IN> {
}
}
}
+
+ @Override
+ public String toString() {
+ return "CountPolicy(" + maxElements + ")";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
index c814f5d..6d8149a 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
@@ -101,4 +101,9 @@ public class CountTriggerPolicy<IN> implements CloneableTriggerPolicy<IN> {
}
}
}
+
+ @Override
+ public String toString() {
+ return "CountPolicy(" + max + ")";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
index 169196f..0811e05 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
@@ -128,4 +128,9 @@ public class DeltaPolicy<DATA> implements CloneableTriggerPolicy<DATA>,
}
}
}
+
+ @Override
+ public String toString() {
+ return "DeltaPolicy(" + threshold + ", " + deltaFuntion.getClass().getSimpleName() + ")";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
index 4b06aee..e37c443 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
@@ -135,4 +135,10 @@ public class PunctuationPolicy<IN, DATA> implements CloneableTriggerPolicy<IN>,
}
}
}
+
+ @Override
+ public String toString() {
+ return "PunctuationPolicy(" + punctuation + extractor != null ? ", "
+ + extractor.getClass().getSimpleName() : "" + ")";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
index de715af..982b6d5 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
@@ -150,4 +150,10 @@ public class TimeEvictionPolicy<DATA> implements ActiveEvictionPolicy<DATA>,
}
}
+ @Override
+ public String toString() {
+ return "TimePolicy(" + granularity + ", " + timestampWrapper.getClass().getSimpleName()
+ + ")";
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
index 84eccf5..7065582 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
@@ -213,4 +213,10 @@ public class TimeTriggerPolicy<DATA> implements ActiveTriggerPolicy<DATA>,
}
}
+ @Override
+ public String toString() {
+ return "TimePolicy(" + granularity + ", " + timestampWrapper.getClass().getSimpleName()
+ + ")";
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
index 8d9f028..08c49e9 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
@@ -96,4 +96,9 @@ public class TumblingEvictionPolicy<DATA> implements CloneableEvictionPolicy<DAT
return true;
}
}
+
+ @Override
+ public String toString() {
+ return "TumblingPolicy";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ca7f36e1/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowIntegrationTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowIntegrationTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowIntegrationTest.java
index 893510b..f111898 100755
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowIntegrationTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowIntegrationTest.java
@@ -32,6 +32,8 @@ import org.apache.flink.streaming.api.function.WindowMapFunction;
import org.apache.flink.streaming.api.function.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.StreamWindow;
import org.apache.flink.streaming.api.windowing.helper.Count;
+import org.apache.flink.streaming.api.windowing.helper.Time;
+import org.apache.flink.streaming.api.windowing.helper.Timestamp;
import org.apache.flink.streaming.util.TestStreamEnvironment;
import org.apache.flink.util.Collector;
import org.junit.Test;
@@ -97,12 +99,29 @@ public class WindowIntegrationTest implements Serializable {
source.window(Count.of(4)).groupBy(new ModKey(2)).mapWindow(new IdentityWindowMap())
.flatten().addSink(new CentralSink2());
- source.groupBy(new ModKey(3)).window(Count.of(2)).sum(0).getDiscretizedStream()
+ KeySelector<Integer, ?> key = new ModKey(2);
+ Timestamp<Integer> ts = new Timestamp<Integer>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public long getTimestamp(Integer value) {
+ return value;
+ }
+ };
+
+ source.groupBy(key).window(Time.of(4, ts, 1)).sum(0).getDiscretizedStream()
.addSink(new DistributedSink1());
source.groupBy(new ModKey(3)).window(Count.of(2)).groupBy(new ModKey(2))
.mapWindow(new IdentityWindowMap()).flatten().addSink(new DistributedSink2());
+ source.window(Count.of(2)).every(Count.of(3)).min(0).getDiscretizedStream()
+ .addSink(new CentralSink3());
+
+ source.groupBy(key).window(Time.of(4, ts, 1)).max(0).getDiscretizedStream()
+ .addSink(new DistributedSink3());
+
env.execute();
// sum ( Count of 2 slide 3 )
@@ -123,14 +142,13 @@ public class WindowIntegrationTest implements Serializable {
validateOutput(expected2, CentralSink2.windows);
- // groupby mod 3 sum ( Tumbling Count of 2)
+ // groupby mod 2 sum ( Tumbling Time of 4)
List<StreamWindow<Integer>> expected3 = new ArrayList<StreamWindow<Integer>>();
expected3.add(StreamWindow.fromElements(4));
expected3.add(StreamWindow.fromElements(5));
- expected3.add(StreamWindow.fromElements(16));
+ expected3.add(StreamWindow.fromElements(22));
+ expected3.add(StreamWindow.fromElements(8));
expected3.add(StreamWindow.fromElements(10));
- expected3.add(StreamWindow.fromElements(11));
- expected3.add(StreamWindow.fromElements(3));
validateOutput(expected3, DistributedSink1.windows);
@@ -146,6 +164,22 @@ public class WindowIntegrationTest implements Serializable {
validateOutput(expected4, DistributedSink2.windows);
+ // min ( Count of 2 slide 3 )
+ List<StreamWindow<Integer>> expected5 = new ArrayList<StreamWindow<Integer>>();
+ expected5.add(StreamWindow.fromElements(2));
+ expected5.add(StreamWindow.fromElements(4));
+ expected5.add(StreamWindow.fromElements(11));
+
+ validateOutput(expected5, CentralSink3.windows);
+
+ // groupby mod 2 max ( Tumbling Time of 4)
+ List<StreamWindow<Integer>> expected6 = new ArrayList<StreamWindow<Integer>>();
+ expected6.add(StreamWindow.fromElements(3));
+ expected6.add(StreamWindow.fromElements(5));
+ expected6.add(StreamWindow.fromElements(11));
+ expected6.add(StreamWindow.fromElements(4));
+ expected6.add(StreamWindow.fromElements(10));
+
}
public static <R> void validateOutput(List<R> expected, List<R> actual) {
@@ -179,6 +213,19 @@ public class WindowIntegrationTest implements Serializable {
}
@SuppressWarnings("serial")
+ private static class CentralSink3 implements SinkFunction<StreamWindow<Integer>> {
+
+ public static List<StreamWindow<Integer>> windows = Collections
+ .synchronizedList(new ArrayList<StreamWindow<Integer>>());
+
+ @Override
+ public void invoke(StreamWindow<Integer> value) throws Exception {
+ windows.add(value);
+ }
+
+ }
+
+ @SuppressWarnings("serial")
private static class DistributedSink1 implements SinkFunction<StreamWindow<Integer>> {
public static List<StreamWindow<Integer>> windows = Collections
@@ -203,4 +250,17 @@ public class WindowIntegrationTest implements Serializable {
}
}
+
+ @SuppressWarnings("serial")
+ private static class DistributedSink3 implements SinkFunction<StreamWindow<Integer>> {
+
+ public static List<StreamWindow<Integer>> windows = Collections
+ .synchronizedList(new ArrayList<StreamWindow<Integer>>());
+
+ @Override
+ public void invoke(StreamWindow<Integer> value) throws Exception {
+ windows.add(value);
+ }
+
+ }
}
[2/6] flink git commit: [streaming] Add MultiEvictionPolicy: It wraps
around multiple eviction policies and allows to use them at the same time. It
supports both,
active and not active policies at the same time. Additionally it introduces
different evict
Posted by gy...@apache.org.
[streaming] Add MultiEvictionPolicy: It wraps around multiple eviction policies and allows to use them at the same time. It supports both, active and not active policies at the same time. Additionally it introduces different eviction strategies the user can choose from.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/09490c4b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/09490c4b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/09490c4b
Branch: refs/heads/master
Commit: 09490c4b95f04079b15c38c4b576f5beb861c038
Parents: 64f7bf7
Author: Jonas Traub (powibol) <jo...@s-traub.com>
Authored: Thu Feb 19 14:51:23 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Thu Feb 19 17:03:00 2015 +0100
----------------------------------------------------------------------
.../windowing/policy/MultiEvictionPolicy.java | 172 +++++++++++++++++
.../policy/MultiEvictionPolicyTest.java | 186 +++++++++++++++++++
2 files changed, 358 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/09490c4b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java
new file mode 100644
index 0000000..d353dc8
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicy.java
@@ -0,0 +1,172 @@
+/*
+ * 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.windowing.policy;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * This policy provides the ability to use multiple eviction policies at the
+ * same time. It allows to use both, active and not active evictions.
+ *
+ * @param <DATA>
+ * The type of data-items handled by the policies
+ */
+public class MultiEvictionPolicy<DATA> implements ActiveEvictionPolicy<DATA> {
+
+ /**
+ * Default version id.
+ */
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * This enum provides the different options for the eviction strategy.
+ *
+ * You can choose from the following:
+ * <ul>
+ * <li>MIN: The number of elements to evict will be the smallest one which
+ * is greater than 0 and was returned by any of the given policies. If all
+ * policies return 0, the result is 0.</li>
+ * <li>MAX: The number of elements to evict will be the greatest one which
+ * was returned by any of the given policies.</li>
+ * <li>SUM: The number of elements to evict will be the sum of all values
+ * returned by the nested eviction policies.</li>
+ * <li>PRIORITY: Depending on the order in which the policies have been
+ * passed to the constructor, the first return value greater than 0 will be
+ * the the number of elements to evict. If all policies return 0, the result
+ * is 0.</li>
+ * </ul>
+ */
+ public enum EvictionStrategy {
+ MIN, MAX, SUM, PRIORITY;
+ }
+
+ private List<EvictionPolicy<DATA>> allEvictionPolicies;
+ private List<ActiveEvictionPolicy<DATA>> activeEvictionPolicies;
+ private EvictionStrategy selectedStrategy;
+
+ /**
+ * This policy provides the ability to use multiple eviction policies at the
+ * same time. It allows to use both, active and not active evictions.
+ *
+ * When using this constructor the MAX strategy is used by default. You can
+ * select other strategies using
+ * {@link MultiEvictionPolicy#MultiEvictionPolicy(EvictionStrategy, EvictionPolicy...)}
+ * .
+ *
+ * @param evictionPolicies
+ * Any active or not active eviction policies. Both types can be
+ * used at the same time.
+ */
+ public MultiEvictionPolicy(EvictionPolicy<DATA>... evictionPolicies) {
+ this(EvictionStrategy.MAX, evictionPolicies);
+ }
+
+ /**
+ * This policy provides the ability to use multiple eviction policies at the
+ * same time. It allows to use both, active and not active evictions.
+ *
+ * @param strategy
+ * the strategy to be used. See {@link EvictionStrategy} for a
+ * list of possible options.
+ * @param evictionPolicies
+ * Any active or not active eviction policies. Both types can be
+ * used at the same time.
+ */
+ public MultiEvictionPolicy(EvictionStrategy strategy, EvictionPolicy<DATA>... evictionPolicies) {
+ // initialize lists of policies
+ this.allEvictionPolicies = new LinkedList<EvictionPolicy<DATA>>();
+ this.activeEvictionPolicies = new LinkedList<ActiveEvictionPolicy<DATA>>();
+
+ // iterate over policies and add them to the lists
+ for (EvictionPolicy<DATA> ep : evictionPolicies) {
+ this.allEvictionPolicies.add(ep);
+ if (ep instanceof ActiveEvictionPolicy) {
+ this.activeEvictionPolicies.add((ActiveEvictionPolicy<DATA>) ep);
+ }
+ }
+
+ // Remember eviction strategy
+ this.selectedStrategy = strategy;
+ }
+
+ @Override
+ public int notifyEviction(DATA datapoint, boolean triggered, int bufferSize) {
+ LinkedList<Integer> results = new LinkedList<Integer>();
+ for (EvictionPolicy<DATA> policy : allEvictionPolicies) {
+ results.add(policy.notifyEviction(datapoint, triggered, bufferSize));
+ }
+ return getNumToEvict(results);
+ }
+
+ @Override
+ public int notifyEvictionWithFakeElement(Object datapoint, int bufferSize) {
+ LinkedList<Integer> results = new LinkedList<Integer>();
+ for (ActiveEvictionPolicy<DATA> policy : activeEvictionPolicies) {
+ results.add(policy.notifyEvictionWithFakeElement(datapoint, bufferSize));
+ }
+ return getNumToEvict(results);
+ }
+
+ private int getNumToEvict(LinkedList<Integer> items) {
+ int result;
+ switch (selectedStrategy) {
+
+ case MIN:
+ result = Integer.MAX_VALUE;
+ for (Integer item : items) {
+ if (result > item) {
+ result = item;
+ }
+ }
+ return result;
+
+ case MAX:
+ result = 0;
+ for (Integer item : items) {
+ if (result < item) {
+ result = item;
+ }
+ }
+ return result;
+
+ case SUM:
+ result = 0;
+ for (Integer item : items) {
+ System.out.print(item+":");
+ result += item;
+ }
+ System.out.println(result);
+ return result;
+
+ case PRIORITY:
+ for (Integer item : items) {
+ if (item > 0) {
+ return item;
+ }
+ }
+ return 0;
+
+ default:
+ // The following line should never be reached. Just for the
+ // compiler.
+ return 0;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/09490c4b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java
new file mode 100644
index 0000000..3214aa7
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiEvictionPolicyTest.java
@@ -0,0 +1,186 @@
+/*
+ * 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.windowing.policy;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.junit.Test;
+
+public class MultiEvictionPolicyTest {
+
+ private final List<Integer> tuples = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+
+ private final CloneableEvictionPolicy<Integer> evictionPolicy1 = new CountEvictionPolicy<Integer>(
+ 7, 3);
+ private final CloneableEvictionPolicy<Integer> evictionPolicy2 = new CountEvictionPolicy<Integer>(
+ 3, 1);
+ private final CloneableEvictionPolicy<Integer> evictionPolicy3 = new CountEvictionPolicy<Integer>(
+ 5, 2);
+
+ private final CloneableEvictionPolicy<Integer> activeEvictionPolicy1 = new ActiveCloneableEvictionPolicyWrapper<Integer>(
+ evictionPolicy1);
+ private final CloneableEvictionPolicy<Integer> activeEvictionPolicy2 = new ActiveCloneableEvictionPolicyWrapper<Integer>(
+ evictionPolicy2);
+ private final CloneableEvictionPolicy<Integer> activeEvictionPolicy3 = new ActiveCloneableEvictionPolicyWrapper<Integer>(
+ evictionPolicy3);
+
+ // From policies specified above the expected output is:
+ // 1.: 0000000300
+ // 2.: 0001111111
+ // 3.: 0000020202
+ private final Integer[] maxResult = { 0, 0, 0, 1, 1, 2, 1, 3, 1, 2 };
+ private final Integer[] minResult = { 0, 0, 0, 0, 0, 0, 0, 1, 0, 0 };
+ private final Integer[] sumResult = { 0, 0, 0, 1, 1, 3, 1, 6, 1, 3 };
+ private final Integer[] priorityResult = { 0, 0, 0, 1, 1, 1, 1, 3, 1, 1 };
+
+ /*
+ * Test cases for not active policies
+ */
+
+ @Test
+ public void notActiveEvictionMAXStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MAX, maxResult);
+ }
+
+ @Test
+ public void notActiveEvictionMINStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MIN, minResult);
+ }
+
+ @Test
+ public void notActiveEvictionSUMStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.SUM, sumResult);
+ }
+
+ @Test
+ public void notActiveEvictionPRIORITYStrategyTest() {
+ runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.PRIORITY, priorityResult);
+ }
+
+ /*
+ * Test cases for active policies
+ */
+
+ @Test
+ public void activeEvictionMAXStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MAX, maxResult);
+ }
+
+ @Test
+ public void activeEvictionMINStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.MIN, minResult);
+ }
+
+ @Test
+ public void activeEvictionSUMStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.SUM, sumResult);
+ }
+
+ @Test
+ public void activeEvictionPRIORITYStrategyTest() {
+ runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy.PRIORITY, priorityResult);
+ }
+
+ /**
+ * Helper method: It runs the test with the given input using the not active
+ * policies and applies the strategy defined in the parameter.
+ *
+ * @param strategy
+ * the eviction strategy to be used
+ * @param expectedResult
+ * the result we expect
+ */
+ private void runNotActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy strategy,
+ Integer[] expectedResult) {
+ @SuppressWarnings("unchecked")
+ MultiEvictionPolicy<Integer> multiEviction = new MultiEvictionPolicy<Integer>(strategy,
+ evictionPolicy1.clone(), evictionPolicy2.clone(), evictionPolicy3.clone());
+
+ List<Integer> result = new LinkedList<Integer>();
+
+ int buffersize = 0;
+ for (Integer tuple : tuples) {
+ // The buffer size should not matter, but we keep it for the case of
+ // later policy changes.
+ // The trigger does not matter. Always set it to false.
+ int eviction = multiEviction.notifyEviction(tuple, false, buffersize);
+ buffersize -= eviction;
+ result.add(eviction);
+
+ if (buffersize < 0) {
+ buffersize = 0;
+ }
+
+ buffersize++;
+ }
+
+ arrayEqualityCheck(expectedResult, result.toArray());
+ }
+
+ /**
+ * Helper method: It runs the test with the given input using the active
+ * policies and applies the strategy defined in the parameter.
+ *
+ * @param strategy
+ * the eviction strategy to be used
+ * @param expectedResult
+ * the result we expect
+ */
+ private void runActiveEvictionTest(MultiEvictionPolicy.EvictionStrategy strategy,
+ Integer[] expectedResult) {
+ @SuppressWarnings("unchecked")
+ MultiEvictionPolicy<Integer> multiEviction = new MultiEvictionPolicy<Integer>(strategy,
+ activeEvictionPolicy1.clone(), activeEvictionPolicy2.clone(),
+ activeEvictionPolicy3.clone());
+
+ List<Integer> result = new LinkedList<Integer>();
+
+ int buffersize = 0;
+ for (Integer tuple : tuples) {
+ // The buffer size should not matter, but we keep it for the case of
+ // later policy changes.
+ // The trigger does not matter. Always set it to false.
+ int eviction = multiEviction.notifyEvictionWithFakeElement(tuple, buffersize);
+ buffersize -= eviction;
+ result.add(eviction);
+
+ if (buffersize < 0) {
+ buffersize = 0;
+ }
+
+ buffersize++;
+ }
+
+ arrayEqualityCheck(expectedResult, result.toArray());
+ }
+
+ private void arrayEqualityCheck(Object[] array1, Object[] array2) {
+ assertEquals(
+ "The result arrays must have the same length. (Expected: " + Arrays.asList(array1)
+ + "; Actual: " + Arrays.asList(array2) + ")", array1.length, array2.length);
+ for (int i = 0; i < array1.length; i++) {
+ assertEquals("Unequal fields at position " + i + "(Expected: " + Arrays.asList(array1)
+ + "; Actual: " + Arrays.asList(array2) + ")", array1[i], array2[i]);
+ }
+ }
+
+}
[6/6] flink git commit: [streaming] Add MultiTriggerPolicy: It wraps
around multiple triggers and allows to use them at the same time. It supports
both, active and not active policies at the same time.
Posted by gy...@apache.org.
[streaming] Add MultiTriggerPolicy: It wraps around multiple triggers and allows to use them at the same time. It supports both, active and not active policies at the same time.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/64f7bf7d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/64f7bf7d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/64f7bf7d
Branch: refs/heads/master
Commit: 64f7bf7d2162dcec23e76f0d2f28367c75bd00a6
Parents: ca7f36e
Author: Jonas Traub (powibol) <jo...@s-traub.com>
Authored: Thu Feb 19 14:48:43 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Thu Feb 19 17:03:00 2015 +0100
----------------------------------------------------------------------
.../windowing/policy/MultiTriggerPolicy.java | 123 ++++++++++
.../policy/MultiTriggerPolicyTest.java | 230 +++++++++++++++++++
2 files changed, 353 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/64f7bf7d/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicy.java
new file mode 100644
index 0000000..a3c6a22
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicy.java
@@ -0,0 +1,123 @@
+/*
+ * 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.windowing.policy;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * This class allows to use multiple trigger policies at the same time. It
+ * allows to use both, active and not active triggers.
+ *
+ * @param <DATA>
+ * the data type handled by this policy
+ */
+public class MultiTriggerPolicy<DATA> implements ActiveTriggerPolicy<DATA> {
+
+ /**
+ * Default version id.
+ */
+ private static final long serialVersionUID = 1L;
+
+ private List<TriggerPolicy<DATA>> allTriggerPolicies;
+ private List<ActiveTriggerPolicy<DATA>> activeTriggerPolicies;
+
+ /**
+ * This policy allows to use multiple trigger policies at the same time. It
+ * allows to use both, active and not active triggers.
+ *
+ * This policy triggers in case at least one of the nested policies
+ * triggered. If active policies are nested all produces fake elements will
+ * be returned.
+ *
+ * @param policies
+ * Any active or not active trigger policies. Both types can be
+ * used at the same time.
+ */
+ public MultiTriggerPolicy(TriggerPolicy<DATA>... policies) {
+ allTriggerPolicies = new LinkedList<TriggerPolicy<DATA>>();
+ activeTriggerPolicies = new LinkedList<ActiveTriggerPolicy<DATA>>();
+
+ for (TriggerPolicy<DATA> policy : policies) {
+ this.allTriggerPolicies.add(policy);
+ if (policy instanceof ActiveTriggerPolicy) {
+ this.activeTriggerPolicies.add((ActiveTriggerPolicy<DATA>) policy);
+ }
+ }
+ }
+
+ @Override
+ public boolean notifyTrigger(DATA datapoint) {
+ boolean trigger = false;
+ for (TriggerPolicy<DATA> policy : allTriggerPolicies) {
+ if (policy.notifyTrigger(datapoint)) {
+ trigger = true;
+ // Do not at a break here. All trigger must see the element!
+ }
+ }
+ return trigger;
+ }
+
+ @Override
+ public Object[] preNotifyTrigger(DATA datapoint) {
+ List<Object> fakeElements = new LinkedList<Object>();
+ for (ActiveTriggerPolicy<DATA> policy : activeTriggerPolicies) {
+ for (Object fakeElement : policy.preNotifyTrigger(datapoint)) {
+ fakeElements.add(fakeElement);
+ }
+ }
+ return fakeElements.toArray();
+ }
+
+ @Override
+ public Runnable createActiveTriggerRunnable(ActiveTriggerCallback callback) {
+ List<Runnable> runnables = new LinkedList<Runnable>();
+ for (ActiveTriggerPolicy<DATA> policy : activeTriggerPolicies) {
+ Runnable tmp = policy.createActiveTriggerRunnable(callback);
+ if (tmp != null) {
+ runnables.add(tmp);
+ }
+ }
+ if (runnables.size() == 0) {
+ return null;
+ } else {
+ return new MultiActiveTriggerRunnable(runnables);
+ }
+ }
+
+ /**
+ * This class serves a nest for all active trigger runnables. Once the run
+ * method gets executed, all the runnables are started in own threads.
+ */
+ private class MultiActiveTriggerRunnable implements Runnable {
+
+ List<Runnable> runnables;
+
+ MultiActiveTriggerRunnable(List<Runnable> runnables) {
+ this.runnables = runnables;
+ }
+
+ @Override
+ public void run() {
+ for (Runnable runnable : runnables) {
+ new Thread(runnable).start();
+ }
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/64f7bf7d/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicyTest.java
new file mode 100644
index 0000000..de6cdcb
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/MultiTriggerPolicyTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.windowing.policy;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.flink.streaming.api.windowing.helper.Timestamp;
+import org.apache.flink.streaming.api.windowing.helper.TimestampWrapper;
+import org.junit.Test;
+
+public class MultiTriggerPolicyTest {
+
+ /**
+ * This test covers all regular notify call. It takes no fake elements into
+ * account.
+ */
+ @Test
+ public void testWithoutActivePolicies() {
+ List<Integer> tuples = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15,
+ 16);
+
+ TriggerPolicy<Integer> firstPolicy = new CountTriggerPolicy<Integer>(3);
+ TriggerPolicy<Integer> secondPolicy = new CountTriggerPolicy<Integer>(5);
+ TriggerPolicy<Integer> thirdPolicy = new CountTriggerPolicy<Integer>(8);
+ @SuppressWarnings("unchecked")
+ TriggerPolicy<Integer> multiTrigger = new MultiTriggerPolicy<Integer>(firstPolicy,
+ secondPolicy, thirdPolicy);
+
+ // From above policies the expected output is (first element is 0):
+ // first: 3, 6, 9, 12, 15,...
+ // second: 5, 10, 15,...
+ // third: 8, 16, 24,...
+ // combination: 3,5,6,8,9,10,12,15,16
+ List<Integer> expectedResult = Arrays.asList(3, 5, 6, 8, 9, 10, 12, 15, 16);
+ List<Integer> actualResult = new LinkedList<Integer>();
+
+ for (int i = 0; i < tuples.size(); i++) {
+ if (multiTrigger.notifyTrigger(tuples.get(i))) {
+ actualResult.add(i);
+ }
+ }
+
+ // check equal sizes
+ assertTrue("The expected result list and the actual result list must have the same size,"
+ + " but they are different. (expected: " + expectedResult.size() + "; actual: "
+ + actualResult.size() + "). Actual result is: " + actualResult
+ + " Expected result is: " + expectedResult,
+ expectedResult.size() == actualResult.size());
+
+ // check equal elements within result list/expected list
+ for (int i = 0; i < expectedResult.size(); i++) {
+ assertTrue("The actual and the expected result does not match at position " + i
+ + ". (expected: " + expectedResult.get(i) + "; actual: " + actualResult.get(i)
+ + "). Actual result is: " + actualResult + " Expected result is: "
+ + expectedResult, expectedResult.get(i) == actualResult.get(i));
+ }
+ }
+
+ /**
+ * This test covers the pre-notify calls to active policies. I takes no
+ * regular notify into account.
+ */
+ @Test
+ public void testWithActivePolicies() {
+
+ // create some test data
+ Integer[] times = { 1, 3, 20, 26 };
+
+ // create a timestamp
+ @SuppressWarnings("serial")
+ Timestamp<Integer> timeStamp = new Timestamp<Integer>() {
+
+ @Override
+ public long getTimestamp(Integer value) {
+ return value;
+ }
+
+ };
+
+ // create policy
+ TimeTriggerPolicy<Integer> firstPolicy = new TimeTriggerPolicy<Integer>(5,
+ new TimestampWrapper<Integer>(timeStamp, 0));
+ TimeTriggerPolicy<Integer> secondPolicy = new TimeTriggerPolicy<Integer>(10,
+ new TimestampWrapper<Integer>(timeStamp, 0));
+ TimeTriggerPolicy<Integer> thirdPolicy = new TimeTriggerPolicy<Integer>(22,
+ new TimestampWrapper<Integer>(timeStamp, 0));
+ @SuppressWarnings("unchecked")
+ MultiTriggerPolicy<Integer> multiTrigger = new MultiTriggerPolicy<Integer>(firstPolicy,
+ secondPolicy, thirdPolicy);
+
+ // expected result
+ // Long[][] result1 = { {}, {}, { 4L, 9L, 14L, 19L }, { 24L } };
+ // Long[][] result2 = { {}, {}, { 9L, 19L }, { } };
+ // Long[][] result3 = { {}, {}, { }, { 21L } };
+ Long[][] result = { {}, {}, { 4L, 9L, 14L, 19L, 9L, 19L }, { 24L, 21L } };
+
+ // call policy
+ for (int i = 0; i < times.length; i++) {
+ arrayEqualityCheck(result[i], multiTrigger.preNotifyTrigger(times[i]));
+ multiTrigger.notifyTrigger(times[i]);
+ }
+ }
+
+ /**
+ * This test verifies, that nestet active trigger runnables are started
+ * correctly.
+ */
+ @Test
+ public void testActiveTriggerRunnables() {
+ TriggerPolicy<Integer> firstPolicy = new ActiveTriggerWithRunnable(1);
+ TriggerPolicy<Integer> secondPolicy = new ActiveTriggerWithRunnable(2);
+ TriggerPolicy<Integer> thirdPolicy = new ActiveTriggerWithRunnable(3);
+ @SuppressWarnings("unchecked")
+ ActiveTriggerPolicy<Integer> multiTrigger = new MultiTriggerPolicy<Integer>(firstPolicy,
+ secondPolicy, thirdPolicy);
+
+ MyCallbackClass cb = new MyCallbackClass();
+ Runnable runnable = multiTrigger.createActiveTriggerRunnable(cb);
+ new Thread(runnable).start();
+
+ assertTrue("Even after 10000ms not all active policy runnables were started.",
+ cb.check(10000, 1, 2, 3));
+ }
+
+ private void arrayEqualityCheck(Object[] array1, Object[] array2) {
+ assertEquals(
+ "The result arrays must have the same length. (Expected: " + Arrays.asList(array1)
+ + "; Actual: " + Arrays.asList(array2) + ")", array1.length, array2.length);
+ for (int i = 0; i < array1.length; i++) {
+ assertEquals("Unequal fields at position " + i + "(Expected: " + Arrays.asList(array1)
+ + "; Actual: " + Arrays.asList(array2) + ")", array1[i], array2[i]);
+ }
+ }
+
+ /**
+ * This helper class is used to simulate active triggers which produce own
+ * runnables.
+ */
+ @SuppressWarnings("serial")
+ private class ActiveTriggerWithRunnable implements ActiveTriggerPolicy<Integer> {
+
+ int id;
+
+ public ActiveTriggerWithRunnable(int id) {
+ this.id = id;
+ }
+
+ @Override
+ public boolean notifyTrigger(Integer datapoint) {
+ // This method is not uses for any test case
+ return false;
+ }
+
+ @Override
+ public Object[] preNotifyTrigger(Integer datapoint) {
+ // This method is not used for any test case
+ return null;
+ }
+
+ @Override
+ public Runnable createActiveTriggerRunnable(final ActiveTriggerCallback callback) {
+ return new Runnable() {
+ @Override
+ public void run() {
+ callback.sendFakeElement(id);
+ }
+ };
+ }
+ }
+
+ /**
+ * This callback class is used to checked whether all nested policy runnable
+ * started up.
+ */
+ private class MyCallbackClass implements ActiveTriggerCallback {
+
+ List<Integer> received = new LinkedList<Integer>();
+
+ @Override
+ public void sendFakeElement(Object datapoint) {
+ received.add((Integer) datapoint);
+ }
+
+ public boolean check(int timeout, int... ids) {
+ int totalTime = 0;
+
+ while (totalTime <= timeout) {
+ boolean result = true;
+ for (int id : ids) {
+ if (!received.contains(id)) {
+ result = false;
+ }
+ }
+
+ if (result) {
+ return true;
+ } else {
+ try {
+ Thread.sleep(1000);
+ totalTime += 1000;
+ } catch (InterruptedException e) {
+ // ignore it here
+ }
+ }
+ }
+ return false;
+ }
+
+ }
+}
\ No newline at end of file
[3/6] flink git commit: [streaming] WindowBuffer separated from
Discretizer for operator sharing + equality checks added for windowing
objects
Posted by gy...@apache.org.
[streaming] WindowBuffer separated from Discretizer for operator sharing + equality checks added for windowing objects
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e5528a25
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e5528a25
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e5528a25
Branch: refs/heads/master
Commit: e5528a2535bd3ced460d7b8a2080330a70cca7d0
Parents: 6c01598
Author: Gyula Fora <gy...@apache.org>
Authored: Thu Feb 19 00:01:52 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Thu Feb 19 17:03:00 2015 +0100
----------------------------------------------------------------------
.../api/datastream/WindowedDataStream.java | 51 ++++++++++----
.../windowing/GroupedStreamDiscretizer.java | 32 ++++++---
.../windowing/GroupedTimeDiscretizer.java | 9 +--
.../windowing/GroupedWindowBufferInvokable.java | 69 ++++++++++++++++++
.../operator/windowing/StreamDiscretizer.java | 66 +++++++++++------
.../windowing/WindowBufferInvokable.java | 74 ++++++++++++++++++++
.../api/windowing/StreamWindowSerializer.java | 7 +-
.../streaming/api/windowing/WindowEvent.java | 71 +++++++++++++++++++
.../api/windowing/helper/TimestampWrapper.java | 16 +++++
.../windowing/policy/CountEvictionPolicy.java | 25 ++++++-
.../windowing/policy/CountTriggerPolicy.java | 17 +++++
.../api/windowing/policy/DeltaPolicy.java | 19 +++++
.../api/windowing/policy/PunctuationPolicy.java | 22 ++++++
.../windowing/policy/TimeEvictionPolicy.java | 16 +++++
.../api/windowing/policy/TimeTriggerPolicy.java | 16 +++++
.../policy/TumblingEvictionPolicy.java | 9 +++
.../windowing/GroupedStreamDiscretizerTest.java | 11 ++-
.../windowing/StreamDiscretizerTest.java | 18 +++--
.../policy/CountEvictionPolicyTest.java | 17 +++++
.../policy/CountTriggerPolicyTest.java | 19 ++++-
.../api/windowing/policy/DeltaPolicyTest.java | 33 +++++++--
.../windowing/policy/PunctuationPolicyTest.java | 33 ++++++++-
.../policy/TimeEvictionPolicyTest.java | 47 ++++++++++++-
.../windowing/policy/TimeTriggerPolicyTest.java | 39 +++++++++++
.../policy/TumblingEvictionPolicyTest.java | 2 +
25 files changed, 662 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/WindowedDataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/WindowedDataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/WindowedDataStream.java
index b63cb96..36551d2 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/WindowedDataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/WindowedDataStream.java
@@ -22,10 +22,12 @@ import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+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.java.functions.KeySelector;
import org.apache.flink.api.java.operators.Keys;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.streaming.api.function.RichWindowMapFunction;
import org.apache.flink.streaming.api.function.WindowMapFunction;
import org.apache.flink.streaming.api.function.aggregation.AggregationFunction;
@@ -35,9 +37,12 @@ import org.apache.flink.streaming.api.function.aggregation.SumAggregator;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.invokable.operator.windowing.GroupedStreamDiscretizer;
import org.apache.flink.streaming.api.invokable.operator.windowing.GroupedTimeDiscretizer;
+import org.apache.flink.streaming.api.invokable.operator.windowing.GroupedWindowBufferInvokable;
import org.apache.flink.streaming.api.invokable.operator.windowing.StreamDiscretizer;
+import org.apache.flink.streaming.api.invokable.operator.windowing.WindowBufferInvokable;
import org.apache.flink.streaming.api.windowing.StreamWindow;
import org.apache.flink.streaming.api.windowing.StreamWindowTypeInfo;
+import org.apache.flink.streaming.api.windowing.WindowEvent;
import org.apache.flink.streaming.api.windowing.helper.Time;
import org.apache.flink.streaming.api.windowing.helper.WindowingHelper;
import org.apache.flink.streaming.api.windowing.policy.CloneableEvictionPolicy;
@@ -316,14 +321,23 @@ public class WindowedDataStream<OUT> {
private DiscretizedStream<OUT> discretize(WindowTransformation transformation,
WindowBuffer<OUT> windowBuffer) {
- StreamInvokable<OUT, StreamWindow<OUT>> discretizer = getDiscretizer(transformation,
- windowBuffer, getTrigger(), getEviction(), discretizerKey);
+ StreamInvokable<OUT, WindowEvent<OUT>> discretizer = getDiscretizer(getTrigger(),
+ getEviction(), discretizerKey);
+
+ StreamInvokable<WindowEvent<OUT>, StreamWindow<OUT>> bufferInvokable = getBufferInvokable(
+ windowBuffer, discretizerKey);
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ TypeInformation<WindowEvent<OUT>> bufferEventType = new TupleTypeInfo(WindowEvent.class,
+ getType(), BasicTypeInfo.INT_TYPE_INFO);
int parallelism = getDiscretizerParallelism();
- return new DiscretizedStream<OUT>(dataStream.transform("Stream Discretizer",
- new StreamWindowTypeInfo<OUT>(getType()), discretizer).setParallelism(parallelism),
- groupByKey, transformation);
+ return new DiscretizedStream<OUT>(dataStream
+ .transform("Stream Discretizer", bufferEventType, discretizer)
+ .setParallelism(parallelism)
+ .transform("WindowBuffer", new StreamWindowTypeInfo<OUT>(getType()),
+ bufferInvokable).setParallelism(parallelism), groupByKey, transformation);
}
@@ -332,26 +346,33 @@ public class WindowedDataStream<OUT> {
.getDegreeOfParallelism() : 1;
}
- private StreamInvokable<OUT, StreamWindow<OUT>> getDiscretizer(
- WindowTransformation transformation, WindowBuffer<OUT> windowBuffer,
- TriggerPolicy<OUT> trigger, EvictionPolicy<OUT> eviction,
- KeySelector<OUT, ?> discretizerKey) {
+ private StreamInvokable<OUT, WindowEvent<OUT>> getDiscretizer(TriggerPolicy<OUT> trigger,
+ EvictionPolicy<OUT> eviction, KeySelector<OUT, ?> discretizerKey) {
if (discretizerKey == null) {
- return new StreamDiscretizer<OUT>(trigger, eviction, windowBuffer);
+ return new StreamDiscretizer<OUT>(trigger, eviction);
} else if (trigger instanceof TimeTriggerPolicy
&& ((TimeTriggerPolicy<OUT>) trigger).timestampWrapper.isDefaultTimestamp()) {
return new GroupedTimeDiscretizer<OUT>(discretizerKey,
- (TimeTriggerPolicy<OUT>) trigger, (CloneableEvictionPolicy<OUT>) eviction,
- windowBuffer);
- } else {
+ (TimeTriggerPolicy<OUT>) trigger, (CloneableEvictionPolicy<OUT>) eviction);
+ }
+
+ else {
return new GroupedStreamDiscretizer<OUT>(discretizerKey,
- (CloneableTriggerPolicy<OUT>) trigger, (CloneableEvictionPolicy<OUT>) eviction,
- windowBuffer);
+ (CloneableTriggerPolicy<OUT>) trigger, (CloneableEvictionPolicy<OUT>) eviction);
}
}
+ private StreamInvokable<WindowEvent<OUT>, StreamWindow<OUT>> getBufferInvokable(
+ WindowBuffer<OUT> windowBuffer, KeySelector<OUT, ?> discretizerKey) {
+ if (discretizerKey == null) {
+ return new WindowBufferInvokable<OUT>(windowBuffer);
+ } else {
+ return new GroupedWindowBufferInvokable<OUT>(windowBuffer, discretizerKey);
+ }
+ }
+
@SuppressWarnings("unchecked")
private WindowBuffer<OUT> getWindowBuffer(WindowTransformation transformation,
TriggerPolicy<OUT> trigger, EvictionPolicy<OUT> eviction,
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
index d6bbb08..ec01e28 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizer.java
@@ -22,7 +22,6 @@ import java.util.Map;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.windowing.StreamWindow;
import org.apache.flink.streaming.api.windowing.policy.CloneableEvictionPolicy;
import org.apache.flink.streaming.api.windowing.policy.CloneableTriggerPolicy;
@@ -35,7 +34,7 @@ import org.apache.flink.streaming.api.windowing.windowbuffer.WindowBuffer;
* transformed in the next stages. </p> To allow pre-aggregations supply an
* appropriate {@link WindowBuffer}
*/
-public class GroupedStreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>> {
+public class GroupedStreamDiscretizer<IN> extends StreamDiscretizer<IN> {
private static final long serialVersionUID = -3469545957144404137L;
@@ -43,15 +42,13 @@ public class GroupedStreamDiscretizer<IN> extends StreamInvokable<IN, StreamWind
protected Configuration parameters;
protected CloneableTriggerPolicy<IN> triggerPolicy;
protected CloneableEvictionPolicy<IN> evictionPolicy;
- protected WindowBuffer<IN> windowBuffer;
protected Map<Object, StreamDiscretizer<IN>> groupedDiscretizers;
public GroupedStreamDiscretizer(KeySelector<IN, ?> keySelector,
- CloneableTriggerPolicy<IN> triggerPolicy, CloneableEvictionPolicy<IN> evictionPolicy,
- WindowBuffer<IN> windowBuffer) {
+ CloneableTriggerPolicy<IN> triggerPolicy, CloneableEvictionPolicy<IN> evictionPolicy) {
- super(null);
+ super(triggerPolicy, evictionPolicy);
this.keySelector = keySelector;
@@ -59,8 +56,6 @@ public class GroupedStreamDiscretizer<IN> extends StreamInvokable<IN, StreamWind
this.evictionPolicy = evictionPolicy;
this.groupedDiscretizers = new HashMap<Object, StreamDiscretizer<IN>>();
- this.windowBuffer = windowBuffer;
-
}
@Override
@@ -103,7 +98,7 @@ public class GroupedStreamDiscretizer<IN> extends StreamInvokable<IN, StreamWind
protected StreamDiscretizer<IN> makeNewGroup(Object key) throws Exception {
StreamDiscretizer<IN> groupDiscretizer = new StreamDiscretizer<IN>(triggerPolicy.clone(),
- evictionPolicy.clone(), windowBuffer.clone());
+ evictionPolicy.clone());
groupDiscretizer.collector = taskContext.getOutputCollector();
groupDiscretizer.open(this.parameters);
@@ -111,4 +106,23 @@ public class GroupedStreamDiscretizer<IN> extends StreamInvokable<IN, StreamWind
return groupDiscretizer;
}
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof GroupedStreamDiscretizer)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ GroupedStreamDiscretizer<IN> otherDiscretizer = (GroupedStreamDiscretizer<IN>) other;
+
+ return triggerPolicy.equals(otherDiscretizer.triggerPolicy)
+ && evictionPolicy.equals(otherDiscretizer.evictionPolicy)
+ && keySelector.equals(otherDiscretizer.keySelector);
+
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedTimeDiscretizer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedTimeDiscretizer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedTimeDiscretizer.java
index a9f78b6..8a1168c 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedTimeDiscretizer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedTimeDiscretizer.java
@@ -20,7 +20,6 @@ package org.apache.flink.streaming.api.invokable.operator.windowing;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.windowing.policy.CloneableEvictionPolicy;
import org.apache.flink.streaming.api.windowing.policy.TimeTriggerPolicy;
-import org.apache.flink.streaming.api.windowing.windowbuffer.WindowBuffer;
/**
* A specialized {@link GroupedStreamDiscretizer} to be used with time only
@@ -34,10 +33,8 @@ public class GroupedTimeDiscretizer<IN> extends GroupedStreamDiscretizer<IN> {
private Thread policyThread;
public GroupedTimeDiscretizer(KeySelector<IN, ?> keySelector,
- TimeTriggerPolicy<IN> triggerPolicy, CloneableEvictionPolicy<IN> evictionPolicy,
- WindowBuffer<IN> windowBuffer) {
-
- super(keySelector, triggerPolicy, evictionPolicy, windowBuffer);
+ TimeTriggerPolicy<IN> triggerPolicy, CloneableEvictionPolicy<IN> evictionPolicy) {
+ super(keySelector, triggerPolicy, evictionPolicy);
this.timeTriggerPolicy = triggerPolicy;
}
@@ -45,7 +42,7 @@ public class GroupedTimeDiscretizer<IN> extends GroupedStreamDiscretizer<IN> {
protected StreamDiscretizer<IN> makeNewGroup(Object key) throws Exception {
StreamDiscretizer<IN> groupDiscretizer = new StreamDiscretizer<IN>(triggerPolicy.clone(),
- evictionPolicy.clone(), windowBuffer.clone());
+ evictionPolicy.clone());
groupDiscretizer.collector = taskContext.getOutputCollector();
// We omit the groupDiscretizer.open(...) call here to avoid starting
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedWindowBufferInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedWindowBufferInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedWindowBufferInvokable.java
new file mode 100644
index 0000000..2872f48
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedWindowBufferInvokable.java
@@ -0,0 +1,69 @@
+/*
+ * 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.invokable.operator.windowing;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.streaming.api.windowing.WindowEvent;
+import org.apache.flink.streaming.api.windowing.windowbuffer.WindowBuffer;
+
+/**
+ * This invokable flattens the results of the window transformations by
+ * outputing the elements of the {@link StreamWindow} one-by-one
+ */
+public class GroupedWindowBufferInvokable<T> extends WindowBufferInvokable<T> {
+
+ private static final long serialVersionUID = 1L;
+ private Map<Object, WindowBuffer<T>> windowMap = new HashMap<Object, WindowBuffer<T>>();
+ private KeySelector<T, ?> keySelector;
+
+ public GroupedWindowBufferInvokable(WindowBuffer<T> buffer, KeySelector<T, ?> keySelector) {
+ super(buffer);
+ this.keySelector = keySelector;
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ while (readNext() != null) {
+ callUserFunctionAndLogException();
+ }
+ }
+
+ @Override
+ protected void callUserFunction() throws Exception {
+ Object key = keySelector.getKey(nextObject.getElement());
+ WindowBuffer<T> currentWindow = windowMap.get(key);
+
+ if (currentWindow == null) {
+ currentWindow = buffer.clone();
+ windowMap.put(key, currentWindow);
+ }
+
+ handleWindowEvent(nextObject, currentWindow);
+ }
+
+ @Override
+ public void collect(WindowEvent<T> record) {
+ nextObject = record;
+ callUserFunctionAndLogException();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
index d053dd9..fb65283 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizer.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.api.invokable.operator.windowing;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.streaming.api.windowing.WindowEvent;
import org.apache.flink.streaming.api.windowing.policy.ActiveEvictionPolicy;
import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerCallback;
import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerPolicy;
@@ -32,7 +33,7 @@ import org.apache.flink.streaming.api.windowing.windowbuffer.WindowBuffer;
* {@link StreamWindow} that will be further transformed in the next stages.
* </p> To allow pre-aggregations supply an appropriate {@link WindowBuffer}
*/
-public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>> {
+public class StreamDiscretizer<IN> extends StreamInvokable<IN, WindowEvent<IN>> {
/**
* Auto-generated serial version UID
@@ -44,12 +45,11 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>>
private boolean isActiveTrigger;
private boolean isActiveEviction;
private Thread activePolicyThread;
- public int emptyCount = 0;
+ private int bufferSize = 0;
- protected WindowBuffer<IN> bufferHandler;
+ protected WindowEvent<IN> windowEvent = new WindowEvent<IN>();
- public StreamDiscretizer(TriggerPolicy<IN> triggerPolicy, EvictionPolicy<IN> evictionPolicy,
- WindowBuffer<IN> bufferHandler) {
+ public StreamDiscretizer(TriggerPolicy<IN> triggerPolicy, EvictionPolicy<IN> evictionPolicy) {
super(null);
this.triggerPolicy = triggerPolicy;
@@ -57,8 +57,14 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>>
this.isActiveTrigger = triggerPolicy instanceof ActiveTriggerPolicy;
this.isActiveEviction = evictionPolicy instanceof ActiveEvictionPolicy;
+ }
+
+ public TriggerPolicy<IN> getTrigger() {
+ return triggerPolicy;
+ }
- this.bufferHandler = bufferHandler;
+ public EvictionPolicy<IN> getEviction() {
+ return evictionPolicy;
}
@Override
@@ -105,7 +111,8 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>>
evict(input, isTriggered);
- bufferHandler.store(input);
+ collector.collect(windowEvent.setElement(input));
+ bufferSize++;
}
@@ -132,11 +139,7 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>>
* if not empty
*/
protected void emitWindow() {
- if (bufferHandler.emitWindow(collector)) {
- emptyCount = 0;
- } else {
- emptyCount++;
- }
+ collector.collect(windowEvent.setTrigger());
}
private void activeEvict(Object input) {
@@ -144,16 +147,24 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>>
if (isActiveEviction) {
ActiveEvictionPolicy<IN> ep = (ActiveEvictionPolicy<IN>) evictionPolicy;
- numToEvict = ep.notifyEvictionWithFakeElement(input, bufferHandler.size());
+ numToEvict = ep.notifyEvictionWithFakeElement(input, bufferSize);
}
- bufferHandler.evict(numToEvict);
+ if (numToEvict > 0) {
+ collector.collect(windowEvent.setEviction(numToEvict));
+ bufferSize -= numToEvict;
+ bufferSize = bufferSize >= 0 ? bufferSize : 0;
+ }
}
private void evict(IN input, boolean isTriggered) {
- int numToEvict = evictionPolicy.notifyEviction(input, isTriggered, bufferHandler.size());
+ int numToEvict = evictionPolicy.notifyEviction(input, isTriggered, bufferSize);
- bufferHandler.evict(numToEvict);
+ if (numToEvict > 0) {
+ collector.collect(windowEvent.setEviction(numToEvict));
+ bufferSize -= numToEvict;
+ bufferSize = bufferSize >= 0 ? bufferSize : 0;
+ }
}
@Override
@@ -171,11 +182,6 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>>
}
}
- @Override
- public String toString() {
- return bufferHandler.toString();
- }
-
/**
* This class allows the active trigger thread to call back and push fake
* elements at any time.
@@ -189,4 +195,22 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, StreamWindow<IN>>
}
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof StreamDiscretizer)
+ || (other instanceof GroupedStreamDiscretizer)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ StreamDiscretizer<IN> otherDiscretizer = (StreamDiscretizer<IN>) other;
+
+ return triggerPolicy.equals(otherDiscretizer.triggerPolicy)
+ && evictionPolicy.equals(otherDiscretizer.evictionPolicy);
+
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowBufferInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowBufferInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowBufferInvokable.java
new file mode 100644
index 0000000..ea4b830
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowBufferInvokable.java
@@ -0,0 +1,74 @@
+/*
+ * 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.invokable.operator.windowing;
+
+import org.apache.flink.streaming.api.invokable.ChainableInvokable;
+import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.streaming.api.windowing.WindowEvent;
+import org.apache.flink.streaming.api.windowing.windowbuffer.WindowBuffer;
+
+/**
+ * This invokable flattens the results of the window transformations by
+ * outputing the elements of the {@link StreamWindow} one-by-one
+ */
+public class WindowBufferInvokable<T> extends
+ ChainableInvokable<WindowEvent<T>, StreamWindow<T>> {
+
+ protected WindowBuffer<T> buffer;
+
+ public WindowBufferInvokable(WindowBuffer<T> buffer) {
+ super(null);
+ this.buffer = buffer;
+ }
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void invoke() throws Exception {
+ while (readNext() != null) {
+ callUserFunctionAndLogException();
+ }
+ }
+
+ @Override
+ protected void callUserFunction() throws Exception {
+ handleWindowEvent(nextObject);
+ }
+
+ protected void handleWindowEvent(WindowEvent<T> windowEvent, WindowBuffer<T> buffer)
+ throws Exception {
+ if (windowEvent.isElement()) {
+ buffer.store(windowEvent.getElement());
+ } else if (windowEvent.isEviction()) {
+ buffer.evict(windowEvent.getEviction());
+ } else if (windowEvent.isTrigger()) {
+ buffer.emitWindow(collector);
+ }
+ }
+
+ private void handleWindowEvent(WindowEvent<T> windowEvent) throws Exception {
+ handleWindowEvent(windowEvent, buffer);
+ }
+
+ @Override
+ public void collect(WindowEvent<T> record) {
+ nextObject = record;
+ callUserFunctionAndLogException();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindowSerializer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindowSerializer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindowSerializer.java
index 5b40e53..e8945f1 100755
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindowSerializer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindowSerializer.java
@@ -21,9 +21,9 @@ package org.apache.flink.streaming.api.windowing;
import java.io.IOException;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
@@ -32,11 +32,10 @@ public final class StreamWindowSerializer<T> extends TypeSerializer<StreamWindow
private static final long serialVersionUID = 1L;
private final TypeSerializer<T> typeSerializer;
- TypeSerializer<Integer> intSerializer;
+ TypeSerializer<Integer> intSerializer = IntSerializer.INSTANCE;
public StreamWindowSerializer(TypeInformation<T> typeInfo, ExecutionConfig conf) {
this.typeSerializer = typeInfo.createSerializer(conf);
- this.intSerializer = BasicTypeInfo.INT_TYPE_INFO.createSerializer(conf);
}
public TypeSerializer<T> getObjectSerializer() {
@@ -124,7 +123,7 @@ public final class StreamWindowSerializer<T> extends TypeSerializer<StreamWindow
@Override
public void copy(DataInputView source, DataOutputView target) throws IOException {
- // Needs to be implemented
+ serialize(deserialize(source), target);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowEvent.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowEvent.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowEvent.java
new file mode 100644
index 0000000..359dfb3
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowEvent.java
@@ -0,0 +1,71 @@
+/*
+ * 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.windowing;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+
+/**
+ * Type representing events sent to the window buffer. The first field should
+ * contain the window element the second field encodes triggers and evictions if
+ * the second field is greater than 0 it represents an eviction if it equals -1
+ * it represents a trigger.
+ */
+public class WindowEvent<T> extends Tuple2<T, Integer> {
+ private static final long serialVersionUID = 1L;
+
+ public boolean isElement() {
+ return f1 == 0;
+ }
+
+ public boolean isEviction() {
+ return f1 > 0;
+ }
+
+ public boolean isTrigger() {
+ return f1 == -1;
+ }
+
+ public Integer getEviction() {
+ return f1;
+ }
+
+ public T getElement() {
+ return f0;
+ }
+
+ public WindowEvent<T> setElement(T element) {
+ f0 = element;
+ f1 = 0;
+ return this;
+ }
+
+ public WindowEvent<T> setTrigger() {
+ f1 = -1;
+ return this;
+ }
+
+ public WindowEvent<T> setEviction(Integer n) {
+ if (n > 0) {
+ f1 = n;
+ return this;
+ } else {
+ throw new RuntimeException("Must evict at least 1");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/helper/TimestampWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/helper/TimestampWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/helper/TimestampWrapper.java
index 8c3a09d..a14b1c1 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/helper/TimestampWrapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/helper/TimestampWrapper.java
@@ -41,4 +41,20 @@ public class TimestampWrapper<T> implements Serializable {
public boolean isDefaultTimestamp() {
return timestamp instanceof SystemTimestamp;
}
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof TimestampWrapper)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ TimestampWrapper<T> otherTSW = (TimestampWrapper<T>) other;
+ return startTime == otherTSW.startTime
+ && timestamp.getClass() == otherTSW.timestamp.getClass();
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
index a9e7b3f..b4bdd43 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicy.java
@@ -34,9 +34,10 @@ public class CountEvictionPolicy<IN> implements CloneableEvictionPolicy<IN> {
*/
private static final long serialVersionUID = 2319201348806427996L;
- int maxElements;
- int counter;
- int deleteOnEviction = 1;
+ private int maxElements;
+ private int counter;
+ private int deleteOnEviction = 1;
+ private int startValue;
/**
* This constructor allows the setup of the simplest possible count based
@@ -94,6 +95,7 @@ public class CountEvictionPolicy<IN> implements CloneableEvictionPolicy<IN> {
this.counter = startValue;
this.deleteOnEviction = deleteOnEviction;
this.maxElements = maxElements;
+ this.startValue = startValue;
}
@Override
@@ -116,4 +118,21 @@ public class CountEvictionPolicy<IN> implements CloneableEvictionPolicy<IN> {
public CountEvictionPolicy<IN> clone() {
return new CountEvictionPolicy<IN>(maxElements, deleteOnEviction, counter);
}
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof CountEvictionPolicy)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ CountEvictionPolicy<IN> otherPolicy = (CountEvictionPolicy<IN>) other;
+ return startValue == otherPolicy.startValue
+ && deleteOnEviction == otherPolicy.deleteOnEviction
+ && maxElements == otherPolicy.maxElements;
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
index 7a3c75a..c814f5d 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicy.java
@@ -34,6 +34,7 @@ public class CountTriggerPolicy<IN> implements CloneableTriggerPolicy<IN> {
private int counter;
private int max;
+ private int startValue;
/**
* This constructor will set up a count based trigger, which triggers after
@@ -64,6 +65,7 @@ public class CountTriggerPolicy<IN> implements CloneableTriggerPolicy<IN> {
public CountTriggerPolicy(int max, int startValue) {
this.max = max;
this.counter = startValue;
+ this.startValue = startValue;
}
@Override
@@ -84,4 +86,19 @@ public class CountTriggerPolicy<IN> implements CloneableTriggerPolicy<IN> {
public CountTriggerPolicy<IN> clone() {
return new CountTriggerPolicy<IN>(max, counter);
}
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof CountTriggerPolicy)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ CountTriggerPolicy<IN> otherPolicy = (CountTriggerPolicy<IN>) other;
+ return max == otherPolicy.max && startValue == otherPolicy.startValue;
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
index 77bc692..169196f 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicy.java
@@ -51,6 +51,7 @@ public class DeltaPolicy<DATA> implements CloneableTriggerPolicy<DATA>,
private List<DATA> windowBuffer;
private double threshold;
private DATA triggerDataPoint;
+ private DATA init;
/**
* Crates a delta policy which calculates a delta between the data point
@@ -75,6 +76,7 @@ public class DeltaPolicy<DATA> implements CloneableTriggerPolicy<DATA>,
this.triggerDataPoint = init;
this.windowBuffer = new LinkedList<DATA>();
this.threshold = threshold;
+ this.init = init;
}
@Override
@@ -109,4 +111,21 @@ public class DeltaPolicy<DATA> implements CloneableTriggerPolicy<DATA>,
public DeltaPolicy<DATA> clone() {
return new DeltaPolicy<DATA>(deltaFuntion, triggerDataPoint, threshold);
}
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof DeltaPolicy)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ DeltaPolicy<DATA> otherPolicy = (DeltaPolicy<DATA>) other;
+ return threshold == otherPolicy.threshold
+ && deltaFuntion.getClass() == otherPolicy.deltaFuntion.getClass()
+ && init.equals(otherPolicy.init);
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
index ee8a3cb..4b06aee 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicy.java
@@ -113,4 +113,26 @@ public class PunctuationPolicy<IN, DATA> implements CloneableTriggerPolicy<IN>,
public PunctuationPolicy<IN, DATA> clone() {
return new PunctuationPolicy<IN, DATA>(punctuation, extractor);
}
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof PunctuationPolicy)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ PunctuationPolicy<IN, DATA> otherPolicy = (PunctuationPolicy<IN, DATA>) other;
+ if (extractor != null) {
+ return extractor.getClass() == otherPolicy.extractor.getClass()
+ && punctuation.equals(otherPolicy.punctuation);
+ } else {
+ return punctuation.equals(otherPolicy.punctuation)
+ && otherPolicy.extractor == null;
+ }
+
+ } catch (Exception e) {
+ return false;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
index 16c30fc..de715af 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicy.java
@@ -134,4 +134,20 @@ public class TimeEvictionPolicy<DATA> implements ActiveEvictionPolicy<DATA>,
return new TimeEvictionPolicy<DATA>(granularity, timestampWrapper);
}
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof TimeEvictionPolicy)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ TimeEvictionPolicy<DATA> otherPolicy = (TimeEvictionPolicy<DATA>) other;
+ return granularity == otherPolicy.granularity
+ && timestampWrapper.equals(otherPolicy.timestampWrapper);
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
index fb249cf..84eccf5 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicy.java
@@ -197,4 +197,20 @@ public class TimeTriggerPolicy<DATA> implements ActiveTriggerPolicy<DATA>,
return new TimeTriggerPolicy<DATA>(granularity, timestampWrapper, delay);
}
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof TimeTriggerPolicy)) {
+ return false;
+ } else {
+ try {
+ @SuppressWarnings("unchecked")
+ TimeTriggerPolicy<DATA> otherPolicy = (TimeTriggerPolicy<DATA>) other;
+ return startTime == otherPolicy.startTime && granularity == otherPolicy.granularity
+ && timestampWrapper.equals(otherPolicy.timestampWrapper);
+ } catch (ClassCastException e) {
+ return false;
+ }
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
index 9a2e3c9..8d9f028 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicy.java
@@ -87,4 +87,13 @@ public class TumblingEvictionPolicy<DATA> implements CloneableEvictionPolicy<DAT
public TumblingEvictionPolicy<DATA> clone() {
return new TumblingEvictionPolicy<DATA>(counter);
}
+
+ @Override
+ public boolean equals(Object other) {
+ if (other == null || !(other instanceof TumblingEvictionPolicy)) {
+ return false;
+ } else {
+ return true;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizerTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizerTest.java
index b514d2b..7049cb4 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizerTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/GroupedStreamDiscretizerTest.java
@@ -27,6 +27,7 @@ import java.util.Set;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.streaming.api.windowing.WindowEvent;
import org.apache.flink.streaming.api.windowing.policy.CloneableEvictionPolicy;
import org.apache.flink.streaming.api.windowing.policy.CloneableTriggerPolicy;
import org.apache.flink.streaming.api.windowing.policy.CountTriggerPolicy;
@@ -84,9 +85,15 @@ public class GroupedStreamDiscretizerTest {
CloneableEvictionPolicy<Integer> eviction = new TumblingEvictionPolicy<Integer>();
GroupedStreamDiscretizer<Integer> discretizer = new GroupedStreamDiscretizer<Integer>(
- keySelector, trigger, eviction, new BasicWindowBuffer<Integer>());
+ keySelector, trigger, eviction);
+
+ WindowBufferInvokable<Integer> buffer = new GroupedWindowBufferInvokable<Integer>(
+ new BasicWindowBuffer<Integer>(), keySelector);
+
+ List<WindowEvent<Integer>> bufferEvents = MockContext.createAndExecute(discretizer,
+ inputs);
+ List<StreamWindow<Integer>> result = MockContext.createAndExecute(buffer, bufferEvents);
- List<StreamWindow<Integer>> result = MockContext.createAndExecute(discretizer, inputs);
assertEquals(expected, new HashSet<StreamWindow<Integer>>(result));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizerTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizerTest.java
index ebc2677..0b7b51f 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizerTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/StreamDiscretizerTest.java
@@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.streaming.api.windowing.WindowEvent;
import org.apache.flink.streaming.api.windowing.helper.Timestamp;
import org.apache.flink.streaming.api.windowing.helper.TimestampWrapper;
import org.apache.flink.streaming.api.windowing.policy.CountTriggerPolicy;
@@ -75,11 +76,15 @@ public class StreamDiscretizerTest {
EvictionPolicy<Integer> eviction = new TimeEvictionPolicy<Integer>(4L,
new TimestampWrapper<Integer>(myTimeStamp, 1));
+
+
- StreamDiscretizer<Integer> discretizer = new StreamDiscretizer<Integer>(trigger, eviction,
- new BasicWindowBuffer<Integer>());
+ StreamDiscretizer<Integer> discretizer = new StreamDiscretizer<Integer>(trigger, eviction);
+ WindowBufferInvokable<Integer> buffer = new WindowBufferInvokable<Integer>(new BasicWindowBuffer<Integer>());
- List<StreamWindow<Integer>> result = MockContext.createAndExecute(discretizer, inputs);
+ List<WindowEvent<Integer>> bufferEvents = MockContext.createAndExecute(discretizer, inputs);
+ List<StreamWindow<Integer>> result = MockContext.createAndExecute(buffer, bufferEvents);
+
assertEquals(expected, result);
}
@@ -102,10 +107,11 @@ public class StreamDiscretizerTest {
EvictionPolicy<Integer> eviction = new TumblingEvictionPolicy<Integer>();
- StreamDiscretizer<Integer> discretizer = new StreamDiscretizer<Integer>(trigger, eviction,
- new BasicWindowBuffer<Integer>());
+ StreamDiscretizer<Integer> discretizer = new StreamDiscretizer<Integer>(trigger, eviction);
+ WindowBufferInvokable<Integer> buffer = new WindowBufferInvokable<Integer>(new BasicWindowBuffer<Integer>());
- List<StreamWindow<Integer>> result = MockContext.createAndExecute(discretizer, inputs);
+ List<WindowEvent<Integer>> bufferEvents = MockContext.createAndExecute(discretizer, inputs);
+ List<StreamWindow<Integer>> result = MockContext.createAndExecute(buffer, bufferEvents);
assertEquals(expected, result);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicyTest.java
index 4773f4e..8a7a011 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicyTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountEvictionPolicyTest.java
@@ -116,4 +116,21 @@ public class CountEvictionPolicyTest {
}
}
+ @Test
+ public void equalityTest() {
+ assertEquals(new CountEvictionPolicy<Integer>(5, 5, 5), new CountEvictionPolicy<Integer>(5,
+ 5, 5));
+
+ assertEquals(new CountEvictionPolicy<Integer>(5, 5), new CountEvictionPolicy<Integer>(5, 5));
+ assertEquals(new CountEvictionPolicy<Integer>(5), new CountEvictionPolicy<Integer>(5));
+
+ assertNotEquals(new CountEvictionPolicy<Integer>(4, 5, 5),
+ new CountEvictionPolicy<Integer>(5, 5, 5));
+ assertNotEquals(new CountEvictionPolicy<Integer>(5, 5, 5),
+ new CountEvictionPolicy<Integer>(5, 4, 5));
+
+ assertNotEquals(new CountEvictionPolicy<Integer>(5, 5, 5),
+ new CountEvictionPolicy<Integer>(5, 5, 4));
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicyTest.java
index b7120e7..ce5ae3b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicyTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/CountTriggerPolicyTest.java
@@ -17,7 +17,9 @@
package org.apache.flink.streaming.api.windowing.policy;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import java.util.Arrays;
@@ -38,8 +40,8 @@ public class CountTriggerPolicyTest {
// Test count of different sizes (0..9)
for (int i = 0; i < 10; i++) {
TriggerPolicy triggerPolicy = Count.of(i).toTrigger();
- counter=0;
-
+ counter = 0;
+
// Test first i steps (should not trigger)
for (int j = 0; j < i; j++) {
counter++;
@@ -91,4 +93,17 @@ public class CountTriggerPolicyTest {
}
}
}
+
+ @Test
+ public void equalityTest() {
+ assertEquals(new CountTriggerPolicy<Integer>(5, 5), new CountTriggerPolicy<Integer>(5, 5));
+
+ assertEquals(new CountTriggerPolicy<Integer>(5, 5), new CountTriggerPolicy<Integer>(5, 5));
+ assertEquals(new CountTriggerPolicy<Integer>(5), new CountTriggerPolicy<Integer>(5));
+
+ assertNotEquals(new CountTriggerPolicy<Integer>(4, 5),
+ new CountTriggerPolicy<Integer>(5, 5));
+ assertNotEquals(new CountTriggerPolicy<Integer>(5, 5),
+ new CountTriggerPolicy<Integer>(5, 4));
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicyTest.java
index 424eb57..52a4d13 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicyTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/DeltaPolicyTest.java
@@ -34,15 +34,13 @@ public class DeltaPolicyTest {
public void testDelta() {
DeltaPolicy deltaPolicy = new DeltaPolicy(new DeltaFunction<Tuple2<Integer, Integer>>() {
@Override
- public double getDelta(Tuple2<Integer, Integer> oldDataPoint, Tuple2<Integer, Integer> newDataPoint) {
+ public double getDelta(Tuple2<Integer, Integer> oldDataPoint,
+ Tuple2<Integer, Integer> newDataPoint) {
return (double) newDataPoint.f0 - oldDataPoint.f0;
}
}, new Tuple2(0, 0), 2);
- List<Tuple2> tuples = Arrays.asList(
- new Tuple2(1, 0),
- new Tuple2(2, 0),
- new Tuple2(3, 0),
+ List<Tuple2> tuples = Arrays.asList(new Tuple2(1, 0), new Tuple2(2, 0), new Tuple2(3, 0),
new Tuple2(6, 0));
assertFalse(deltaPolicy.notifyTrigger(tuples.get(0)));
@@ -58,5 +56,30 @@ public class DeltaPolicyTest {
assertEquals(2, deltaPolicy.notifyEviction(tuples.get(3), true, 2));
}
+ @Test
+ public void testEquality() {
+
+ DeltaFunction<Tuple2<Integer, Integer>> df = new DeltaFunction<Tuple2<Integer, Integer>>() {
+ private static final long serialVersionUID = 1L;
+ @Override
+ public double getDelta(Tuple2<Integer, Integer> oldDataPoint,
+ Tuple2<Integer, Integer> newDataPoint) {
+ return (double) newDataPoint.f0 - oldDataPoint.f0;
+ }
+ };
+
+ assertEquals(new DeltaPolicy<Tuple2<Integer, Integer>>(df, new Tuple2<Integer, Integer>(0,
+ 0), 2), new DeltaPolicy<Tuple2<Integer, Integer>>(df, new Tuple2<Integer, Integer>(
+ 0, 0), 2));
+
+ assertNotEquals(new DeltaPolicy<Tuple2<Integer, Integer>>(df, new Tuple2<Integer, Integer>(
+ 0, 1), 2), new DeltaPolicy<Tuple2<Integer, Integer>>(df,
+ new Tuple2<Integer, Integer>(0, 0), 2));
+
+ assertNotEquals(new DeltaPolicy<Tuple2<Integer, Integer>>(df, new Tuple2<Integer, Integer>(0,
+ 0), 2), new DeltaPolicy<Tuple2<Integer, Integer>>(df, new Tuple2<Integer, Integer>(
+ 0, 0), 3));
+
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicyTest.java
index 8224b12..89a3148 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicyTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/PunctuationPolicyTest.java
@@ -17,9 +17,13 @@
package org.apache.flink.streaming.api.windowing.policy;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.windowing.extractor.Extractor;
import org.apache.flink.streaming.api.windowing.extractor.FieldFromTuple;
import org.junit.Test;
@@ -98,6 +102,33 @@ public class PunctuationPolicyTest {
}
}
+ @Test
+ public void testEquals() {
+ Extractor<Integer, Integer> extractor = new Extractor<Integer, Integer>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Integer extract(Integer in) {
+ return in;
+ }
+ };
+
+ assertEquals(new PunctuationPolicy<Integer, Integer>(4),
+ new PunctuationPolicy<Integer, Integer>(4));
+ assertNotEquals(new PunctuationPolicy<Integer, Integer>(4),
+ new PunctuationPolicy<Integer, Integer>(5));
+
+ assertNotEquals(new PunctuationPolicy<Integer, Integer>(4, extractor),
+ new PunctuationPolicy<Integer, Integer>(4));
+
+ assertEquals(new PunctuationPolicy<Integer, Integer>(4, extractor),
+ new PunctuationPolicy<Integer, Integer>(4, extractor));
+
+ assertNotEquals(new PunctuationPolicy<Integer, Integer>(4),
+ new PunctuationPolicy<Integer, Integer>(4, extractor));
+
+ }
+
private class TestObject {
private int id;
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
index b5d502b..08a5c32 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeEvictionPolicyTest.java
@@ -17,14 +17,17 @@
package org.apache.flink.streaming.api.windowing.policy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
import java.util.LinkedList;
import org.apache.flink.streaming.api.windowing.helper.Timestamp;
import org.apache.flink.streaming.api.windowing.helper.TimestampWrapper;
import org.junit.Test;
-import static org.junit.Assert.*;
-
public class TimeEvictionPolicyTest {
@Test
@@ -118,4 +121,44 @@ public class TimeEvictionPolicyTest {
}
}
+ @Test
+ public void equalsTest() {
+
+ @SuppressWarnings("serial")
+ Timestamp<Integer> timeStamp = new Timestamp<Integer>() {
+
+ @Override
+ public long getTimestamp(Integer value) {
+ return value;
+ }
+
+ };
+
+ @SuppressWarnings("serial")
+ Timestamp<Integer> timeStamp2 = new Timestamp<Integer>() {
+
+ @Override
+ public long getTimestamp(Integer value) {
+ return value;
+ }
+
+ };
+
+ assertEquals(
+ new TimeEvictionPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp, 0)),
+ new TimeEvictionPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp, 0)));
+
+ assertNotEquals(new TimeEvictionPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp,
+ 0)), new TimeEvictionPolicy<Integer>(5,
+ new TimestampWrapper<Integer>(timeStamp2, 0)));
+
+ assertNotEquals(new TimeEvictionPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp,
+ 0)),
+ new TimeEvictionPolicy<Integer>(2, new TimestampWrapper<Integer>(timeStamp, 0)));
+
+ assertNotEquals(new TimeEvictionPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp,
+ 0)),
+ new TimeEvictionPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp, 3)));
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
index 2bdbd96..2fabbae 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TimeTriggerPolicyTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.api.windowing.policy;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -75,6 +76,44 @@ public class TimeTriggerPolicyTest {
}
@Test
+ public void equalsTest() {
+
+ @SuppressWarnings("serial")
+ Timestamp<Integer> timeStamp = new Timestamp<Integer>() {
+
+ @Override
+ public long getTimestamp(Integer value) {
+ return value;
+ }
+
+ };
+
+ @SuppressWarnings("serial")
+ Timestamp<Integer> timeStamp2 = new Timestamp<Integer>() {
+
+ @Override
+ public long getTimestamp(Integer value) {
+ return value;
+ }
+
+ };
+
+ assertEquals(
+ new TimeTriggerPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp, 0)),
+ new TimeTriggerPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp, 0)));
+
+ assertNotEquals(new TimeTriggerPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp,
+ 0)),
+ new TimeTriggerPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp2, 0)));
+
+ assertNotEquals(new TimeTriggerPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp,
+ 0)), new TimeTriggerPolicy<Integer>(2, new TimestampWrapper<Integer>(timeStamp, 0)));
+
+ assertNotEquals(new TimeTriggerPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp,
+ 0)), new TimeTriggerPolicy<Integer>(5, new TimestampWrapper<Integer>(timeStamp, 3)));
+ }
+
+ @Test
public void timeTriggerPreNotifyTest() {
// create some test data
Integer[] times = { 1, 3, 20, 26 };
http://git-wip-us.apache.org/repos/asf/flink/blob/e5528a25/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicyTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicyTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicyTest.java
index 8cb03a7..a3a7d73 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicyTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/policy/TumblingEvictionPolicyTest.java
@@ -36,6 +36,8 @@ public class TumblingEvictionPolicyTest {
assertEquals(counter, policy.notifyEviction(0, true, counter));
counter = 1;
}
+
+ assertEquals(new TumblingEvictionPolicy<Integer>(), new TumblingEvictionPolicy<Integer>());
}
}