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/03/20 13:41:33 UTC
[01/10] flink git commit: [streaming] Added Reader and Writer
factories & minor cleanup
Repository: flink
Updated Branches:
refs/heads/master 633b0d6a9 -> ac888b239
[streaming] Added Reader and Writer factories & minor cleanup
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/adcdfdac
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/adcdfdac
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/adcdfdac
Branch: refs/heads/master
Commit: adcdfdaccea20d3f6d6d0a7a85710ce6e9747072
Parents: d832400
Author: Gábor Hermann <re...@gmail.com>
Authored: Thu Mar 5 15:51:51 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:03 2015 +0100
----------------------------------------------------------------------
.../flink/streaming/api/StreamConfig.java | 8 +--
.../apache/flink/streaming/api/StreamEdge.java | 14 ++++-
.../flink/streaming/api/StreamEdgeList.java | 4 ++
.../api/StreamingJobGraphGenerator.java | 1 -
.../selector/DirectedOutputSelectorWrapper.java | 2 -
.../api/streamvertex/CoStreamVertex.java | 58 +++++++++-----------
.../api/streamvertex/InputHandler.java | 13 +++--
.../api/streamvertex/OutputHandler.java | 25 ++-------
.../flink/streaming/io/InputGateFactory.java | 42 ++++++++++++++
.../flink/streaming/io/RecordWriterFactory.java | 52 ++++++++++++++++++
10 files changed, 153 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
index f58f0ad..fc1f435 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
@@ -49,9 +49,7 @@ public class StreamConfig implements Serializable {
private static final String VERTEX_NAME = "vertexID";
private static final String OPERATOR_NAME = "operatorName";
private static final String ITERATION_ID = "iteration-id";
-// private static final String OUTPUT_SELECTOR = "outputSelector";
private static final String OUTPUT_SELECTOR_WRAPPER = "outputSelectorWrapper";
- private static final String DIRECTED_EMIT = "directedEmit";
private static final String SERIALIZEDUDF = "serializedudf";
private static final String USER_FUNCTION = "userfunction";
private static final String BUFFER_TIMEOUT = "bufferTimeout";
@@ -60,9 +58,7 @@ public class StreamConfig implements Serializable {
private static final String TYPE_SERIALIZER_OUT_1 = "typeSerializer_out_1";
private static final String TYPE_SERIALIZER_OUT_2 = "typeSerializer_out_2";
private static final String ITERATON_WAIT = "iterationWait";
- private static final String OUTPUTS = "outvertexIDs";
private static final String NONCHAINED_OUTPUTS = "NONCHAINED_OUTPUTS";
- private static final String CHAINED_OUTPUT_EDGES = "CHAINED_OUTPUTS";
private static final String EDGES_IN_ORDER = "rwOrder";
private static final String OUT_STREAM_EDGES = "out stream edges";
private static final String IN_STREAM_EDGES = "out stream edges";
@@ -247,7 +243,7 @@ public class StreamConfig implements Serializable {
SerializationUtils.serialize((Serializable) selected));
} else {
config.setBytes(OUTPUT_NAME + output,
- SerializationUtils.serialize((Serializable) new ArrayList<String>()));
+ SerializationUtils.serialize(new ArrayList<String>()));
}
}
@@ -305,6 +301,7 @@ public class StreamConfig implements Serializable {
config.setBytes(OUT_STREAM_EDGES, SerializationUtils.serialize((Serializable) outEdges));
}
+ @SuppressWarnings("unchecked")
public List<StreamEdge> getOutEdges(ClassLoader cl) {
try {
return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(
@@ -318,6 +315,7 @@ public class StreamConfig implements Serializable {
config.setBytes(IN_STREAM_EDGES, SerializationUtils.serialize((Serializable) inEdges));
}
+ @SuppressWarnings("unchecked")
public List<StreamEdge> getInEdges(ClassLoader cl) {
try {
return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
index 7363e08..a300fc2 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
@@ -22,13 +22,24 @@ import java.util.List;
import org.apache.flink.streaming.partitioner.StreamPartitioner;
+/**
+ * An edge in the streaming topology. One edge like this does not necessarily
+ * gets converted to a connection between two job vertices (due to chaining/optimization).
+ */
public class StreamEdge implements Serializable {
final private int sourceVertex;
final private int targetVertex;
+
+ /**
+ * The type number of the input for co-tasks.
+ */
final private int typeNumber;
+
+ /**
+ * A list of output names that the target vertex listens to (if there is output selection).
+ */
final private List<String> selectedNames;
-// private OutputSelector<?> outputSelector;
final private StreamPartitioner<?> outputPartitioner;
public StreamEdge(int sourceVertex, int targetVertex, int typeNumber, List<String> selectedNames, StreamPartitioner<?> outputPartitioner) {
@@ -36,7 +47,6 @@ public class StreamEdge implements Serializable {
this.targetVertex = targetVertex;
this.typeNumber = typeNumber;
this.selectedNames = selectedNames;
-// this.outputSelector = outputSelector;
this.outputPartitioner = outputPartitioner;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
index d15116b..d421079 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
@@ -23,6 +23,10 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
+/**
+ * Stores the stream topology with adjacency list graph representation.
+ * Edges are represented with {@link org.apache.flink.streaming.api.StreamEdge}s
+ */
public class StreamEdgeList {
private Map<Integer, List<StreamEdge>> outEdgeLists;
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/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 79d43c0..d420030 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
@@ -253,7 +253,6 @@ public class StreamingJobGraphGenerator {
headVertex.getConfiguration()) : chainedConfigs.get(headOfChain).get(
upStreamvertexID);
-// List<Integer> outEdgeIndexList = streamGraph.getOutEdgeTypes(upStreamvertexID);
int numOfInputs = downStreamConfig.getNumberOfInputs();
downStreamConfig.setInputIndex(numOfInputs++, streamGraph.getEdge(upStreamvertexID, downStreamvertexID).getTypeNumber());
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
index 624fac1..a04900c 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
@@ -37,11 +37,9 @@ public class DirectedOutputSelectorWrapper<OUT> implements OutputSelectorWrapper
private Map<String, List<Collector<OUT>>> outputMap;
private Set<Collector<OUT>> selectAllOutputs;
-// private Set<Collector<OUT>> emitted;
public DirectedOutputSelectorWrapper(List<OutputSelector<OUT>> outputSelectors) {
this.outputSelectors = outputSelectors;
-// this.emitted = new HashSet<Collector<OUT>>();
this.selectAllOutputs = new HashSet<Collector<OUT>>(); //new LinkedList<Collector<OUT>>();
this.outputMap = new HashMap<String, List<Collector<OUT>>>();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
index b919e0f..b45fc44 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
@@ -17,22 +17,21 @@
package org.apache.flink.streaming.api.streamvertex;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
-import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.streaming.api.StreamEdge;
-import org.apache.flink.streaming.api.invokable.operator.co.CoInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.io.CoReaderIterator;
import org.apache.flink.streaming.io.CoRecordReader;
import org.apache.flink.streaming.io.IndexedReaderIterator;
+import org.apache.flink.streaming.io.InputGateFactory;
import org.apache.flink.util.MutableObjectIterator;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
public class CoStreamVertex<IN1, IN2, OUT> extends StreamVertex<IN1, OUT> {
protected StreamRecordSerializer<IN1> inputDeserializer1 = null;
@@ -87,22 +86,19 @@ public class CoStreamVertex<IN1, IN2, OUT> extends StreamVertex<IN1, OUT> {
int inputType = inEdges.get(i).getTypeNumber();
InputGate reader = getEnvironment().getInputGate(i);
switch (inputType) {
- case 1:
- inputList1.add(reader);
- break;
- case 2:
- inputList2.add(reader);
- break;
- default:
- throw new RuntimeException("Invalid input type number: " + inputType);
+ case 1:
+ inputList1.add(reader);
+ break;
+ case 2:
+ inputList2.add(reader);
+ break;
+ default:
+ throw new RuntimeException("Invalid input type number: " + inputType);
}
}
- final InputGate reader1 = inputList1.size() == 1 ? inputList1.get(0) : new UnionInputGate(
- inputList1.toArray(new InputGate[inputList1.size()]));
-
- final InputGate reader2 = inputList2.size() == 1 ? inputList2.get(0) : new UnionInputGate(
- inputList2.toArray(new InputGate[inputList2.size()]));
+ final InputGate reader1 = InputGateFactory.createInputGate(inputList1);
+ final InputGate reader2 = InputGateFactory.createInputGate(inputList2);
coReader = new CoRecordReader<DeserializationDelegate<StreamRecord<IN1>>, DeserializationDelegate<StreamRecord<IN2>>>(
reader1, reader2);
@@ -112,12 +108,12 @@ public class CoStreamVertex<IN1, IN2, OUT> extends StreamVertex<IN1, OUT> {
@Override
public <X> MutableObjectIterator<X> getInput(int index) {
switch (index) {
- case 0:
- return (MutableObjectIterator<X>) inputIter1;
- case 1:
- return (MutableObjectIterator<X>) inputIter2;
- default:
- throw new IllegalArgumentException("CoStreamVertex has only 2 inputs");
+ case 0:
+ return (MutableObjectIterator<X>) inputIter1;
+ case 1:
+ return (MutableObjectIterator<X>) inputIter2;
+ default:
+ throw new IllegalArgumentException("CoStreamVertex has only 2 inputs");
}
}
@@ -130,12 +126,12 @@ public class CoStreamVertex<IN1, IN2, OUT> extends StreamVertex<IN1, OUT> {
@Override
public <X> StreamRecordSerializer<X> getInputSerializer(int index) {
switch (index) {
- case 0:
- return (StreamRecordSerializer<X>) inputDeserializer1;
- case 1:
- return (StreamRecordSerializer<X>) inputDeserializer2;
- default:
- throw new IllegalArgumentException("CoStreamVertex has only 2 inputs");
+ case 0:
+ return (StreamRecordSerializer<X>) inputDeserializer1;
+ case 1:
+ return (StreamRecordSerializer<X>) inputDeserializer2;
+ default:
+ throw new IllegalArgumentException("CoStreamVertex has only 2 inputs");
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
index 48a00d8..3988ec1 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
@@ -22,13 +22,13 @@ import java.io.IOException;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.io.network.api.reader.MutableReader;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
-import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.streaming.api.StreamConfig;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.io.IndexedMutableReader;
import org.apache.flink.streaming.io.IndexedReaderIterator;
+import org.apache.flink.streaming.io.InputGateFactory;
public class InputHandler<IN> {
private StreamRecordSerializer<IN> inputSerializer = null;
@@ -56,17 +56,20 @@ public class InputHandler<IN> {
int numberOfInputs = configuration.getNumberOfInputs();
if (numberOfInputs > 0) {
- InputGate inputGate = numberOfInputs < 2 ? streamVertex.getEnvironment()
- .getInputGate(0) : new UnionInputGate(streamVertex.getEnvironment()
- .getAllInputGates());
-
+ InputGate inputGate = InputGateFactory.createInputGate(streamVertex.getEnvironment().getAllInputGates());
inputs = new IndexedMutableReader<DeserializationDelegate<StreamRecord<IN>>>(inputGate);
+
inputs.registerTaskEventListener(streamVertex.getSuperstepListener(),
StreamingSuperstep.class);
inputIter = new IndexedReaderIterator<StreamRecord<IN>>(inputs, inputSerializer);
}
+ }
+ private IndexedReaderIterator<StreamRecord<IN>> createInputIterator() {
+ final IndexedReaderIterator<StreamRecord<IN>> iter = new IndexedReaderIterator<StreamRecord<IN>>(
+ inputs, inputSerializer);
+ return iter;
}
protected static <T> IndexedReaderIterator<StreamRecord<T>> staticCreateInputIterator(
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
index 18ddc79..bb66c63 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
@@ -26,6 +26,7 @@ import java.util.Map;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.StreamConfig;
import org.apache.flink.streaming.api.StreamEdge;
@@ -35,7 +36,7 @@ import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
import org.apache.flink.streaming.api.invokable.ChainableInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
-import org.apache.flink.streaming.io.StreamRecordWriter;
+import org.apache.flink.streaming.io.RecordWriterFactory;
import org.apache.flink.streaming.partitioner.StreamPartitioner;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
@@ -181,26 +182,10 @@ public class OutputHandler<OUT> {
StreamPartitioner<T> outputPartitioner = configuration.getPartitioner(cl, outputVertex);
- RecordWriter<SerializationDelegate<StreamRecord<T>>> output;
+ ResultPartitionWriter bufferWriter = vertex.getEnvironment().getWriter(outputIndex);
- if (configuration.getBufferTimeout() >= 0) {
-
- output = new StreamRecordWriter<SerializationDelegate<StreamRecord<T>>>(vertex
- .getEnvironment().getWriter(outputIndex), outputPartitioner,
- configuration.getBufferTimeout());
-
- if (LOG.isTraceEnabled()) {
- LOG.trace("StreamRecordWriter initiated with {} bufferTimeout for {}",
- configuration.getBufferTimeout(), vertex.getClass().getSimpleName());
- }
- } else {
- output = new RecordWriter<SerializationDelegate<StreamRecord<T>>>(vertex
- .getEnvironment().getWriter(outputIndex), outputPartitioner);
-
- if (LOG.isTraceEnabled()) {
- LOG.trace("RecordWriter initiated for {}", vertex.getClass().getSimpleName());
- }
- }
+ RecordWriter<SerializationDelegate<StreamRecord<T>>> output =
+ RecordWriterFactory.createRecordWriter(bufferWriter, outputPartitioner, configuration.getBufferTimeout());
StreamOutput<T> streamOutput = new StreamOutput<T>(output, vertex.instanceID,
outSerializationDelegate);
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/InputGateFactory.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/InputGateFactory.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/InputGateFactory.java
new file mode 100644
index 0000000..3e6edb9
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/InputGateFactory.java
@@ -0,0 +1,42 @@
+/*
+ * 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.io;
+
+import java.util.Collection;
+
+import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
+
+public class InputGateFactory {
+
+ public static InputGate createInputGate(Collection<InputGate> inputGates) {
+ return createInputGate(inputGates.toArray(new InputGate[inputGates.size()]));
+ }
+
+ public static InputGate createInputGate(InputGate[] inputGates) {
+ if (inputGates.length <= 0) {
+ throw new RuntimeException("No such input gate.");
+ }
+
+ if (inputGates.length < 2) {
+ return inputGates[0];
+ } else {
+ return new UnionInputGate(inputGates);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/adcdfdac/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/RecordWriterFactory.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/RecordWriterFactory.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/RecordWriterFactory.java
new file mode 100644
index 0000000..e859225
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/io/RecordWriterFactory.java
@@ -0,0 +1,52 @@
+/*
+ * 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.io;
+
+import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.io.network.api.writer.ChannelSelector;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RecordWriterFactory {
+ private static final Logger LOG = LoggerFactory.getLogger(RecordWriterFactory.class);
+
+ public static <OUT extends IOReadableWritable> RecordWriter<OUT> createRecordWriter(ResultPartitionWriter bufferWriter, ChannelSelector<OUT> channelSelector, long bufferTimeout) {
+
+ RecordWriter<OUT> output;
+
+ if (bufferTimeout >= 0) {
+ output = new StreamRecordWriter<OUT>(bufferWriter, channelSelector, bufferTimeout);
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("StreamRecordWriter initiated with {} bufferTimeout.", bufferTimeout);
+ }
+ } else {
+ output = new RecordWriter<OUT>(bufferWriter, channelSelector);
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("RecordWriter initiated.");
+ }
+ }
+
+ return output;
+
+ }
+
+}
[05/10] flink git commit: [FLINK-1594] [streaming] Added
OutputSelector wrapping
Posted by gy...@apache.org.
[FLINK-1594] [streaming] Added OutputSelector wrapping
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d832400b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d832400b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d832400b
Branch: refs/heads/master
Commit: d832400b3fee6c73553492e5954587958b4f7137
Parents: 3158d1d
Author: Gábor Hermann <re...@gmail.com>
Authored: Mon Mar 2 10:20:14 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:03 2015 +0100
----------------------------------------------------------------------
.../flink/streaming/api/StreamConfig.java | 32 ++---
.../apache/flink/streaming/api/StreamGraph.java | 8 +-
.../api/StreamingJobGraphGenerator.java | 2 +-
.../api/collector/CollectorWrapper.java | 17 ++-
.../api/collector/DirectedCollectorWrapper.java | 131 -------------------
.../streaming/api/collector/OutputSelector.java | 44 -------
.../BroadcastOutputSelectorWrapper.java | 43 ++++++
.../selector/DirectedOutputSelectorWrapper.java | 95 ++++++++++++++
.../api/collector/selector/OutputSelector.java | 44 +++++++
.../selector/OutputSelectorWrapper.java | 31 +++++
.../selector/OutputSelectorWrapperFactory.java | 32 +++++
.../streaming/api/datastream/DataStream.java | 6 +-
.../api/datastream/SplitDataStream.java | 2 +-
.../api/streamvertex/OutputHandler.java | 39 +++---
.../flink/streaming/api/OutputSplitterTest.java | 2 +-
.../api/collector/DirectedOutputTest.java | 1 +
.../api/collector/OutputSelectorTest.java | 1 +
.../examples/iteration/IterateExample.java | 2 +-
.../flink/streaming/api/scala/DataStream.scala | 2 +-
19 files changed, 293 insertions(+), 241 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
index 5b6de85..f58f0ad 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
@@ -27,7 +27,7 @@ import org.apache.commons.lang3.SerializationException;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.collector.OutputSelector;
+import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.api.streamvertex.StreamVertexException;
@@ -49,7 +49,8 @@ public class StreamConfig implements Serializable {
private static final String VERTEX_NAME = "vertexID";
private static final String OPERATOR_NAME = "operatorName";
private static final String ITERATION_ID = "iteration-id";
- private static final String OUTPUT_SELECTOR = "outputSelector";
+// private static final String OUTPUT_SELECTOR = "outputSelector";
+ private static final String OUTPUT_SELECTOR_WRAPPER = "outputSelectorWrapper";
private static final String DIRECTED_EMIT = "directedEmit";
private static final String SERIALIZEDUDF = "serializedudf";
private static final String USER_FUNCTION = "userfunction";
@@ -67,7 +68,6 @@ public class StreamConfig implements Serializable {
private static final String IN_STREAM_EDGES = "out stream edges";
// DEFAULT VALUES
-
private static final long DEFAULT_TIMEOUT = 100;
public static final String STATE_MONITORING = "STATE_MONITORING";
@@ -189,33 +189,21 @@ public class StreamConfig implements Serializable {
}
}
- public void setDirectedEmit(boolean directedEmit) {
- config.setBoolean(DIRECTED_EMIT, directedEmit);
- }
-
- public boolean isDirectedEmit() {
- return config.getBoolean(DIRECTED_EMIT, false);
- }
-
- public void setOutputSelectors(List<OutputSelector<?>> outputSelector) {
+ public void setOutputSelectorWrapper(OutputSelectorWrapper<?> outputSelectorWrapper) {
try {
- if (outputSelector != null && !outputSelector.isEmpty()) {
- setDirectedEmit(true);
- config.setBytes(OUTPUT_SELECTOR,
- SerializationUtils.serialize((Serializable) outputSelector));
- }
+ config.setBytes(OUTPUT_SELECTOR_WRAPPER, SerializationUtils.serialize(outputSelectorWrapper));
} catch (SerializationException e) {
- throw new RuntimeException("Cannot serialize OutputSelector");
+ throw new RuntimeException("Cannot serialize OutputSelectorWrapper");
}
}
@SuppressWarnings("unchecked")
- public <T> List<OutputSelector<T>> getOutputSelectors(ClassLoader cl) {
+ public <T> OutputSelectorWrapper<T> getOutputSelectorWrapper(ClassLoader cl) {
try {
- return (List<OutputSelector<T>>) InstantiationUtil.readObjectFromConfig(this.config,
- OUTPUT_SELECTOR, cl);
+ return (OutputSelectorWrapper<T>) InstantiationUtil.readObjectFromConfig(this.config,
+ OUTPUT_SELECTOR_WRAPPER, cl);
} catch (Exception e) {
- throw new StreamVertexException("Cannot deserialize and instantiate OutputSelector", e);
+ throw new StreamVertexException("Cannot deserialize and instantiate OutputSelectorWrapper", e);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/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 dfe66a5..948ea5e 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
@@ -37,7 +37,9 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.optimizer.plan.StreamingPlan;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.streaming.api.collector.OutputSelector;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapperFactory;
+import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.invokable.operator.co.CoInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
@@ -537,8 +539,8 @@ public class StreamGraph extends StreamingPlan {
return inputFormatLists.get(vertexID);
}
- public List<OutputSelector<?>> getOutputSelector(Integer vertexID) {
- return outputSelectors.get(vertexID);
+ public OutputSelectorWrapper<?> getOutputSelectorWrapper(Integer vertexID) {
+ return OutputSelectorWrapperFactory.create(outputSelectors.get(vertexID));
}
public Integer getIterationID(Integer vertexID) {
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/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 3ff64ca..79d43c0 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
@@ -213,7 +213,7 @@ public class StreamingJobGraphGenerator {
config.setTypeSerializerOut2(streamGraph.getOutSerializer2(vertexID));
config.setUserInvokable(streamGraph.getInvokable(vertexID));
- config.setOutputSelectors(streamGraph.getOutputSelector(vertexID));
+ config.setOutputSelectorWrapper(streamGraph.getOutputSelectorWrapper(vertexID));
config.setNumberOfOutputs(nonChainableOutputs.size());
config.setNonChainedOutputs(nonChainableOutputs);
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
index 1281bf0..bb0268a 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
@@ -17,27 +17,26 @@
package org.apache.flink.streaming.api.collector;
-import java.util.LinkedList;
-import java.util.List;
-
+import org.apache.flink.streaming.api.StreamEdge;
+import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
import org.apache.flink.util.Collector;
public class CollectorWrapper<OUT> implements Collector<OUT> {
- private List<Collector<OUT>> outputs;
+ private OutputSelectorWrapper<OUT> outputSelectorWrapper;
- public CollectorWrapper() {
- this.outputs = new LinkedList<Collector<OUT>>();
+ public CollectorWrapper(OutputSelectorWrapper<OUT> outputSelectorWrapper) {
+ this.outputSelectorWrapper = outputSelectorWrapper;
}
@SuppressWarnings("unchecked")
- public void addCollector(Collector<?> output) {
- outputs.add((Collector<OUT>) output);
+ public void addCollector(Collector<?> output, StreamEdge edge) {
+ outputSelectorWrapper.addCollector(output, edge);
}
@Override
public void collect(OUT record) {
- for(Collector<OUT> output: outputs){
+ for (Collector<OUT> output : outputSelectorWrapper.getSelectedOutputs(record)) {
output.collect(record);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedCollectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedCollectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedCollectorWrapper.java
deleted file mode 100755
index 4681cd3..0000000
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedCollectorWrapper.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.collector;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.flink.util.Collector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A StreamCollector that uses user defined output names and a user defined
- * output selector to make directed emits.
- *
- * @param <OUT>
- * Type of the Tuple collected.
- */
-public class DirectedCollectorWrapper<OUT> extends CollectorWrapper<OUT> {
-
- private static final Logger LOG = LoggerFactory.getLogger(DirectedCollectorWrapper.class);
-
- List<OutputSelector<OUT>> outputSelectors;
-
- protected Map<String, List<Collector<OUT>>> outputMap;
-
- private List<Collector<OUT>> selectAllOutputs;
- private Set<Collector<OUT>> emitted;
-
- /**
- * Creates a new DirectedStreamCollector
- *
- * @param outputSelector
- * User defined {@link OutputSelector}
- */
- public DirectedCollectorWrapper(List<OutputSelector<OUT>> outputSelectors) {
- this.outputSelectors = outputSelectors;
- this.emitted = new HashSet<Collector<OUT>>();
- this.selectAllOutputs = new LinkedList<Collector<OUT>>();
- this.outputMap = new HashMap<String, List<Collector<OUT>>>();
-
- }
-
- @Override
- public void addCollector(Collector<?> output) {
- addCollector(output, new ArrayList<String>());
- }
-
- @SuppressWarnings("unchecked")
- public void addCollector(Collector<?> output, List<String> selectedNames) {
-
- if (selectedNames.isEmpty()) {
- selectAllOutputs.add((Collector<OUT>) output);
- } else {
- for (String selectedName : selectedNames) {
-
- if (!outputMap.containsKey(selectedName)) {
- outputMap.put(selectedName, new LinkedList<Collector<OUT>>());
- outputMap.get(selectedName).add((Collector<OUT>) output);
- } else {
- if (!outputMap.get(selectedName).contains(output)) {
- outputMap.get(selectedName).add((Collector<OUT>) output);
- }
- }
-
- }
- }
- }
-
- @Override
- public void collect(OUT record) {
- emitted.clear();
-
- for (Collector<OUT> output : selectAllOutputs) {
- output.collect(record);
- emitted.add(output);
- }
-
- for (OutputSelector<OUT> outputSelector : outputSelectors) {
- Iterable<String> outputNames = outputSelector.select(record);
-
- for (String outputName : outputNames) {
- List<Collector<OUT>> outputList = outputMap.get(outputName);
- if (outputList == null) {
- if (LOG.isErrorEnabled()) {
- String format = String.format(
- "Cannot emit because no output is selected with the name: %s",
- outputName);
- LOG.error(format);
-
- }
- } else {
- for (Collector<OUT> output : outputList) {
- if (!emitted.contains(output)) {
- output.collect(record);
- emitted.add(output);
- }
- }
-
- }
-
- }
- }
-
- }
-
- @Override
- public void close() {
-
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/OutputSelector.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/OutputSelector.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/OutputSelector.java
deleted file mode 100644
index 6dbcff4..0000000
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/OutputSelector.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.collector;
-
-import java.io.Serializable;
-
-import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
-import org.apache.flink.streaming.api.datastream.SplitDataStream;
-
-/**
- * Interface for defining an OutputSelector for a {@link SplitDataStream} using
- * the {@link SingleOutputStreamOperator#split} call. Every output object of a
- * {@link SplitDataStream} will run through this operator to select outputs.
- *
- * @param <OUT>
- * Type parameter of the split values.
- */
-public interface OutputSelector<OUT> extends Serializable {
- /**
- * Method for selecting output names for the emitted objects when using the
- * {@link SingleOutputStreamOperator#split} method. The values will be
- * emitted only to output names which are contained in the returned
- * iterable.
- *
- * @param value
- * Output object for which the output selection should be made.
- */
- public Iterable<String> select(OUT value);
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
new file mode 100644
index 0000000..44371f0
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
@@ -0,0 +1,43 @@
+/*
+ * 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.collector.selector;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.streaming.api.StreamEdge;
+import org.apache.flink.util.Collector;
+
+public class BroadcastOutputSelectorWrapper<OUT> implements OutputSelectorWrapper<OUT> {
+
+ private List<Collector<OUT>> outputs;
+
+ public BroadcastOutputSelectorWrapper() {
+ outputs = new ArrayList<Collector<OUT>>();
+ }
+
+ @Override
+ public void addCollector(Collector<?> output, StreamEdge edge) {
+ outputs.add((Collector<OUT>) output);
+ }
+
+ @Override
+ public Iterable<Collector<OUT>> getSelectedOutputs(OUT record) {
+ return outputs;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
new file mode 100644
index 0000000..624fac1
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
@@ -0,0 +1,95 @@
+/*
+ * 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.collector.selector;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.flink.streaming.api.StreamEdge;
+import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DirectedOutputSelectorWrapper<OUT> implements OutputSelectorWrapper<OUT> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DirectedOutputSelectorWrapper.class);
+
+ private List<OutputSelector<OUT>> outputSelectors;
+
+ private Map<String, List<Collector<OUT>>> outputMap;
+ private Set<Collector<OUT>> selectAllOutputs;
+// private Set<Collector<OUT>> emitted;
+
+ public DirectedOutputSelectorWrapper(List<OutputSelector<OUT>> outputSelectors) {
+ this.outputSelectors = outputSelectors;
+// this.emitted = new HashSet<Collector<OUT>>();
+ this.selectAllOutputs = new HashSet<Collector<OUT>>(); //new LinkedList<Collector<OUT>>();
+ this.outputMap = new HashMap<String, List<Collector<OUT>>>();
+ }
+
+ @Override
+ public void addCollector(Collector<?> output, StreamEdge edge) {
+ List<String> selectedNames = edge.getSelectedNames();
+
+ if (selectedNames.isEmpty()) {
+ selectAllOutputs.add((Collector<OUT>) output);
+ } else {
+ for (String selectedName : selectedNames) {
+
+ if (!outputMap.containsKey(selectedName)) {
+ outputMap.put(selectedName, new LinkedList<Collector<OUT>>());
+ outputMap.get(selectedName).add((Collector<OUT>) output);
+ } else {
+ if (!outputMap.get(selectedName).contains(output)) {
+ outputMap.get(selectedName).add((Collector<OUT>) output);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public Iterable<Collector<OUT>> getSelectedOutputs(OUT record) {
+ Set<Collector<OUT>> selectedOutputs = new HashSet<Collector<OUT>>(selectAllOutputs);
+
+ for (OutputSelector<OUT> outputSelector : outputSelectors) {
+ Iterable<String> outputNames = outputSelector.select(record);
+
+ for (String outputName : outputNames) {
+ List<Collector<OUT>> outputList = outputMap.get(outputName);
+
+ try {
+ selectedOutputs.addAll(outputList);
+ } catch (NullPointerException e) {
+ if (LOG.isErrorEnabled()) {
+ String format = String.format(
+ "Cannot emit because no output is selected with the name: %s",
+ outputName);
+ LOG.error(format);
+ }
+ }
+ }
+ }
+
+ return selectedOutputs;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelector.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelector.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelector.java
new file mode 100644
index 0000000..b886fa6
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelector.java
@@ -0,0 +1,44 @@
+/*
+ * 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.collector.selector;
+
+import java.io.Serializable;
+
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.datastream.SplitDataStream;
+
+/**
+ * Interface for defining an OutputSelector for a {@link SplitDataStream} using
+ * the {@link SingleOutputStreamOperator#split} call. Every output object of a
+ * {@link SplitDataStream} will run through this operator to select outputs.
+ *
+ * @param <OUT>
+ * Type parameter of the split values.
+ */
+public interface OutputSelector<OUT> extends Serializable {
+ /**
+ * Method for selecting output names for the emitted objects when using the
+ * {@link SingleOutputStreamOperator#split} method. The values will be
+ * emitted only to output names which are contained in the returned
+ * iterable.
+ *
+ * @param value
+ * Output object for which the output selection should be made.
+ */
+ public Iterable<String> select(OUT value);
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java
new file mode 100644
index 0000000..850a1d9
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapper.java
@@ -0,0 +1,31 @@
+/*
+ * 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.collector.selector;
+
+import java.io.Serializable;
+
+import org.apache.flink.streaming.api.StreamEdge;
+import org.apache.flink.util.Collector;
+
+public interface OutputSelectorWrapper<OUT> extends Serializable {
+
+ public void addCollector(Collector<?> output, StreamEdge edge);
+
+ public Iterable<Collector<OUT>> getSelectedOutputs(OUT record);
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
new file mode 100644
index 0000000..c0f22c7
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
@@ -0,0 +1,32 @@
+/*
+ * 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.collector.selector;
+
+import java.util.List;
+
+public class OutputSelectorWrapperFactory {
+
+ public static OutputSelectorWrapper<?> create(List<OutputSelector<?>> outputSelectors) {
+ if (outputSelectors.size() == 0) {
+ return new BroadcastOutputSelectorWrapper();
+ } else {
+ return new DirectedOutputSelectorWrapper(outputSelectors);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index 879a98c..5f6f981 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -47,7 +47,7 @@ import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.StreamGraph;
-import org.apache.flink.streaming.api.collector.OutputSelector;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.temporaloperator.StreamCrossOperator;
import org.apache.flink.streaming.api.datastream.temporaloperator.StreamJoinOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -238,12 +238,12 @@ public class DataStream<OUT> {
/**
* Operator used for directing tuples to specific named outputs using an
- * {@link org.apache.flink.streaming.api.collector.OutputSelector}. Calling
+ * {@link org.apache.flink.streaming.api.collector.selector.OutputSelector}. Calling
* this method on an operator creates a new {@link SplitDataStream}.
*
* @param outputSelector
* The user defined
- * {@link org.apache.flink.streaming.api.collector.OutputSelector}
+ * {@link org.apache.flink.streaming.api.collector.selector.OutputSelector}
* for directing the tuples.
* @return The {@link SplitDataStream}
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/SplitDataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/SplitDataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/SplitDataStream.java
index 97458a8..69e059e 100755
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/SplitDataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/SplitDataStream.java
@@ -19,7 +19,7 @@ package org.apache.flink.streaming.api.datastream;
import java.util.Arrays;
-import org.apache.flink.streaming.api.collector.OutputSelector;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
/**
* The SplitDataStream represents an operator that has been split using an
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
index ca6b34d..18ddc79 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
@@ -30,8 +30,8 @@ import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.StreamConfig;
import org.apache.flink.streaming.api.StreamEdge;
import org.apache.flink.streaming.api.collector.CollectorWrapper;
-import org.apache.flink.streaming.api.collector.DirectedCollectorWrapper;
import org.apache.flink.streaming.api.collector.StreamOutput;
+import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
import org.apache.flink.streaming.api.invokable.ChainableInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
@@ -100,22 +100,22 @@ public class OutputHandler<OUT> {
* This method builds up a nested collector which encapsulates all the
* chained operators and their network output. The result of this recursive
* call will be passed as collector to the first invokable in the chain.
- *
+ *
* @param chainedTaskConfig
- * The configuration of the starting operator of the chain, we
- * use this paramater to recursively build the whole chain
+ * The configuration of the starting operator of the chain, we
+ * use this paramater to recursively build the whole chain
* @return Returns the collector for the chain starting from the given
- * config
+ * config
*/
- @SuppressWarnings({ "unchecked", "rawtypes" })
+ @SuppressWarnings({"unchecked", "rawtypes"})
private Collector<OUT> createChainedCollector(StreamConfig chainedTaskConfig) {
- boolean isDirectEmit = chainedTaskConfig.isDirectedEmit();
// We create a wrapper that will encapsulate the chained operators and
// network outputs
- CollectorWrapper<OUT> wrapper = isDirectEmit ? new DirectedCollectorWrapper(
- chainedTaskConfig.getOutputSelectors(cl)) : new CollectorWrapper<OUT>();
+
+ OutputSelectorWrapper<OUT> outputSelectorWrapper = chainedTaskConfig.getOutputSelectorWrapper(cl);
+ CollectorWrapper<OUT> wrapper = new CollectorWrapper<OUT>(outputSelectorWrapper);
// Create collectors for the network outputs
for (StreamEdge outputEdge : chainedTaskConfig.getNonChainedOutputs(cl)) {
@@ -123,12 +123,7 @@ public class OutputHandler<OUT> {
Collector<?> outCollector = outputMap.get(output);
- if (isDirectEmit) {
- ((DirectedCollectorWrapper<OUT>) wrapper).addCollector(outCollector,
- chainedTaskConfig.getSelectedNames(output));
- } else {
- wrapper.addCollector(outCollector);
- }
+ wrapper.addCollector(outCollector, outputEdge);
}
// Create collectors for the chained outputs
@@ -136,12 +131,8 @@ public class OutputHandler<OUT> {
Integer output = outputEdge.getTargetVertex();
Collector<?> outCollector = createChainedCollector(chainedConfigs.get(output));
- if (isDirectEmit) {
- ((DirectedCollectorWrapper<OUT>) wrapper).addCollector(outCollector,
- chainedTaskConfig.getSelectedNames(output));
- } else {
- wrapper.addCollector(outCollector);
- }
+
+ wrapper.addCollector(outCollector, outputEdge);
}
if (chainedTaskConfig.isChainStart()) {
@@ -169,11 +160,11 @@ public class OutputHandler<OUT> {
/**
* We create the StreamOutput for the specific output given by the id, and
* the configuration of its source task
- *
+ *
* @param outputVertex
- * Name of the output to which the streamoutput will be set up
+ * Name of the output to which the streamoutput will be set up
* @param configuration
- * The config of upStream task
+ * The config of upStream task
* @return
*/
private <T> StreamOutput<T> createStreamOutput(Integer outputVertex,
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
index cf6bb3c..14f0fa0 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
@@ -23,7 +23,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import org.apache.flink.streaming.api.collector.OutputSelector;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.TestListResultSink;
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
index ffc7c74..13bf457 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.SplitDataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.function.sink.SinkFunction;
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/OutputSelectorTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/OutputSelectorTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/OutputSelectorTest.java
index 1615a45..a3d89f2 100755
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/OutputSelectorTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/OutputSelectorTest.java
@@ -24,6 +24,7 @@ import java.util.Arrays;
import java.util.List;
import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.junit.Test;
public class OutputSelectorTest {
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
index 998e818..bbd5433 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
@@ -25,7 +25,7 @@ import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.collector.OutputSelector;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeDataStream;
import org.apache.flink.streaming.api.datastream.SplitDataStream;
http://git-wip-us.apache.org/repos/asf/flink/blob/d832400b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
index 59b1906..3dc54d6 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.scala
import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.streaming.api.collector.selector.OutputSelector
import org.apache.flink.streaming.api.datastream.{DataStream => JavaStream,
SingleOutputStreamOperator, GroupedDataStream}
import scala.collection.JavaConverters._
@@ -38,7 +39,6 @@ import org.apache.flink.streaming.api.function.sink.SinkFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import org.apache.flink.streaming.api.windowing.helper.WindowingHelper
import org.apache.flink.streaming.api.windowing.policy.{ EvictionPolicy, TriggerPolicy }
-import org.apache.flink.streaming.api.collector.OutputSelector
import scala.collection.JavaConversions._
import java.util.HashMap
import org.apache.flink.streaming.api.function.aggregation.SumFunction
[03/10] flink git commit: [FLINK-1594] [streaming] Embedded
StreamEdges
Posted by gy...@apache.org.
[FLINK-1594] [streaming] Embedded StreamEdges
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a8ba72b1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a8ba72b1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a8ba72b1
Branch: refs/heads/master
Commit: a8ba72b165dfd2b769783d5612b08559cbf24bf9
Parents: 29a6615
Author: Gábor Hermann <re...@gmail.com>
Authored: Thu Feb 26 16:15:17 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:03 2015 +0100
----------------------------------------------------------------------
.../apache/flink/streaming/api/StreamEdge.java | 2 +-
.../flink/streaming/api/StreamEdgeList.java | 48 ++++++-
.../apache/flink/streaming/api/StreamGraph.java | 125 ++++++++-----------
.../api/StreamingJobGraphGenerator.java | 20 +--
.../flink/streaming/api/WindowingOptimizer.java | 14 +--
5 files changed, 109 insertions(+), 100 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a8ba72b1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
index 8743233..479ae93 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
@@ -55,7 +55,7 @@ public class StreamEdge {
return selectedNames;
}
- public StreamPartitioner<?> getOutputPartitioner() {
+ public StreamPartitioner<?> getPartitioner() {
return outputPartitioner;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8ba72b1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
index 85202ab..d15116b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
@@ -90,23 +90,59 @@ public class StreamEdgeList {
}
}
- public List<StreamEdge> getOutEdges(int i) {
- List<StreamEdge> outEdges = outEdgeLists.get(i);
+ public StreamEdge getEdge(int sourceId, int targetId) {
+ Iterator<StreamEdge> outIterator = outEdgeLists.get(sourceId).iterator();
+ while (outIterator.hasNext()) {
+ StreamEdge edge = outIterator.next();
+
+ if (edge.getTargetVertex() == targetId) {
+ return edge;
+ }
+ }
+
+ throw new RuntimeException("No such edge in stream graph: " + sourceId + " -> " + targetId);
+ }
+
+ public List<StreamEdge> getOutEdges(int vertexId) {
+ List<StreamEdge> outEdges = outEdgeLists.get(vertexId);
if (outEdges == null) {
- throw new RuntimeException("No such vertex in stream graph: " + i);
+ throw new RuntimeException("No such vertex in stream graph: " + vertexId);
}
return outEdges;
}
- public List<StreamEdge> getInEdges(int i) {
- List<StreamEdge> inEdges = inEdgeLists.get(i);
+ public List<StreamEdge> getInEdges(int vertexId) {
+ List<StreamEdge> inEdges = inEdgeLists.get(vertexId);
if (inEdges == null) {
- throw new RuntimeException("No such vertex in stream graph: " + i);
+ throw new RuntimeException("No such vertex in stream graph: " + vertexId);
}
return inEdges;
}
+
+ public List<Integer> getOutEdgeIndices(int vertexId) {
+ List<StreamEdge> outEdges = getOutEdges(vertexId);
+ List<Integer> outEdgeIndices = new ArrayList<Integer>();
+
+ for (StreamEdge edge : outEdges) {
+ outEdgeIndices.add(edge.getTargetVertex());
+ }
+
+ return outEdgeIndices;
+ }
+
+ public List<Integer> getInEdgeIndices(int vertexId) {
+ List<StreamEdge> inEdges = getInEdges(vertexId);
+
+ List<Integer> inEdgeIndices = new ArrayList<Integer>();
+
+ for (StreamEdge edge : inEdges) {
+ inEdgeIndices.add(edge.getSourceVertex());
+ }
+
+ return inEdgeIndices;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/a8ba72b1/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 9f00c8e..dfe66a5 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
@@ -66,11 +66,11 @@ public class StreamGraph extends StreamingPlan {
// Graph attributes
private Map<Integer, Integer> operatorParallelisms;
private Map<Integer, Long> bufferTimeouts;
- private Map<Integer, List<Integer>> outEdgeLists;
- private Map<Integer, List<Integer>> outEdgeTypes;
- private Map<Integer, List<List<String>>> selectedNames;
- private Map<Integer, List<Integer>> inEdgeLists;
- private Map<Integer, List<StreamPartitioner<?>>> outputPartitioners;
+
+ private StreamEdgeList edges;
+
+ private Map<Integer, List<OutputSelector<?>>> outputSelectors;
+
private Map<Integer, String> operatorNames;
private Map<Integer, StreamInvokable<?, ?>> invokableObjects;
private Map<Integer, StreamRecordSerializer<?>> typeSerializersIn1;
@@ -78,7 +78,6 @@ public class StreamGraph extends StreamingPlan {
private Map<Integer, StreamRecordSerializer<?>> typeSerializersOut1;
private Map<Integer, StreamRecordSerializer<?>> typeSerializersOut2;
private Map<Integer, Class<? extends AbstractInvokable>> jobVertexClasses;
- private Map<Integer, List<OutputSelector<?>>> outputSelectors;
private Map<Integer, Integer> iterationIds;
private Map<Integer, Integer> iterationIDtoHeadID;
private Map<Integer, Integer> iterationIDtoTailID;
@@ -112,17 +111,9 @@ public class StreamGraph extends StreamingPlan {
operatorParallelisms = new HashMap<Integer, Integer>();
containingMaps.add(operatorParallelisms);
bufferTimeouts = new HashMap<Integer, Long>();
- containingMaps.add(bufferTimeouts);
- outEdgeLists = new HashMap<Integer, List<Integer>>();
- containingMaps.add(outEdgeLists);
- outEdgeTypes = new HashMap<Integer, List<Integer>>();
- containingMaps.add(outEdgeTypes);
- selectedNames = new HashMap<Integer, List<List<String>>>();
- containingMaps.add(selectedNames);
- inEdgeLists = new HashMap<Integer, List<Integer>>();
- containingMaps.add(inEdgeLists);
- outputPartitioners = new HashMap<Integer, List<StreamPartitioner<?>>>();
- containingMaps.add(outputPartitioners);
+
+ edges = new StreamEdgeList();
+
operatorNames = new HashMap<Integer, String>();
containingMaps.add(operatorNames);
invokableObjects = new HashMap<Integer, StreamInvokable<?, ?>>();
@@ -221,9 +212,10 @@ public class StreamGraph extends StreamingPlan {
setSerializersFrom(iterationHead, vertexID);
- setEdge(vertexID, iterationHead,
- outputPartitioners.get(inEdgeLists.get(iterationHead).get(0)).get(0), 0,
- new ArrayList<String>());
+ int outpartitionerIndexToCopy = edges.getInEdgeIndices(iterationHead).get(0);
+ StreamPartitioner<?> outputPartitioner = edges.getOutEdges(outpartitionerIndexToCopy).get(0).getPartitioner();
+
+ setEdge(vertexID, iterationHead, outputPartitioner, 0, new ArrayList<String>());
iterationTimeouts.put(iterationIDtoHeadID.get(iterationID), waitTime);
@@ -290,7 +282,7 @@ public class StreamGraph extends StreamingPlan {
/**
* Sets vertex parameters in the JobGraph
- *
+ *
* @param vertexID
* Name of the vertex
* @param vertexClass
@@ -307,12 +299,10 @@ public class StreamGraph extends StreamingPlan {
setParallelism(vertexID, parallelism);
invokableObjects.put(vertexID, invokableObject);
operatorNames.put(vertexID, operatorName);
- outEdgeLists.put(vertexID, new ArrayList<Integer>());
- outEdgeTypes.put(vertexID, new ArrayList<Integer>());
- selectedNames.put(vertexID, new ArrayList<List<String>>());
+
+ edges.addVertex(vertexID);
outputSelectors.put(vertexID, new ArrayList<OutputSelector<?>>());
- inEdgeLists.put(vertexID, new ArrayList<Integer>());
- outputPartitioners.put(vertexID, new ArrayList<StreamPartitioner<?>>());
+
iterationTailCount.put(vertexID, 0);
}
@@ -333,40 +323,21 @@ public class StreamGraph extends StreamingPlan {
*/
public void setEdge(Integer upStreamVertexID, Integer downStreamVertexID,
StreamPartitioner<?> partitionerObject, int typeNumber, List<String> outputNames) {
- outEdgeLists.get(upStreamVertexID).add(downStreamVertexID);
- outEdgeTypes.get(upStreamVertexID).add(typeNumber);
- inEdgeLists.get(downStreamVertexID).add(upStreamVertexID);
- outputPartitioners.get(upStreamVertexID).add(partitionerObject);
- selectedNames.get(upStreamVertexID).add(outputNames);
+
+ StreamEdge edge = new StreamEdge(upStreamVertexID, downStreamVertexID, typeNumber, outputNames, partitionerObject);
+ edges.addEdge(edge);
}
public void removeEdge(Integer upStream, Integer 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);
+ edges.removeEdge(upStream, downStream);
}
public void removeVertex(Integer toRemove) {
- List<Integer> outEdges = new ArrayList<Integer>(getOutEdges(toRemove));
- List<Integer> inEdges = new ArrayList<Integer>(getInEdges(toRemove));
-
- for (Integer output : outEdges) {
- removeEdge(toRemove, output);
- }
-
- for (Integer input : inEdges) {
- removeEdge(input, toRemove);
- }
+ edges.removeVertex(toRemove);
for (Map<Integer, ?> map : containingMaps) {
map.remove(toRemove);
}
-
}
private void addTypeSerializers(Integer vertexID, StreamRecordSerializer<?> in1,
@@ -418,11 +389,11 @@ public class StreamGraph extends StreamingPlan {
/**
* Sets a user defined {@link OutputSelector} for the given operator. Used
* for directed emits.
- *
+ *
* @param vertexID
- * Name of the vertex for which the output selector will be set
+ * Name of the vertex for which the output selector will be set
* @param outputSelector
- * The user defined output selector.
+ * The user defined output selector.
*/
public <T> void setOutputSelector(Integer vertexID, OutputSelector<T> outputSelector) {
outputSelectors.get(vertexID).add(outputSelector);
@@ -470,11 +441,11 @@ public class StreamGraph extends StreamingPlan {
/**
* Sets TypeSerializerWrapper from one vertex to another, used with some
* sinks.
- *
+ *
* @param from
- * from
+ * from
* @param to
- * to
+ * to
*/
public void setSerializersFrom(Integer from, Integer to) {
operatorNames.put(to, operatorNames.get(from));
@@ -495,9 +466,9 @@ public class StreamGraph extends StreamingPlan {
/**
* Gets the assembled {@link JobGraph} and adds a user specified name for
* it.
- *
+ *
* @param jobGraphName
- * name of the jobGraph
+ * name of the jobGraph
*/
public JobGraph getJobGraph(String jobGraphName) {
@@ -526,28 +497,24 @@ public class StreamGraph extends StreamingPlan {
return sources;
}
- public List<Integer> getOutEdges(Integer vertexID) {
- return outEdgeLists.get(vertexID);
+ public StreamEdge getEdge(Integer sourceId, Integer targetId) {
+ return edges.getEdge(sourceId, targetId);
}
- public List<Integer> getInEdges(Integer vertexID) {
- return inEdgeLists.get(vertexID);
+ public List<StreamEdge> getOutEdges(Integer vertexID) {
+ return edges.getOutEdges(vertexID);
}
- public List<Integer> getOutEdgeTypes(Integer vertexID) {
-
- return outEdgeTypes.get(vertexID);
+ public List<StreamEdge> getInEdges(Integer vertexID) {
+ return edges.getInEdges(vertexID);
}
- public StreamPartitioner<?> getOutPartitioner(Integer upStreamVertex, Integer downStreamVertex) {
- return outputPartitioners.get(upStreamVertex).get(
- outEdgeLists.get(upStreamVertex).indexOf(downStreamVertex));
+ public List<Integer> getOutEdgeIndices(Integer vertexID) {
+ return edges.getOutEdgeIndices(vertexID);
}
- public List<String> getSelectedNames(Integer upStreamVertex, Integer downStreamVertex) {
-
- return selectedNames.get(upStreamVertex).get(
- outEdgeLists.get(upStreamVertex).indexOf(downStreamVertex));
+ public List<Integer> getInEdgeIndices(Integer vertexID) {
+ return edges.getInEdgeIndices(vertexID);
}
public Collection<Integer> getIterationIDs() {
@@ -668,7 +635,9 @@ public class StreamGraph extends StreamingPlan {
JSONArray inputs = new JSONArray();
node.put(PREDECESSORS, inputs);
- for (int inputID : getInEdges(vertexID)) {
+ for (StreamEdge inEdge : getInEdges(vertexID)) {
+ int inputID = inEdge.getSourceVertex();
+
Integer mappedID = (edgeRemapings.keySet().contains(inputID)) ?
edgeRemapings.get(inputID) : inputID;
decorateEdge(inputs, vertexID, mappedID, inputID);
@@ -678,7 +647,9 @@ public class StreamGraph extends StreamingPlan {
toVisit.remove(vertexID);
} else {
Integer iterationHead = -1;
- for (int operator : getInEdges(vertexID)) {
+ for (StreamEdge inEdge : getInEdges(vertexID)) {
+ int operator = inEdge.getSourceVertex();
+
if (iterationIds.keySet().contains(operator)) {
iterationHead = operator;
}
@@ -718,7 +689,9 @@ public class StreamGraph extends StreamingPlan {
JSONArray inEdges = new JSONArray();
obj.put(PREDECESSORS, inEdges);
- for (int inputID : getInEdges(vertexID)) {
+ for (StreamEdge inEdge : getInEdges(vertexID)) {
+ int inputID = inEdge.getSourceVertex();
+
if (edgeRemapings.keySet().contains(inputID)) {
decorateEdge(inEdges, vertexID, inputID, inputID);
} else if (!iterationIds.containsKey(inputID)) {
@@ -737,7 +710,7 @@ public class StreamGraph extends StreamingPlan {
JSONObject input = new JSONObject();
inputArray.put(input);
input.put(ID, mappedInputID);
- input.put(SHIP_STRATEGY, getOutPartitioner(inputID, vertexID).getStrategy());
+ input.put(SHIP_STRATEGY, edges.getEdge(inputID, vertexID).getPartitioner().getStrategy());
input.put(SIDE, (inputArray.length() == 0) ? "first" : "second");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8ba72b1/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 ecb6455..607d041 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
@@ -104,11 +104,12 @@ public class StreamingJobGraphGenerator {
List<Integer> chainableOutputs = new ArrayList<Integer>();
List<Integer> nonChainableOutputs = new ArrayList<Integer>();
- for (Integer outName : streamGraph.getOutEdges(current)) {
- if (isChainable(current, outName)) {
- chainableOutputs.add(outName);
+ for (StreamEdge outEdge : streamGraph.getOutEdges(current)) {
+ Integer outID = outEdge.getTargetVertex();
+ if (isChainable(current, outID)) {
+ chainableOutputs.add(outID);
} else {
- nonChainableOutputs.add(outName);
+ nonChainableOutputs.add(outID);
}
}
@@ -230,7 +231,7 @@ public class StreamingJobGraphGenerator {
allOutputs.addAll(nonChainableOutputs);
for (Integer output : allOutputs) {
- config.setSelectedNames(output, streamGraph.getSelectedNames(vertexID, output));
+ config.setSelectedNames(output, streamGraph.getEdge(vertexID, output).getSelectedNames());
}
vertexConfigs.put(vertexID, config);
@@ -251,14 +252,13 @@ public class StreamingJobGraphGenerator {
headVertex.getConfiguration()) : chainedConfigs.get(headOfChain).get(
upStreamvertexID);
- List<Integer> outEdgeIndexList = streamGraph.getOutEdgeTypes(upStreamvertexID);
+// List<Integer> outEdgeIndexList = streamGraph.getOutEdgeTypes(upStreamvertexID);
int numOfInputs = downStreamConfig.getNumberOfInputs();
- downStreamConfig.setInputIndex(numOfInputs++, outEdgeIndexList.get(outputIndex));
+ downStreamConfig.setInputIndex(numOfInputs++, streamGraph.getEdge(upStreamvertexID, downStreamvertexID).getTypeNumber());
downStreamConfig.setNumberOfInputs(numOfInputs);
- StreamPartitioner<?> partitioner = streamGraph.getOutPartitioner(upStreamvertexID,
- downStreamvertexID);
+ StreamPartitioner<?> partitioner = streamGraph.getEdge(upStreamvertexID, downStreamvertexID).getPartitioner();
upStreamConfig.setPartitioner(downStreamvertexID, partitioner);
@@ -284,7 +284,7 @@ public class StreamingJobGraphGenerator {
&& outInvokable.getChainingStrategy() == ChainingStrategy.ALWAYS
&& (headInvokable.getChainingStrategy() == ChainingStrategy.HEAD || headInvokable
.getChainingStrategy() == ChainingStrategy.ALWAYS)
- && streamGraph.getOutPartitioner(vertexID, outName).getStrategy() == PartitioningStrategy.FORWARD
+ && streamGraph.getEdge(vertexID, outName).getPartitioner().getStrategy() == PartitioningStrategy.FORWARD
&& streamGraph.getParallelism(vertexID) == streamGraph.getParallelism(outName)
&& streamGraph.chaining;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a8ba72b1/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimizer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimizer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimizer.java
index e2cbc4b..3e98bda 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimizer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/WindowingOptimizer.java
@@ -54,13 +54,13 @@ public class WindowingOptimizer {
for (Integer flattener : flatteners) {
// Flatteners should have exactly one input
- Integer input = streamGraph.getInEdges(flattener).get(0);
+ Integer input = streamGraph.getInEdges(flattener).get(0).getSourceVertex();
// Check whether the flatten is applied after a merge
if (streamGraph.getInvokable(input) instanceof WindowMerger) {
// Mergers should have exactly one input
- Integer mergeInput = streamGraph.getInEdges(input).get(0);
+ Integer mergeInput = streamGraph.getInEdges(input).get(0).getSourceVertex();
streamGraph.setEdge(mergeInput, flattener, new DistributePartitioner(true), 0,
new ArrayList<String>());
@@ -97,9 +97,9 @@ public class WindowingOptimizer {
boolean inMatching = false;
for (Tuple2<StreamDiscretizer<?>, List<Integer>> matching : matchingDiscretizers) {
Set<Integer> discretizerInEdges = new HashSet<Integer>(
- streamGraph.getInEdges(discretizer.f0));
+ streamGraph.getInEdgeIndices(discretizer.f0));
Set<Integer> matchingInEdges = new HashSet<Integer>(
- streamGraph.getInEdges(matching.f1.get(0)));
+ streamGraph.getInEdgeIndices(matching.f1.get(0)));
if (discretizer.f1.equals(matching.f0)
&& discretizerInEdges.equals(matchingInEdges)) {
@@ -130,7 +130,7 @@ public class WindowingOptimizer {
private static void replaceDiscretizer(StreamGraph streamGraph, Integer toReplace,
Integer replaceWith) {
// Convert to array to create a copy
- List<Integer> outEdges = new ArrayList<Integer>(streamGraph.getOutEdges(toReplace));
+ List<Integer> outEdges = new ArrayList<Integer>(streamGraph.getOutEdgeIndices(toReplace));
int numOutputs = outEdges.size();
@@ -139,11 +139,11 @@ public class WindowingOptimizer {
Integer output = outEdges.get(i);
streamGraph.setEdge(replaceWith, output,
- streamGraph.getOutPartitioner(toReplace, output), 0, new ArrayList<String>());
+ streamGraph.getEdge(toReplace, output).getPartitioner(), 0, new ArrayList<String>());
streamGraph.removeEdge(toReplace, output);
}
- List<Integer> inEdges = new ArrayList<Integer>(streamGraph.getInEdges(toReplace));
+ List<Integer> inEdges = new ArrayList<Integer>(streamGraph.getInEdgeIndices(toReplace));
// Remove inputs
for (Integer input : inEdges) {
streamGraph.removeEdge(input, toReplace);
[10/10] flink git commit: [FLINK-1715] [streaming] Removed
unnecessary print messages
Posted by gy...@apache.org.
[FLINK-1715] [streaming] Removed unnecessary print messages
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ac888b23
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ac888b23
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ac888b23
Branch: refs/heads/master
Commit: ac888b239936d146b1109eca8b9a193458cdaf9f
Parents: 1377ca9
Author: Gyula Fora <gy...@apache.org>
Authored: Fri Mar 20 12:27:48 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 12:27:48 2015 +0100
----------------------------------------------------------------------
.../apache/flink/streaming/api/datastream/GroupedDataStream.java | 1 -
.../streaming/api/windowing/policy/MultiEvictionPolicy.java | 2 --
.../apache/flink/streaming/api/collector/DirectedOutputTest.java | 4 ++--
3 files changed, 2 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ac888b23/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/GroupedDataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/GroupedDataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/GroupedDataStream.java
index e3c22da..b565931 100755
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/GroupedDataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/GroupedDataStream.java
@@ -223,7 +223,6 @@ public class GroupedDataStream<OUT> extends DataStream<OUT> {
@Override
protected DataStream<OUT> setConnectionType(StreamPartitioner<OUT> partitioner) {
- System.out.println("Setting the partitioning after groupBy can affect the grouping");
return super.setConnectionType(partitioner);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ac888b23/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
index d353dc8..0ad1605 100644
--- 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
@@ -148,10 +148,8 @@ public class MultiEvictionPolicy<DATA> implements ActiveEvictionPolicy<DATA> {
case SUM:
result = 0;
for (Integer item : items) {
- System.out.print(item+":");
result += item;
}
- System.out.println(result);
return result;
case PRIORITY:
http://git-wip-us.apache.org/repos/asf/flink/blob/ac888b23/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
index 13bf457..62872fd 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
@@ -114,7 +114,7 @@ public class DirectedOutputTest {
assertEquals(Arrays.asList(2L, 4L, 6L, 8L, 10L), evenSink.getSortedResult());
assertEquals(Arrays.asList(1L, 3L, 5L, 7L, 9L, 10L, 11L), oddAndTenSink.getSortedResult());
assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L, 11L),
- evenAndOddSink.getResult());
- assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L, 11L), allSink.getResult());
+ evenAndOddSink.getSortedResult());
+ assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L, 11L), allSink.getSortedResult());
}
}
[04/10] flink git commit: [streaming] Added thread-safe list to tests
Posted by gy...@apache.org.
[streaming] Added thread-safe list to tests
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6b4ee2a1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6b4ee2a1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6b4ee2a1
Branch: refs/heads/master
Commit: 6b4ee2a12174adde5ca2badea70f4ebc35aa5ae3
Parents: 633b0d6
Author: Gábor Hermann <re...@gmail.com>
Authored: Fri Mar 13 09:59:09 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:03 2015 +0100
----------------------------------------------------------------------
.../flink/streaming/api/CoStreamTest.java | 97 +++++++++++++++++
.../flink/streaming/api/OutputSplitterTest.java | 106 +++++--------------
.../streaming/api/WindowCrossJoinTest.java | 67 +++++-------
.../api/collector/DirectedOutputTest.java | 39 +++----
.../streaming/util/TestListResultSink.java | 78 ++++++++++++++
.../flink/streaming/util/TestListWrapper.java | 60 +++++++++++
6 files changed, 309 insertions(+), 138 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6b4ee2a1/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java
new file mode 100644
index 0000000..20cd189
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.function.co.CoMapFunction;
+import org.apache.flink.streaming.util.TestListResultSink;
+import org.apache.flink.streaming.util.TestStreamEnvironment;
+
+public class CoStreamTest {
+
+ private static final long MEMORY_SIZE = 32;
+
+ private static ArrayList<String> expected;
+
+ public static void main(String[] args) throws InterruptedException {
+ for (int i = 0; i < 200; i++) {
+ test();
+ }
+ }
+
+ // @Test
+ public static void test() {
+ expected = new ArrayList<String>();
+
+ StreamExecutionEnvironment env = new TestStreamEnvironment(3, MEMORY_SIZE);
+
+ TestListResultSink<String> resultSink = new TestListResultSink<String>();
+
+ DataStream<Integer> src = env.fromElements(1, 3, 5);
+ DataStream<Integer> src2 = env.fromElements(1, 3, 5);
+
+ DataStream<Integer> grouped = src.groupBy(new KeySelector<Integer, Integer>() {
+ @Override
+ public Integer getKey(Integer value) throws Exception {
+ return value;
+ }
+ });
+
+ DataStream<Integer> grouped2 = src2.groupBy(new KeySelector<Integer, Integer>() {
+ @Override
+ public Integer getKey(Integer value) throws Exception {
+ return value;
+ }
+ });
+
+ DataStream<String> connected = grouped.connect(grouped2).map(new CoMapFunction<Integer, Integer, String>() {
+ @Override
+ public String map1(Integer value) {
+ return value.toString();
+ }
+
+ @Override
+ public String map2(Integer value) {
+ return value.toString();
+ }
+ });
+
+ connected.addSink(resultSink);
+
+ connected.print();
+
+ try {
+ env.execute();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ expected = new ArrayList<String>();
+ expected.addAll(Arrays.asList("1", "1", "3", "3", "5", "5"));
+
+ System.out.println(resultSink.getResult());
+ assertEquals(expected, expected);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/6b4ee2a1/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
index a214fbf..cf6bb3c 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/OutputSplitterTest.java
@@ -19,39 +19,34 @@ package org.apache.flink.streaming.api;
import static org.junit.Assert.assertEquals;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
import org.apache.flink.streaming.api.collector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.sink.SinkFunction;
+import org.apache.flink.streaming.util.TestListResultSink;
import org.apache.flink.streaming.util.TestStreamEnvironment;
import org.junit.Test;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-
public class OutputSplitterTest {
private static final long MEMORYSIZE = 32;
- private static ArrayList<Integer> splitterResult1 = new ArrayList<Integer>();
- private static ArrayList<Integer> splitterResult2 = new ArrayList<Integer>();
-
-
private static ArrayList<Integer> expectedSplitterResult = new ArrayList<Integer>();
@SuppressWarnings("unchecked")
@Test
public void testOnMergedDataStream() throws Exception {
- splitterResult1.clear();
- splitterResult2.clear();
+ TestListResultSink<Integer> splitterResultSink1 = new TestListResultSink<Integer>();
+ TestListResultSink<Integer> splitterResultSink2 = new TestListResultSink<Integer>();
StreamExecutionEnvironment env = new TestStreamEnvironment(1, MEMORYSIZE);
env.setBufferTimeout(1);
- DataStream<Integer> d1 = env.fromElements(0,2,4,6,8);
- DataStream<Integer> d2 = env.fromElements(1,3,5,7,9);
+ DataStream<Integer> d1 = env.fromElements(0, 2, 4, 6, 8);
+ DataStream<Integer> d2 = env.fromElements(1, 3, 5, 7, 9);
d1 = d1.merge(d2);
@@ -68,19 +63,7 @@ public class OutputSplitterTest {
}
return s;
}
- }).select(">").addSink(new SinkFunction<Integer>() {
-
- private static final long serialVersionUID = 5827187510526388104L;
-
- @Override
- public void invoke(Integer value) {
- splitterResult1.add(value);
- }
-
- @Override
- public void cancel() {
- }
- });
+ }).select(">").addSink(splitterResultSink1);
d1.split(new OutputSelector<Integer>() {
private static final long serialVersionUID = -6822487543355994807L;
@@ -95,41 +78,27 @@ public class OutputSplitterTest {
}
return s;
}
- }).select("yes").addSink(new SinkFunction<Integer>() {
- private static final long serialVersionUID = -2674335071267854599L;
-
- @Override
- public void invoke(Integer value) {
- splitterResult2.add(value);
- }
-
- @Override
- public void cancel() {
- }
- });
+ }).select("yes").addSink(splitterResultSink2);
env.execute();
- Collections.sort(splitterResult1);
- Collections.sort(splitterResult2);
-
expectedSplitterResult.clear();
- expectedSplitterResult.addAll(Arrays.asList(5,6,7,8,9));
- assertEquals(expectedSplitterResult, splitterResult1);
+ expectedSplitterResult.addAll(Arrays.asList(5, 6, 7, 8, 9));
+ assertEquals(expectedSplitterResult, splitterResultSink1.getSortedResult());
expectedSplitterResult.clear();
- expectedSplitterResult.addAll(Arrays.asList(0,3,6,9));
- assertEquals(expectedSplitterResult, splitterResult2);
+ expectedSplitterResult.addAll(Arrays.asList(0, 3, 6, 9));
+ assertEquals(expectedSplitterResult, splitterResultSink2.getSortedResult());
}
@Test
public void testOnSingleDataStream() throws Exception {
- splitterResult1.clear();
- splitterResult2.clear();
+ TestListResultSink<Integer> splitterResultSink1 = new TestListResultSink<Integer>();
+ TestListResultSink<Integer> splitterResultSink2 = new TestListResultSink<Integer>();
StreamExecutionEnvironment env = new TestStreamEnvironment(1, MEMORYSIZE);
env.setBufferTimeout(1);
- DataStream<Integer> ds = env.fromElements(0,1,2,3,4,5,6,7,8,9);
+ DataStream<Integer> ds = env.fromElements(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
ds.split(new OutputSelector<Integer>() {
private static final long serialVersionUID = 2524335410904414121L;
@@ -144,19 +113,7 @@ public class OutputSplitterTest {
}
return s;
}
- }).select("even").addSink(new SinkFunction<Integer>() {
-
- private static final long serialVersionUID = -2995092337537209535L;
-
- @Override
- public void invoke(Integer value) {
- splitterResult1.add(value);
- }
-
- @Override
- public void cancel() {
- }
- });
+ }).select("even").addSink(splitterResultSink1);
ds.split(new OutputSelector<Integer>() {
@@ -172,30 +129,15 @@ public class OutputSplitterTest {
}
return s;
}
- }).select("yes").addSink(new SinkFunction<Integer>() {
-
- private static final long serialVersionUID = -1749077049727705424L;
-
- @Override
- public void invoke(Integer value) {
- splitterResult2.add(value);
- }
-
- @Override
- public void cancel() {
- }
- });
+ }).select("yes").addSink(splitterResultSink2);
env.execute();
- Collections.sort(splitterResult1);
- Collections.sort(splitterResult2);
-
expectedSplitterResult.clear();
- expectedSplitterResult.addAll(Arrays.asList(0,2,4,6,8));
- assertEquals(expectedSplitterResult, splitterResult1);
+ expectedSplitterResult.addAll(Arrays.asList(0, 2, 4, 6, 8));
+ assertEquals(expectedSplitterResult, splitterResultSink1.getSortedResult());
expectedSplitterResult.clear();
- expectedSplitterResult.addAll(Arrays.asList(0,4,8));
- assertEquals(expectedSplitterResult, splitterResult2);
+ expectedSplitterResult.addAll(Arrays.asList(0, 4, 8));
+ assertEquals(expectedSplitterResult, splitterResultSink2.getSortedResult());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b4ee2a1/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/WindowCrossJoinTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/WindowCrossJoinTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/WindowCrossJoinTest.java
index bd97917..cccec40 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/WindowCrossJoinTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/WindowCrossJoinTest.java
@@ -24,12 +24,13 @@ import java.util.ArrayList;
import java.util.HashSet;
import org.apache.flink.api.common.functions.CrossFunction;
+import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.helper.Timestamp;
+import org.apache.flink.streaming.util.TestListResultSink;
import org.apache.flink.streaming.util.TestStreamEnvironment;
import org.junit.Test;
@@ -39,17 +40,29 @@ public class WindowCrossJoinTest implements Serializable {
private static final long MEMORYSIZE = 32;
- private static ArrayList<Tuple2<Tuple2<Integer, String>, Integer>> joinResults = new ArrayList<Tuple2<Tuple2<Integer, String>, Integer>>();
private static ArrayList<Tuple2<Tuple2<Integer, String>, Integer>> joinExpectedResults = new ArrayList<Tuple2<Tuple2<Integer, String>, Integer>>();
-
- private static ArrayList<Tuple2<Tuple2<Integer, String>, Integer>> crossResults = new ArrayList<Tuple2<Tuple2<Integer, String>, Integer>>();
private static ArrayList<Tuple2<Tuple2<Integer, String>, Integer>> crossExpectedResults = new ArrayList<Tuple2<Tuple2<Integer, String>, Integer>>();
+ private static class MyTimestamp<T> implements Timestamp<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public long getTimestamp(T value) {
+ return 101L;
+ }
+ }
+
@Test
public void test() throws Exception {
StreamExecutionEnvironment env = new TestStreamEnvironment(1, MEMORYSIZE);
env.setBufferTimeout(1);
+ TestListResultSink<Tuple2<Tuple2<Integer, String>, Integer>> joinResultSink =
+ new TestListResultSink<Tuple2<Tuple2<Integer, String>, Integer>>();
+ TestListResultSink<Tuple2<Tuple2<Integer, String>, Integer>> crossResultSink =
+ new TestListResultSink<Tuple2<Tuple2<Integer, String>, Integer>>();
+
ArrayList<Tuple2<Integer, String>> in1 = new ArrayList<Tuple2<Integer, String>>();
ArrayList<Tuple1<Integer>> in2 = new ArrayList<Tuple1<Integer>>();
@@ -101,7 +114,8 @@ public class WindowCrossJoinTest implements Serializable {
.join(inStream2)
.onWindow(1000, new MyTimestamp<Tuple2<Integer, String>>(),
new MyTimestamp<Tuple1<Integer>>(), 100).where(0).equalTo(0)
- .addSink(new JoinResultSink());
+ .map(new ResultMap())
+ .addSink(joinResultSink);
inStream1
.cross(inStream2)
@@ -116,50 +130,27 @@ public class WindowCrossJoinTest implements Serializable {
Tuple2<Integer, String> val1, Tuple1<Integer> val2) throws Exception {
return new Tuple2<Tuple2<Integer, String>, Tuple1<Integer>>(val1, val2);
}
- }).addSink(new CrossResultSink());
+ })
+ .map(new ResultMap())
+ .addSink(crossResultSink);
env.execute();
assertEquals(new HashSet<Tuple2<Tuple2<Integer, String>, Integer>>(joinExpectedResults),
- new HashSet<Tuple2<Tuple2<Integer, String>, Integer>>(joinResults));
+ new HashSet<Tuple2<Tuple2<Integer, String>, Integer>>(joinResultSink.getResult()));
assertEquals(new HashSet<Tuple2<Tuple2<Integer, String>, Integer>>(crossExpectedResults),
- new HashSet<Tuple2<Tuple2<Integer, String>, Integer>>(crossResults));
- }
-
- private static class MyTimestamp<T> implements Timestamp<T> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public long getTimestamp(T value) {
- return 101L;
- }
+ new HashSet<Tuple2<Tuple2<Integer, String>, Integer>>(crossResultSink.getResult()));
}
- private static class JoinResultSink implements
- SinkFunction<Tuple2<Tuple2<Integer, String>, Tuple1<Integer>>> {
+ private static class ResultMap implements
+ MapFunction<Tuple2<Tuple2<Integer, String>, Tuple1<Integer>>,
+ Tuple2<Tuple2<Integer, String>, Integer>> {
private static final long serialVersionUID = 1L;
@Override
- public void invoke(Tuple2<Tuple2<Integer, String>, Tuple1<Integer>> value) {
- joinResults.add(new Tuple2<Tuple2<Integer, String>, Integer>(value.f0, value.f1.f0));
- }
-
- @Override
- public void cancel() {
+ public Tuple2<Tuple2<Integer, String>, Integer> map(Tuple2<Tuple2<Integer, String>, Tuple1<Integer>> value) throws Exception {
+ return new Tuple2<Tuple2<Integer, String>, Integer>(value.f0, value.f1.f0);
}
}
- private static class CrossResultSink implements
- SinkFunction<Tuple2<Tuple2<Integer, String>, Tuple1<Integer>>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void invoke(Tuple2<Tuple2<Integer, String>, Tuple1<Integer>> value) {
- crossResults.add(new Tuple2<Tuple2<Integer, String>, Integer>(value.f0, value.f1.f0));
- }
-
- @Override
- public void cancel() {
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b4ee2a1/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
index 9d166e5..ffc7c74 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/collector/DirectedOutputTest.java
@@ -17,11 +17,7 @@
package org.apache.flink.streaming.api.collector;
-import org.apache.flink.streaming.api.datastream.SplitDataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.sink.SinkFunction;
-import org.apache.flink.streaming.util.TestStreamEnvironment;
-import org.junit.Test;
+import static org.junit.Assert.assertEquals;
import java.io.IOException;
import java.util.ArrayList;
@@ -30,15 +26,17 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.junit.Assert.assertEquals;
+import org.apache.flink.streaming.api.datastream.SplitDataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.function.sink.SinkFunction;
+import org.apache.flink.streaming.util.TestListResultSink;
+import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.junit.Test;
public class DirectedOutputTest {
private static final String TEN = "ten";
private static final String ODD = "odd";
- private static final String ALL = "all";
- private static final String EVEN_AND_ODD = "evenAndOdd";
- private static final String ODD_AND_TEN = "oddAndTen";
private static final String EVEN = "even";
private static final String NON_SELECTED = "nonSelected";
@@ -98,19 +96,24 @@ public class DirectedOutputTest {
@Test
public void outputSelectorTest() throws Exception {
- StreamExecutionEnvironment env = new TestStreamEnvironment(1, 32);
+ StreamExecutionEnvironment env = new TestStreamEnvironment(3, 32);
+
+ TestListResultSink<Long> evenSink = new TestListResultSink<Long>();
+ TestListResultSink<Long> oddAndTenSink = new TestListResultSink<Long>();
+ TestListResultSink<Long> evenAndOddSink = new TestListResultSink<Long>();
+ TestListResultSink<Long> allSink = new TestListResultSink<Long>();
SplitDataStream<Long> source = env.generateSequence(1, 11).split(new MyOutputSelector());
- source.select(EVEN).addSink(new ListSink(EVEN));
- source.select(ODD, TEN).addSink(new ListSink(ODD_AND_TEN));
- source.select(EVEN, ODD).addSink(new ListSink(EVEN_AND_ODD));
- source.addSink(new ListSink(ALL));
+ source.select(EVEN).addSink(evenSink);
+ source.select(ODD, TEN).addSink(oddAndTenSink);
+ source.select(EVEN, ODD).addSink(evenAndOddSink);
+ source.addSink(allSink);
env.execute();
- assertEquals(Arrays.asList(2L, 4L, 6L, 8L, 10L), outputs.get(EVEN));
- assertEquals(Arrays.asList(1L, 3L, 5L, 7L, 9L, 10L, 11L), outputs.get(ODD_AND_TEN));
+ assertEquals(Arrays.asList(2L, 4L, 6L, 8L, 10L), evenSink.getSortedResult());
+ assertEquals(Arrays.asList(1L, 3L, 5L, 7L, 9L, 10L, 11L), oddAndTenSink.getSortedResult());
assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L, 11L),
- outputs.get(EVEN_AND_ODD));
- assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L, 11L), outputs.get(ALL));
+ evenAndOddSink.getResult());
+ assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L, 11L), allSink.getResult());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b4ee2a1/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java
new file mode 100644
index 0000000..7f72173
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java
@@ -0,0 +1,78 @@
+/*
+ * 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.util;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.TreeSet;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.function.sink.RichSinkFunction;
+
+public class TestListResultSink<T> extends RichSinkFunction<T> {
+
+ private int resultListId;
+
+ public TestListResultSink() {
+ this.resultListId = TestListWrapper.getInstance().createList();
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ }
+
+ @Override
+ public void invoke(T value) throws Exception {
+ synchronized (resultList()) {
+ resultList().add(value);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ }
+
+ private List<T> resultList() {
+ synchronized (TestListWrapper.getInstance()) {
+ return (List<T>) TestListWrapper.getInstance().getList(resultListId);
+ }
+ }
+
+ public List<T> getResult() {
+ synchronized (resultList()) {
+ ArrayList<T> copiedList = new ArrayList<T>(resultList());
+ return copiedList;
+ }
+ }
+
+ public List<T> getSortedResult() {
+ synchronized (resultList()) {
+ TreeSet<T> treeSet = new TreeSet<T>(resultList());
+ ArrayList<T> sortedList = new ArrayList<T>(treeSet);
+ return sortedList;
+ }
+ }
+
+ @Override
+ public void cancel() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b4ee2a1/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java
new file mode 100644
index 0000000..3c50f63
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java
@@ -0,0 +1,60 @@
+/*
+ * 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.util;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class TestListWrapper {
+
+ private static TestListWrapper instance;
+
+ private List<List<? extends Comparable>> lists;
+
+ private TestListWrapper() {
+ lists = Collections.synchronizedList(new ArrayList<List<? extends Comparable>>());
+ }
+
+ public static TestListWrapper getInstance() {
+ if (instance == null) {
+ instance = new TestListWrapper();
+ }
+ return instance;
+ }
+
+ /**
+ * Creates and stores a list, returns with the id.
+ *
+ * @return The ID of the list.
+ */
+ public int createList() {
+ lists.add(new ArrayList<Comparable>());
+ return lists.size() - 1;
+ }
+
+ public List<?> getList(int listId) {
+ List<? extends Comparable> list = lists.get(listId);
+ if (list == null) {
+ throw new RuntimeException("No such list.");
+ }
+
+ return list;
+ }
+
+}
\ No newline at end of file
[07/10] flink git commit: [FLINK-1594] [streaming] Self connect tests
and fixes
Posted by gy...@apache.org.
[FLINK-1594] [streaming] Self connect tests and fixes
Closes #472
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/221e5e6a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/221e5e6a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/221e5e6a
Branch: refs/heads/master
Commit: 221e5e6aeed1210d11eb6c73c73b79ab0f10c8cc
Parents: adcdfda
Author: Gábor Hermann <re...@gmail.com>
Authored: Thu Mar 12 08:12:21 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:04 2015 +0100
----------------------------------------------------------------------
.../flink/streaming/api/StreamConfig.java | 56 +----
.../apache/flink/streaming/api/StreamEdge.java | 41 ++-
.../api/StreamingJobGraphGenerator.java | 88 ++++---
.../api/collector/CollectorWrapper.java | 1 -
.../BroadcastOutputSelectorWrapper.java | 2 +
.../selector/DirectedOutputSelectorWrapper.java | 3 +
.../selector/OutputSelectorWrapperFactory.java | 1 +
.../api/streamvertex/CoStreamVertex.java | 2 +-
.../api/streamvertex/InputHandler.java | 6 -
.../api/streamvertex/OutputHandler.java | 39 +--
.../flink/streaming/api/CoStreamTest.java | 92 ++++---
.../operator/co/SelfConnectionTest.java | 252 +++++++++++++++++++
.../api/streamvertex/StreamVertexTest.java | 2 +
.../streaming/util/TestListResultSink.java | 2 +
.../flink/streaming/util/TestListWrapper.java | 4 +
15 files changed, 455 insertions(+), 136 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
index fc1f435..c1e9606 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
@@ -25,27 +25,23 @@ import java.util.Map;
import org.apache.commons.lang3.SerializationException;
import org.apache.commons.lang3.SerializationUtils;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.collector.selector.OutputSelectorWrapper;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.api.streamvertex.StreamVertexException;
-import org.apache.flink.streaming.partitioner.StreamPartitioner;
import org.apache.flink.util.InstantiationUtil;
public class StreamConfig implements Serializable {
private static final long serialVersionUID = 1L;
- private static final String INPUT_TYPE = "inputType_";
private static final String NUMBER_OF_OUTPUTS = "numberOfOutputs";
private static final String NUMBER_OF_INPUTS = "numberOfInputs";
private static final String CHAINED_OUTPUTS = "chainedOutputs";
private static final String CHAINED_TASK_CONFIG = "chainedTaskConfig_";
private static final String IS_CHAINED_VERTEX = "isChainedSubtask";
private static final String OUTPUT_NAME = "outputName_";
- private static final String PARTITIONER_OBJECT = "partitionerObject_";
private static final String VERTEX_NAME = "vertexID";
private static final String OPERATOR_NAME = "operatorName";
private static final String ITERATION_ID = "iteration-id";
@@ -219,24 +215,6 @@ public class StreamConfig implements Serializable {
return config.getLong(ITERATON_WAIT, 0);
}
- public <T> void setPartitioner(Integer output, StreamPartitioner<T> partitionerObject) {
-
- config.setBytes(PARTITIONER_OBJECT + output,
- SerializationUtils.serialize(partitionerObject));
- }
-
- @SuppressWarnings("unchecked")
- public <T> StreamPartitioner<T> getPartitioner(ClassLoader cl, Integer output) {
- StreamPartitioner<T> partitioner = null;
- try {
- partitioner = (StreamPartitioner<T>) InstantiationUtil.readObjectFromConfig(
- this.config, PARTITIONER_OBJECT + output, cl);
- } catch (Exception e) {
- throw new RuntimeException("Partitioner could not be instantiated.");
- }
- return partitioner;
- }
-
public void setSelectedNames(Integer output, List<String> selected) {
if (selected != null) {
config.setBytes(OUTPUT_NAME + output,
@@ -311,12 +289,12 @@ public class StreamConfig implements Serializable {
}
}
- public void setInEdges(List<StreamEdge> inEdges) {
+ public void setInPhysicalEdges(List<StreamEdge> inEdges) {
config.setBytes(IN_STREAM_EDGES, SerializationUtils.serialize((Serializable) inEdges));
}
@SuppressWarnings("unchecked")
- public List<StreamEdge> getInEdges(ClassLoader cl) {
+ public List<StreamEdge> getInPhysicalEdges(ClassLoader cl) {
try {
return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(
this.config, IN_STREAM_EDGES, cl);
@@ -325,41 +303,31 @@ public class StreamConfig implements Serializable {
}
}
- public void setOutEdgesInOrder(List<Tuple2<Integer, Integer>> outEdgeList) {
-
- config.setBytes(EDGES_IN_ORDER, SerializationUtils.serialize((Serializable) outEdgeList));
- }
-
-
public void setStateMonitoring(boolean stateMonitoring) {
-
+
config.setBoolean(STATE_MONITORING, stateMonitoring);
-
+
}
-
+
public boolean getStateMonitoring()
{
return config.getBoolean(STATE_MONITORING, false);
}
+ public void setOutEdgesInOrder(List<StreamEdge> outEdgeList) {
+ config.setBytes(EDGES_IN_ORDER, SerializationUtils.serialize((Serializable) outEdgeList));
+ }
+
@SuppressWarnings("unchecked")
- public List<Tuple2<Integer, Integer>> getOutEdgesInOrder(ClassLoader cl) {
+ public List<StreamEdge> getOutEdgesInOrder(ClassLoader cl) {
try {
- return (List<Tuple2<Integer, Integer>>) InstantiationUtil.readObjectFromConfig(
+ return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(
this.config, EDGES_IN_ORDER, cl);
} catch (Exception e) {
throw new RuntimeException("Could not instantiate outputs.");
}
}
- public void setInputIndex(int inputNumber, Integer inputTypeNumber) {
- config.setInteger(INPUT_TYPE + inputNumber++, inputTypeNumber);
- }
-
- public int getInputIndex(int inputNumber) {
- return config.getInteger(INPUT_TYPE + inputNumber, 0);
- }
-
public void setTransitiveChainedTaskConfigs(Map<Integer, StreamConfig> chainedTaskConfigs) {
config.setBytes(CHAINED_TASK_CONFIG,
SerializationUtils.serialize((Serializable) chainedTaskConfigs));
@@ -402,7 +370,7 @@ public class StreamConfig implements Serializable {
builder.append("\nPartitioning:");
for (StreamEdge output : getNonChainedOutputs(cl)) {
int outputname = output.getTargetVertex();
- builder.append("\n\t" + outputname + ": " + getPartitioner(cl, outputname));
+ builder.append("\n\t" + outputname + ": " + output.getPartitioner());
}
builder.append("\nChained subtasks: " + getChainedOutputs(cl));
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
index a300fc2..74edb00 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
@@ -28,6 +28,10 @@ import org.apache.flink.streaming.partitioner.StreamPartitioner;
*/
public class StreamEdge implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ final private String edgeId;
+
final private int sourceVertex;
final private int targetVertex;
@@ -48,6 +52,12 @@ public class StreamEdge implements Serializable {
this.typeNumber = typeNumber;
this.selectedNames = selectedNames;
this.outputPartitioner = outputPartitioner;
+
+ this.edgeId = sourceVertex + "_"
+ + targetVertex + "_"
+ + typeNumber + "_"
+ + selectedNames + "_"
+ + outputPartitioner;
}
public int getSourceVertex() {
@@ -71,13 +81,36 @@ public class StreamEdge implements Serializable {
}
@Override
+ public int hashCode() {
+ return edgeId.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ StreamEdge that = (StreamEdge) o;
+
+ if (!edgeId.equals(that.edgeId)) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
public String toString() {
- return "StreamGraphEdge{" +
- "sourceVertex=" + sourceVertex +
- ", targetVertex=" + targetVertex +
+ return "(" +
+ sourceVertex +
+ " -> " + targetVertex +
", typeNumber=" + typeNumber +
", selectedNames=" + selectedNames +
", outputPartitioner=" + outputPartitioner +
- '}';
+ ')';
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/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 d420030..8a110bf 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
@@ -25,7 +25,6 @@ import java.util.List;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
@@ -53,7 +52,10 @@ public class StreamingJobGraphGenerator {
private JobGraph jobGraph;
private Collection<Integer> builtVertices;
+ private List<StreamEdge> physicalEdgesInOrder;
+
private Map<Integer, Map<Integer, StreamConfig>> chainedConfigs;
+
private Map<Integer, StreamConfig> vertexConfigs;
private Map<Integer, String> chainedNames;
@@ -67,6 +69,7 @@ public class StreamingJobGraphGenerator {
this.chainedConfigs = new HashMap<Integer, Map<Integer, StreamConfig>>();
this.vertexConfigs = new HashMap<Integer, StreamConfig>();
this.chainedNames = new HashMap<Integer, String>();
+ this.physicalEdgesInOrder = new ArrayList<StreamEdge>();
}
public JobGraph createJobGraph(String jobName) {
@@ -77,30 +80,57 @@ public class StreamingJobGraphGenerator {
jobGraph.setJobType(JobGraph.JobType.STREAMING);
jobGraph.setMonitoringEnabled(streamGraph.isMonitoringEnabled());
jobGraph.setMonitorInterval(streamGraph.getMonitoringInterval());
- if(jobGraph.isMonitoringEnabled())
- {
+ if (jobGraph.isMonitoringEnabled()) {
jobGraph.setNumberOfExecutionRetries(Integer.MAX_VALUE);
}
init();
setChaining();
+ setPhysicalEdges();
+
setSlotSharing();
return jobGraph;
}
+ private void setPhysicalEdges() {
+ Map<Integer, List<StreamEdge>> physicalInEdgesInOrder = new HashMap<Integer, List<StreamEdge>>();
+
+ for (StreamEdge edge : physicalEdgesInOrder) {
+ int target = edge.getTargetVertex();
+
+ List<StreamEdge> inEdges = physicalInEdgesInOrder.get(target);
+
+ // create if not set
+ if (inEdges == null) {
+ inEdges = new ArrayList<StreamEdge>();
+ physicalInEdgesInOrder.put(target, inEdges);
+ }
+
+ inEdges.add(edge);
+ }
+
+ for (Map.Entry<Integer, List<StreamEdge>> inEdges : physicalInEdgesInOrder.entrySet()) {
+ int vertex = inEdges.getKey();
+ List<StreamEdge> edgeList = inEdges.getValue();
+
+ vertexConfigs.get(vertex).setInPhysicalEdges(edgeList);
+ }
+ }
+
private void setChaining() {
for (Integer sourceName : streamGraph.getSources()) {
createChain(sourceName, sourceName);
}
}
- private List<Tuple2<Integer, Integer>> createChain(Integer startNode, Integer current) {
+ private List<StreamEdge> createChain(Integer startNode, Integer current) {
if (!builtVertices.contains(startNode)) {
- List<Tuple2<Integer, Integer>> transitiveOutEdges = new ArrayList<Tuple2<Integer, Integer>>();
+ List<StreamEdge> transitiveOutEdges = new ArrayList<StreamEdge>();
+
List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();
@@ -117,7 +147,7 @@ public class StreamingJobGraphGenerator {
}
for (StreamEdge nonChainable : nonChainableOutputs) {
- transitiveOutEdges.add(new Tuple2<Integer, Integer>(current, nonChainable.getTargetVertex()));
+ transitiveOutEdges.add(nonChainable);
createChain(nonChainable.getTargetVertex(), nonChainable.getTargetVertex());
}
@@ -133,9 +163,8 @@ public class StreamingJobGraphGenerator {
config.setChainStart();
config.setOutEdgesInOrder(transitiveOutEdges);
config.setOutEdges(streamGraph.getOutEdges(current));
- config.setInEdges(streamGraph.getInEdges(current));
- for (Tuple2<Integer, Integer> edge : transitiveOutEdges) {
+ for (StreamEdge edge : transitiveOutEdges) {
connect(startNode, edge);
}
@@ -154,7 +183,7 @@ public class StreamingJobGraphGenerator {
return transitiveOutEdges;
} else {
- return new ArrayList<Tuple2<Integer, Integer>>();
+ return new ArrayList<StreamEdge>();
}
}
@@ -165,9 +194,11 @@ public class StreamingJobGraphGenerator {
for (StreamEdge chainable : chainedOutputs) {
outputChainedNames.add(chainedNames.get(chainable.getTargetVertex()));
}
- return operatorName + " -> (" + StringUtils.join(outputChainedNames, ", ") + ")";
+ String returnOperatorName = operatorName + " -> (" + StringUtils.join(outputChainedNames, ", ") + ")";
+ return returnOperatorName;
} else if (chainedOutputs.size() == 1) {
- return operatorName + " -> " + chainedNames.get(chainedOutputs.get(0));
+ String returnOperatorName = operatorName + " -> " + chainedNames.get(chainedOutputs.get(0).getTargetVertex());
+ return returnOperatorName;
} else {
return operatorName;
}
@@ -238,30 +269,20 @@ public class StreamingJobGraphGenerator {
vertexConfigs.put(vertexID, config);
}
- private <T> void connect(Integer headOfChain, Tuple2<Integer, Integer> edge) {
+ private void connect(Integer headOfChain, StreamEdge edge) {
- Integer upStreamvertexID = edge.f0;
- Integer downStreamvertexID = edge.f1;
+ physicalEdgesInOrder.add(edge);
- int outputIndex = streamGraph.getOutEdges(upStreamvertexID).indexOf(downStreamvertexID);
+ Integer downStreamvertexID = edge.getTargetVertex();
AbstractJobVertex headVertex = streamVertices.get(headOfChain);
AbstractJobVertex downStreamVertex = streamVertices.get(downStreamvertexID);
StreamConfig downStreamConfig = new StreamConfig(downStreamVertex.getConfiguration());
- StreamConfig upStreamConfig = headOfChain.equals(upStreamvertexID) ? new StreamConfig(
- headVertex.getConfiguration()) : chainedConfigs.get(headOfChain).get(
- upStreamvertexID);
-
- int numOfInputs = downStreamConfig.getNumberOfInputs();
-
- downStreamConfig.setInputIndex(numOfInputs++, streamGraph.getEdge(upStreamvertexID, downStreamvertexID).getTypeNumber());
- downStreamConfig.setNumberOfInputs(numOfInputs);
-
- StreamPartitioner<?> partitioner = streamGraph.getEdge(upStreamvertexID, downStreamvertexID).getPartitioner();
- upStreamConfig.setPartitioner(downStreamvertexID, partitioner);
+ downStreamConfig.setNumberOfInputs(downStreamConfig.getNumberOfInputs() + 1);
+ StreamPartitioner<?> partitioner = edge.getPartitioner();
if (partitioner.getStrategy() == PartitioningStrategy.FORWARD) {
downStreamVertex.connectNewDataSetAsInput(headVertex, DistributionPattern.POINTWISE);
} else {
@@ -281,14 +302,15 @@ public class StreamingJobGraphGenerator {
StreamInvokable<?, ?> headInvokable = streamGraph.getInvokable(vertexID);
StreamInvokable<?, ?> outInvokable = streamGraph.getInvokable(outName);
- return streamGraph.getInEdges(outName).size() == 1
- && outInvokable != null
- && outInvokable.getChainingStrategy() == ChainingStrategy.ALWAYS
- && (headInvokable.getChainingStrategy() == ChainingStrategy.HEAD || headInvokable
+ return
+ streamGraph.getInEdges(outName).size() == 1
+ && outInvokable != null
+ && outInvokable.getChainingStrategy() == ChainingStrategy.ALWAYS
+ && (headInvokable.getChainingStrategy() == ChainingStrategy.HEAD || headInvokable
.getChainingStrategy() == ChainingStrategy.ALWAYS)
- && streamGraph.getEdge(vertexID, outName).getPartitioner().getStrategy() == PartitioningStrategy.FORWARD
- && streamGraph.getParallelism(vertexID) == streamGraph.getParallelism(outName)
- && streamGraph.chaining;
+ && edge.getPartitioner().getStrategy() == PartitioningStrategy.FORWARD
+ && streamGraph.getParallelism(vertexID) == streamGraph.getParallelism(outName)
+ && streamGraph.chaining;
}
private void setSlotSharing() {
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
index bb0268a..4a0369c 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/CollectorWrapper.java
@@ -29,7 +29,6 @@ public class CollectorWrapper<OUT> implements Collector<OUT> {
this.outputSelectorWrapper = outputSelectorWrapper;
}
- @SuppressWarnings("unchecked")
public void addCollector(Collector<?> output, StreamEdge edge) {
outputSelectorWrapper.addCollector(output, edge);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
index 44371f0..78ef914 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/BroadcastOutputSelectorWrapper.java
@@ -25,12 +25,14 @@ import org.apache.flink.util.Collector;
public class BroadcastOutputSelectorWrapper<OUT> implements OutputSelectorWrapper<OUT> {
+ private static final long serialVersionUID = 1L;
private List<Collector<OUT>> outputs;
public BroadcastOutputSelectorWrapper() {
outputs = new ArrayList<Collector<OUT>>();
}
+ @SuppressWarnings("unchecked")
@Override
public void addCollector(Collector<?> output, StreamEdge edge) {
outputs.add((Collector<OUT>) output);
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
index a04900c..1cb20d9 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/DirectedOutputSelectorWrapper.java
@@ -31,6 +31,8 @@ import org.slf4j.LoggerFactory;
public class DirectedOutputSelectorWrapper<OUT> implements OutputSelectorWrapper<OUT> {
+ private static final long serialVersionUID = 1L;
+
private static final Logger LOG = LoggerFactory.getLogger(DirectedOutputSelectorWrapper.class);
private List<OutputSelector<OUT>> outputSelectors;
@@ -44,6 +46,7 @@ public class DirectedOutputSelectorWrapper<OUT> implements OutputSelectorWrapper
this.outputMap = new HashMap<String, List<Collector<OUT>>>();
}
+ @SuppressWarnings("unchecked")
@Override
public void addCollector(Collector<?> output, StreamEdge edge) {
List<String> selectedNames = edge.getSelectedNames();
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
index c0f22c7..dca2ede 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/selector/OutputSelectorWrapperFactory.java
@@ -21,6 +21,7 @@ import java.util.List;
public class OutputSelectorWrapperFactory {
+ @SuppressWarnings({ "rawtypes", "unchecked" })
public static OutputSelectorWrapper<?> create(List<OutputSelector<?>> outputSelectors) {
if (outputSelectors.size() == 0) {
return new BroadcastOutputSelectorWrapper();
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
index b45fc44..f277be0 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
@@ -80,7 +80,7 @@ public class CoStreamVertex<IN1, IN2, OUT> extends StreamVertex<IN1, OUT> {
ArrayList<InputGate> inputList1 = new ArrayList<InputGate>();
ArrayList<InputGate> inputList2 = new ArrayList<InputGate>();
- List<StreamEdge> inEdges = configuration.getInEdges(userClassLoader);
+ List<StreamEdge> inEdges = configuration.getInPhysicalEdges(userClassLoader);
for (int i = 0; i < numberOfInputs; i++) {
int inputType = inEdges.get(i).getTypeNumber();
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
index 3988ec1..c6a4377 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/InputHandler.java
@@ -66,12 +66,6 @@ public class InputHandler<IN> {
}
}
- private IndexedReaderIterator<StreamRecord<IN>> createInputIterator() {
- final IndexedReaderIterator<StreamRecord<IN>> iter = new IndexedReaderIterator<StreamRecord<IN>>(
- inputs, inputSerializer);
- return iter;
- }
-
protected static <T> IndexedReaderIterator<StreamRecord<T>> staticCreateInputIterator(
MutableReader<?> inputReader, TypeSerializer<StreamRecord<T>> serializer) {
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
index bb66c63..42afcae 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
@@ -24,7 +24,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.plugable.SerializationDelegate;
@@ -52,9 +51,10 @@ public class OutputHandler<OUT> {
public List<ChainableInvokable<?, ?>> chainedInvokables;
- private Map<Integer, StreamOutput<?>> outputMap;
+ private Map<StreamEdge, StreamOutput<?>> outputMap;
+
private Map<Integer, StreamConfig> chainedConfigs;
- private List<Tuple2<Integer, Integer>> outEdgesInOrder;
+ private List<StreamEdge> outEdgesInOrder;
public OutputHandler(StreamVertex<?, OUT> vertex) {
@@ -62,7 +62,7 @@ public class OutputHandler<OUT> {
this.vertex = vertex;
this.configuration = new StreamConfig(vertex.getTaskConfiguration());
this.chainedInvokables = new ArrayList<ChainableInvokable<?, ?>>();
- this.outputMap = new HashMap<Integer, StreamOutput<?>>();
+ this.outputMap = new HashMap<StreamEdge, StreamOutput<?>>();
this.cl = vertex.getUserCodeClassLoader();
// We read the chained configs, and the order of record writer
@@ -74,16 +74,18 @@ public class OutputHandler<OUT> {
// We iterate through all the out edges from this job vertex and create
// a stream output
- for (Tuple2<Integer, Integer> outEdge : outEdgesInOrder) {
- StreamOutput<?> streamOutput = createStreamOutput(outEdge.f1,
- chainedConfigs.get(outEdge.f0), outEdgesInOrder.indexOf(outEdge));
- outputMap.put(outEdge.f1, streamOutput);
+ for (StreamEdge outEdge : outEdgesInOrder) {
+ StreamOutput<?> streamOutput = createStreamOutput(
+ outEdge,
+ outEdge.getTargetVertex(),
+ chainedConfigs.get(outEdge.getSourceVertex()),
+ outEdgesInOrder.indexOf(outEdge));
+ outputMap.put(outEdge, streamOutput);
}
// We create the outer collector that will be passed to the first task
// in the chain
this.outerCollector = createChainedCollector(configuration);
-
}
public void broadcastBarrier(long id) throws IOException, InterruptedException {
@@ -120,9 +122,7 @@ public class OutputHandler<OUT> {
// Create collectors for the network outputs
for (StreamEdge outputEdge : chainedTaskConfig.getNonChainedOutputs(cl)) {
- Integer output = outputEdge.getTargetVertex();
-
- Collector<?> outCollector = outputMap.get(output);
+ Collector<?> outCollector = outputMap.get(outputEdge);
wrapper.addCollector(outCollector, outputEdge);
}
@@ -164,14 +164,14 @@ public class OutputHandler<OUT> {
*
* @param outputVertex
* Name of the output to which the streamoutput will be set up
- * @param configuration
+ * @param upStreamConfig
* The config of upStream task
- * @return
+ * @return The created StreamOutput
*/
- private <T> StreamOutput<T> createStreamOutput(Integer outputVertex,
- StreamConfig configuration, int outputIndex) {
+ private <T> StreamOutput<T> createStreamOutput(StreamEdge edge, Integer outputVertex,
+ StreamConfig upStreamConfig, int outputIndex) {
- StreamRecordSerializer<T> outSerializer = configuration
+ StreamRecordSerializer<T> outSerializer = upStreamConfig
.getTypeSerializerOut1(vertex.userClassLoader);
SerializationDelegate<StreamRecord<T>> outSerializationDelegate = null;
@@ -180,12 +180,13 @@ public class OutputHandler<OUT> {
outSerializationDelegate.setInstance(outSerializer.createInstance());
}
- StreamPartitioner<T> outputPartitioner = configuration.getPartitioner(cl, outputVertex);
+ @SuppressWarnings("unchecked")
+ StreamPartitioner<T> outputPartitioner = (StreamPartitioner<T>) edge.getPartitioner();
ResultPartitionWriter bufferWriter = vertex.getEnvironment().getWriter(outputIndex);
RecordWriter<SerializationDelegate<StreamRecord<T>>> output =
- RecordWriterFactory.createRecordWriter(bufferWriter, outputPartitioner, configuration.getBufferTimeout());
+ RecordWriterFactory.createRecordWriter(bufferWriter, outputPartitioner, upStreamConfig.getBufferTimeout());
StreamOutput<T> streamOutput = new StreamOutput<T>(output, vertex.instanceID,
outSerializationDelegate);
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java
index 20cd189..f527de4 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/CoStreamTest.java
@@ -21,67 +21,101 @@ import static org.junit.Assert.assertEquals;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.co.CoMapFunction;
+import org.apache.flink.streaming.api.function.co.CoFlatMapFunction;
+import org.apache.flink.streaming.api.invokable.StreamInvokable;
import org.apache.flink.streaming.util.TestListResultSink;
import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.apache.flink.util.Collector;
+import org.junit.Test;
public class CoStreamTest {
private static final long MEMORY_SIZE = 32;
- private static ArrayList<String> expected;
+ private static ArrayList<String> expected = new ArrayList<String>();
- public static void main(String[] args) throws InterruptedException {
- for (int i = 0; i < 200; i++) {
- test();
- }
- }
+ @Test
+ public void test() {
- // @Test
- public static void test() {
- expected = new ArrayList<String>();
-
- StreamExecutionEnvironment env = new TestStreamEnvironment(3, MEMORY_SIZE);
+ StreamExecutionEnvironment env = new TestStreamEnvironment(1, MEMORY_SIZE);
TestListResultSink<String> resultSink = new TestListResultSink<String>();
DataStream<Integer> src = env.fromElements(1, 3, 5);
- DataStream<Integer> src2 = env.fromElements(1, 3, 5);
- DataStream<Integer> grouped = src.groupBy(new KeySelector<Integer, Integer>() {
+ DataStream<Integer> filter1 = src.filter(new FilterFunction<Integer>() {
+
+ private static final long serialVersionUID = 1L;
+
@Override
- public Integer getKey(Integer value) throws Exception {
- return value;
+ public boolean filter(Integer value) throws Exception {
+ return true;
}
- });
+ }).groupBy(new KeySelector<Integer, Integer>() {
+
+ private static final long serialVersionUID = 1L;
- DataStream<Integer> grouped2 = src2.groupBy(new KeySelector<Integer, Integer>() {
@Override
public Integer getKey(Integer value) throws Exception {
return value;
}
});
- DataStream<String> connected = grouped.connect(grouped2).map(new CoMapFunction<Integer, Integer, String>() {
+ DataStream<Tuple2<Integer, Integer>> filter2 = src
+ .map(new MapFunction<Integer, Tuple2<Integer, Integer>>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<Integer, Integer> map(Integer value) throws Exception {
+ return new Tuple2<Integer, Integer>(value, value + 1);
+ }
+ })
+ .distribute()
+ .filter(new FilterFunction<Tuple2<Integer, Integer>>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean filter(Tuple2<Integer, Integer> value) throws Exception {
+ return true;
+ }
+ }).setChainingStrategy(StreamInvokable.ChainingStrategy.NEVER).groupBy(new KeySelector<Tuple2<Integer, Integer>, Integer>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Integer getKey(Tuple2<Integer, Integer> value) throws Exception {
+ return value.f0;
+ }
+ });
+
+ DataStream<String> connected = filter1.connect(filter2).flatMap(new CoFlatMapFunction<Integer, Tuple2<Integer, Integer>, String>() {
+
+ private static final long serialVersionUID = 1L;
+
@Override
- public String map1(Integer value) {
- return value.toString();
+ public void flatMap1(Integer value, Collector<String> out) throws Exception {
+ out.collect(value.toString());
}
@Override
- public String map2(Integer value) {
- return value.toString();
+ public void flatMap2(Tuple2<Integer, Integer> value, Collector<String> out) throws Exception {
+ out.collect(value.toString());
}
});
connected.addSink(resultSink);
- connected.print();
-
try {
env.execute();
} catch (Exception e) {
@@ -89,9 +123,11 @@ public class CoStreamTest {
}
expected = new ArrayList<String>();
- expected.addAll(Arrays.asList("1", "1", "3", "3", "5", "5"));
+ expected.addAll(Arrays.asList("(1,2)", "(3,4)", "(5,6)", "1", "3", "5"));
+
+ List<String> result = resultSink.getResult();
+ Collections.sort(result);
- System.out.println(resultSink.getResult());
- assertEquals(expected, expected);
+ assertEquals(expected, result);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/co/SelfConnectionTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/co/SelfConnectionTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/co/SelfConnectionTest.java
new file mode 100644
index 0000000..b58245a
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/co/SelfConnectionTest.java
@@ -0,0 +1,252 @@
+/*
+ * 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.co;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.function.co.CoMapFunction;
+import org.apache.flink.streaming.api.invokable.StreamInvokable;
+import org.apache.flink.streaming.api.windowing.helper.Timestamp;
+import org.apache.flink.streaming.util.TestListResultSink;
+import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.apache.flink.util.Collector;
+import org.junit.Test;
+
+public class SelfConnectionTest implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int MEMORY_SIZE = 32;
+
+ private static List<String> expected;
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Test
+ public void sameDataStreamTest() {
+
+ StreamExecutionEnvironment env = new TestStreamEnvironment(3, MEMORY_SIZE);
+
+ TestListResultSink<String> resultSink = new TestListResultSink<String>();
+
+ Timestamp<Integer> timeStamp = new Timestamp<Integer>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public long getTimestamp(Integer value) {
+ return value;
+ }
+
+ };
+
+ KeySelector keySelector = new KeySelector<Integer, Integer>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Integer getKey(Integer value) throws Exception {
+ return value;
+ }
+ };
+
+ DataStream<Integer> src = env.fromElements(1, 3, 5);
+
+ @SuppressWarnings("unused")
+ DataStream<Tuple2<Integer, Integer>> dataStream =
+ src.join(src).onWindow(50L, timeStamp, timeStamp).where(keySelector).equalTo(keySelector)
+ .map(new MapFunction<Tuple2<Integer, Integer>, String>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String map(Tuple2<Integer, Integer> value) throws Exception {
+ return value.toString();
+ }
+ })
+ .addSink(resultSink);
+
+
+ try {
+ env.execute();
+
+ expected = new ArrayList<String>();
+
+ expected.addAll(Arrays.asList("(1,1)", "(3,3)", "(5,5)"));
+
+ List<String> result = resultSink.getResult();
+
+ Collections.sort(expected);
+ Collections.sort(result);
+
+ assertEquals(expected, result);
+ } catch (Exception e) {
+ fail();
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * We connect two different data streams in a chain to a CoMap.
+ */
+ @Test
+ public void differentDataStreamSameChain() {
+
+ TestListResultSink<String> resultSink = new TestListResultSink<String>();
+
+ StreamExecutionEnvironment env = new TestStreamEnvironment(1, MEMORY_SIZE);
+
+ DataStream<Integer> src = env.fromElements(1, 3, 5);
+
+ DataStream<String> stringMap = src.map(new MapFunction<Integer, String>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String map(Integer value) throws Exception {
+ return "x " + value;
+ }
+ }).setChainingStrategy(StreamInvokable.ChainingStrategy.ALWAYS);
+
+ stringMap.connect(src).map(new CoMapFunction<String, Integer, String>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String map1(String value) {
+ return value;
+ }
+
+ @Override
+ public String map2(Integer value) {
+ return String.valueOf(value + 1);
+ }
+ }).addSink(resultSink);
+
+ try {
+ env.execute();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ expected = new ArrayList<String>();
+
+ expected.addAll(Arrays.asList("x 1", "x 3", "x 5", "2", "4", "6"));
+
+ List<String> result = resultSink.getResult();
+
+ Collections.sort(expected);
+ Collections.sort(result);
+
+ assertEquals(expected, result);
+ }
+
+ /**
+ * We connect two different data streams in different chains to a CoMap.
+ * (This is not actually self-connect.)
+ */
+ @Test
+ public void differentDataStreamDifferentChain() {
+
+ TestListResultSink<String> resultSink = new TestListResultSink<String>();
+
+ StreamExecutionEnvironment env = new TestStreamEnvironment(3, MEMORY_SIZE);
+
+ DataStream<Integer> src = env.fromElements(1, 3, 5).setChainingStrategy(StreamInvokable.ChainingStrategy.NEVER);
+
+ DataStream<String> stringMap = src.flatMap(new FlatMapFunction<Integer, String>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void flatMap(Integer value, Collector<String> out) throws Exception {
+ out.collect("x " + value);
+ }
+ }).groupBy(new KeySelector<String, Integer>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Integer getKey(String value) throws Exception {
+ return value.length();
+ }
+ });
+
+ DataStream<Long> longMap = src.map(new MapFunction<Integer, Long>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Long map(Integer value) throws Exception {
+ return Long.valueOf(value + 1);
+ }
+ }).groupBy(new KeySelector<Long, Long>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Long getKey(Long value) throws Exception {
+ return value;
+ }
+ });
+
+
+ stringMap.connect(longMap).map(new CoMapFunction<String, Long, String>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String map1(String value) {
+ return value;
+ }
+
+ @Override
+ public String map2(Long value) {
+ return value.toString();
+ }
+ }).addSink(resultSink);
+
+ try {
+ env.execute();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ expected = new ArrayList<String>();
+
+ expected.addAll(Arrays.asList("x 1", "x 3", "x 5", "2", "4", "6"));
+
+ List<String> result = resultSink.getResult();
+
+ Collections.sort(expected);
+ Collections.sort(result);
+
+ assertEquals(expected, result);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamvertex/StreamVertexTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamvertex/StreamVertexTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamvertex/StreamVertexTest.java
index 18a36ac..a88a60d 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamvertex/StreamVertexTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamvertex/StreamVertexTest.java
@@ -134,11 +134,13 @@ public class StreamVertexTest {
@Override
public String map1(String value) {
+// System.out.println(value);
return value;
}
@Override
public String map2(Long value) {
+// System.out.println(value);
return value.toString();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java
index 7f72173..8b78a42 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListResultSink.java
@@ -27,6 +27,7 @@ import org.apache.flink.streaming.api.function.sink.RichSinkFunction;
public class TestListResultSink<T> extends RichSinkFunction<T> {
+ private static final long serialVersionUID = 1L;
private int resultListId;
public TestListResultSink() {
@@ -50,6 +51,7 @@ public class TestListResultSink<T> extends RichSinkFunction<T> {
super.close();
}
+ @SuppressWarnings("unchecked")
private List<T> resultList() {
synchronized (TestListWrapper.getInstance()) {
return (List<T>) TestListWrapper.getInstance().getList(resultListId);
http://git-wip-us.apache.org/repos/asf/flink/blob/221e5e6a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java
index 3c50f63..751f836 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/TestListWrapper.java
@@ -25,8 +25,10 @@ public class TestListWrapper {
private static TestListWrapper instance;
+ @SuppressWarnings("rawtypes")
private List<List<? extends Comparable>> lists;
+ @SuppressWarnings("rawtypes")
private TestListWrapper() {
lists = Collections.synchronizedList(new ArrayList<List<? extends Comparable>>());
}
@@ -43,12 +45,14 @@ public class TestListWrapper {
*
* @return The ID of the list.
*/
+ @SuppressWarnings("rawtypes")
public int createList() {
lists.add(new ArrayList<Comparable>());
return lists.size() - 1;
}
public List<?> getList(int listId) {
+ @SuppressWarnings("rawtypes")
List<? extends Comparable> list = lists.get(listId);
if (list == null) {
throw new RuntimeException("No such list.");
[08/10] flink git commit: [FLINK-1618] [streaming] Parallel time
reduce
Posted by gy...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/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 d7338a0..34986c8 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
@@ -30,6 +30,8 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.function.WindowMapFunction;
import org.apache.flink.streaming.api.function.sink.SinkFunction;
+import org.apache.flink.streaming.api.function.source.ParallelSourceFunction;
+import org.apache.flink.streaming.api.function.source.RichParallelSourceFunction;
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;
@@ -129,81 +131,113 @@ public class WindowIntegrationTest implements Serializable {
source.window(Time.of(5, ts, 1)).every(Time.of(4, ts, 1)).groupBy(new ModKey(2)).sum(0)
.getDiscretizedStream().addSink(new DistributedSink5());
+ DataStream<Integer> source2 = env.addSource(new ParallelSourceFunction<Integer>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void run(Collector<Integer> collector) throws Exception {
+ for (int i = 1; i <= 10; i++) {
+ collector.collect(i);
+ }
+ }
+
+ @Override
+ public void cancel() {
+ }
+ });
+
+ DataStream<Integer> source3 = env.addSource(new RichParallelSourceFunction<Integer>() {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void run(Collector<Integer> collector) throws Exception {
+ for (int i = 1; i <= 11; i++) {
+ if (i % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
+ collector.collect(i);
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ }
+ });
+
+ source2.window(Time.of(2, ts, 1)).sum(0).getDiscretizedStream()
+ .addSink(new DistributedSink6());
+
+ source3.window(Time.of(5, ts, 1)).groupBy(new ModKey(2)).sum(0).getDiscretizedStream()
+ .addSink(new DistributedSink7());
+
env.execute();
- // sum ( Time of 3 slide 2 )
- List<StreamWindow<Integer>> expected1 = new
- ArrayList<StreamWindow<Integer>>();
- expected1.add(StreamWindow.fromElements(5));
- expected1.add(StreamWindow.fromElements(11));
- expected1.add(StreamWindow.fromElements(9));
- expected1.add(StreamWindow.fromElements(10));
- expected1.add(StreamWindow.fromElements(32));
-
- validateOutput(expected1, CentralSink1.windows);
-
- // Tumbling Time of 4 grouped by mod 2
- List<StreamWindow<Integer>> expected2 = new
- ArrayList<StreamWindow<Integer>>();
- expected2.add(StreamWindow.fromElements(2, 2, 4));
- expected2.add(StreamWindow.fromElements(1, 3));
- expected2.add(StreamWindow.fromElements(5));
- expected2.add(StreamWindow.fromElements(10));
- expected2.add(StreamWindow.fromElements(11, 11));
-
- validateOutput(expected2, CentralSink2.windows);
-
- // 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(22));
- expected3.add(StreamWindow.fromElements(8));
- expected3.add(StreamWindow.fromElements(10));
-
- validateOutput(expected3, DistributedSink1.windows);
-
- // groupby mod3 Tumbling Count of 2 grouped by mod 2
- List<StreamWindow<Integer>> expected4 = new
- ArrayList<StreamWindow<Integer>>();
- expected4.add(StreamWindow.fromElements(2, 2));
- expected4.add(StreamWindow.fromElements(1));
- expected4.add(StreamWindow.fromElements(4));
- expected4.add(StreamWindow.fromElements(5, 11));
- expected4.add(StreamWindow.fromElements(10));
- expected4.add(StreamWindow.fromElements(11));
- expected4.add(StreamWindow.fromElements(3));
-
- validateOutput(expected4, DistributedSink2.windows);
-
- // min ( Time of 2 slide 3 )
- List<StreamWindow<Integer>> expected5 = new
- ArrayList<StreamWindow<Integer>>();
- expected5.add(StreamWindow.fromElements(1));
- expected5.add(StreamWindow.fromElements(4));
- expected5.add(StreamWindow.fromElements(10));
-
- 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));
-
- validateOutput(expected6, DistributedSink3.windows);
-
- List<StreamWindow<Integer>> expected7 = new
- ArrayList<StreamWindow<Integer>>();
- expected7.add(StreamWindow.fromElements(1, 2, 2, 3, 4, 5));
- expected7.add(StreamWindow.fromElements(10));
- expected7.add(StreamWindow.fromElements(10, 11, 11));
-
- validateOutput(expected7, DistributedSink4.windows);
+ // sum ( Time of 3 slide 2 )
+ List<StreamWindow<Integer>> expected1 = new ArrayList<StreamWindow<Integer>>();
+ expected1.add(StreamWindow.fromElements(5));
+ expected1.add(StreamWindow.fromElements(11));
+ expected1.add(StreamWindow.fromElements(9));
+ expected1.add(StreamWindow.fromElements(10));
+ expected1.add(StreamWindow.fromElements(32));
+
+ validateOutput(expected1, CentralSink1.windows);
+
+ // Tumbling Time of 4 grouped by mod 2
+ List<StreamWindow<Integer>> expected2 = new ArrayList<StreamWindow<Integer>>();
+ expected2.add(StreamWindow.fromElements(2, 2, 4));
+ expected2.add(StreamWindow.fromElements(1, 3));
+ expected2.add(StreamWindow.fromElements(5));
+ expected2.add(StreamWindow.fromElements(10));
+ expected2.add(StreamWindow.fromElements(11, 11));
+
+ validateOutput(expected2, CentralSink2.windows);
+
+ // 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(22));
+ expected3.add(StreamWindow.fromElements(8));
+ expected3.add(StreamWindow.fromElements(10));
+
+ validateOutput(expected3, DistributedSink1.windows);
+
+ // groupby mod3 Tumbling Count of 2 grouped by mod 2
+ List<StreamWindow<Integer>> expected4 = new ArrayList<StreamWindow<Integer>>();
+ expected4.add(StreamWindow.fromElements(2, 2));
+ expected4.add(StreamWindow.fromElements(1));
+ expected4.add(StreamWindow.fromElements(4));
+ expected4.add(StreamWindow.fromElements(5, 11));
+ expected4.add(StreamWindow.fromElements(10));
+ expected4.add(StreamWindow.fromElements(11));
+ expected4.add(StreamWindow.fromElements(3));
+
+ validateOutput(expected4, DistributedSink2.windows);
+
+ // min ( Time of 2 slide 3 )
+ List<StreamWindow<Integer>> expected5 = new ArrayList<StreamWindow<Integer>>();
+ expected5.add(StreamWindow.fromElements(1));
+ expected5.add(StreamWindow.fromElements(4));
+ expected5.add(StreamWindow.fromElements(10));
+
+ 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));
+
+ validateOutput(expected6, DistributedSink3.windows);
+
+ List<StreamWindow<Integer>> expected7 = new ArrayList<StreamWindow<Integer>>();
+ expected7.add(StreamWindow.fromElements(1, 2, 2, 3, 4, 5));
+ expected7.add(StreamWindow.fromElements(10));
+ expected7.add(StreamWindow.fromElements(10, 11, 11));
+
+ validateOutput(expected7, DistributedSink4.windows);
List<StreamWindow<Integer>> expected8 = new ArrayList<StreamWindow<Integer>>();
expected8.add(StreamWindow.fromElements(4, 8));
@@ -216,6 +250,25 @@ public class WindowIntegrationTest implements Serializable {
validateOutput(expected8, DistributedSink5.windows);
+ List<StreamWindow<Integer>> expected9 = new ArrayList<StreamWindow<Integer>>();
+ expected9.add(StreamWindow.fromElements(6));
+ expected9.add(StreamWindow.fromElements(14));
+ expected9.add(StreamWindow.fromElements(22));
+ expected9.add(StreamWindow.fromElements(30));
+ expected9.add(StreamWindow.fromElements(38));
+
+ validateOutput(expected9, DistributedSink6.windows);
+
+ List<StreamWindow<Integer>> expected10 = new ArrayList<StreamWindow<Integer>>();
+ expected10.add(StreamWindow.fromElements(6, 9));
+ expected10.add(StreamWindow.fromElements(16, 24));
+
+ for (List<Integer> sw : DistributedSink7.windows) {
+ Collections.sort(sw);
+ }
+
+ validateOutput(expected10, DistributedSink7.windows);
+
}
public static <R> void validateOutput(List<R> expected, List<R> actual) {
@@ -357,4 +410,39 @@ public class WindowIntegrationTest implements Serializable {
}
}
+
+ @SuppressWarnings("serial")
+ private static class DistributedSink6 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);
+ }
+
+ @Override
+ public void cancel() {
+ }
+
+ }
+
+ @SuppressWarnings("serial")
+ private static class DistributedSink7 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);
+ }
+
+ @Override
+ public void cancel() {
+ }
+
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMergerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMergerTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMergerTest.java
index ac3f583..77037d3 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMergerTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMergerTest.java
@@ -52,15 +52,15 @@ public class WindowMergerTest {
List<StreamWindow<Integer>> input = new ArrayList<StreamWindow<Integer>>();
input.add(w1);
input.add(w4_1);
- input.addAll(w2.split(2));
- input.addAll(w3.partitionBy(new KeySelector<Integer, Integer>() {
+ input.addAll(StreamWindow.split(w2, 2));
+ input.addAll(StreamWindow.partitionBy(w3, new KeySelector<Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer getKey(Integer value) throws Exception {
return value % 2;
}
- }));
+ }, false));
input.add(w4_2);
List<StreamWindow<Integer>> output = MockContext.createAndExecute(windowMerger, input);
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitionerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitionerTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitionerTest.java
index da68211..9a2416c 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitionerTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitionerTest.java
@@ -42,12 +42,12 @@ public class WindowPartitionerTest {
StreamWindow<Integer> w2 = StreamWindow.fromElements(1, 2, 3, 4);
List<StreamWindow<Integer>> expected1 = new ArrayList<StreamWindow<Integer>>();
- expected1.addAll(w1.split(2));
- expected1.addAll(w2.split(2));
+ expected1.addAll(StreamWindow.split(w1,2));
+ expected1.addAll(StreamWindow.split(w2,2));
List<StreamWindow<Integer>> expected2 = new ArrayList<StreamWindow<Integer>>();
- expected2.addAll(w1.partitionBy(new MyKey()));
- expected2.addAll(w2.partitionBy(new MyKey()));
+ expected2.addAll(StreamWindow.partitionBy(w1,new MyKey(),false));
+ expected2.addAll(StreamWindow.partitionBy(w2,new MyKey(),false));
List<StreamWindow<Integer>> input = new ArrayList<StreamWindow<Integer>>();
input.add(w1);
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/StreamWindowTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/StreamWindowTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/StreamWindowTest.java
index a07b681..c3efc7b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/StreamWindowTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/StreamWindowTest.java
@@ -92,7 +92,7 @@ public class StreamWindowTest {
// good
}
- List<StreamWindow<Integer>> wList = merged.split(3);
+ List<StreamWindow<Integer>> wList = StreamWindow.split(merged,3);
StreamWindow<Integer> merged2 = StreamWindow.merge(wList);
@@ -133,12 +133,12 @@ public class StreamWindowTest {
streamWindow.add(5);
streamWindow.add(6);
- List<StreamWindow<Integer>> split = streamWindow.split(2);
+ List<StreamWindow<Integer>> split = StreamWindow.split(streamWindow,2);
assertEquals(2, split.size());
assertEquals(StreamWindow.fromElements(1, 2, 3), split.get(0));
assertEquals(StreamWindow.fromElements(4, 5, 6), split.get(1));
- List<StreamWindow<Integer>> split2 = streamWindow.split(6);
+ List<StreamWindow<Integer>> split2 = StreamWindow.split(streamWindow,6);
assertEquals(6, split2.size());
assertEquals(StreamWindow.fromElements(1), split2.get(0));
assertEquals(StreamWindow.fromElements(2), split2.get(1));
@@ -147,7 +147,7 @@ public class StreamWindowTest {
assertEquals(StreamWindow.fromElements(5), split2.get(4));
assertEquals(StreamWindow.fromElements(6), split2.get(5));
- List<StreamWindow<Integer>> split3 = streamWindow.split(10);
+ List<StreamWindow<Integer>> split3 = StreamWindow.split(streamWindow,10);
assertEquals(6, split3.size());
assertEquals(StreamWindow.fromElements(1), split3.get(0));
assertEquals(StreamWindow.fromElements(2), split3.get(1));
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducerTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducerTest.java
index 95aace0..55abc72 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducerTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducerTest.java
@@ -97,6 +97,32 @@ public class TumblingGroupedPreReducerTest {
}
+ @Test
+ public void testEmitWindow2() throws Exception {
+
+ List<Tuple2<Integer, Integer>> inputs = new ArrayList<Tuple2<Integer, Integer>>();
+ inputs.add(new Tuple2<Integer, Integer>(1, 1));
+ inputs.add(new Tuple2<Integer, Integer>(0, 0));
+ inputs.add(new Tuple2<Integer, Integer>(1, -1));
+ inputs.add(new Tuple2<Integer, Integer>(1, -2));
+
+ TestCollector<StreamWindow<Tuple2<Integer, Integer>>> collector = new TestCollector<StreamWindow<Tuple2<Integer, Integer>>>();
+ List<StreamWindow<Tuple2<Integer, Integer>>> collected = collector.getCollected();
+
+ WindowBuffer<Tuple2<Integer, Integer>> wb = new TumblingGroupedPreReducer<Tuple2<Integer, Integer>>(
+ reducer, key, serializer).sequentialID();
+
+ wb.store(serializer.copy(inputs.get(0)));
+ wb.store(serializer.copy(inputs.get(1)));
+ wb.emitWindow(collector);
+
+ System.out.println(collected);
+
+ wb.store(serializer.copy(inputs.get(0)));
+ wb.store(serializer.copy(inputs.get(1)));
+ wb.store(serializer.copy(inputs.get(2)));
+ }
+
private static <T> void assertSetEquals(Collection<T> first, Collection<T> second) {
assertEquals(new HashSet<T>(first), new HashSet<T>(second));
}
[06/10] flink git commit: [FLINK-1594] [streaming] Added
StreamGraphEdges
Posted by gy...@apache.org.
[FLINK-1594] [streaming] Added StreamGraphEdges
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/29a66155
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/29a66155
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/29a66155
Branch: refs/heads/master
Commit: 29a661551539ad96d31b92999d088ede9186b9ba
Parents: 6b4ee2a
Author: Gábor Hermann <re...@gmail.com>
Authored: Thu Feb 26 13:16:32 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:03 2015 +0100
----------------------------------------------------------------------
.../apache/flink/streaming/api/StreamEdge.java | 72 ++++++++++++
.../flink/streaming/api/StreamEdgeList.java | 112 ++++++++++++++++++
.../flink/streaming/api/StreamEdgeListTest.java | 116 +++++++++++++++++++
3 files changed, 300 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/29a66155/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
new file mode 100644
index 0000000..8743233
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
@@ -0,0 +1,72 @@
+/*
+ * 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.List;
+
+import org.apache.flink.streaming.partitioner.StreamPartitioner;
+
+public class StreamEdge {
+
+ final private int sourceVertex;
+ final private int targetVertex;
+ final private int typeNumber;
+ final private List<String> selectedNames;
+// private OutputSelector<?> outputSelector;
+ final private StreamPartitioner<?> outputPartitioner;
+
+ public StreamEdge(int sourceVertex, int targetVertex, int typeNumber, List<String> selectedNames, StreamPartitioner<?> outputPartitioner) {
+ this.sourceVertex = sourceVertex;
+ this.targetVertex = targetVertex;
+ this.typeNumber = typeNumber;
+ this.selectedNames = selectedNames;
+// this.outputSelector = outputSelector;
+ this.outputPartitioner = outputPartitioner;
+ }
+
+ public int getSourceVertex() {
+ return sourceVertex;
+ }
+
+ public int getTargetVertex() {
+ return targetVertex;
+ }
+
+ public int getTypeNumber() {
+ return typeNumber;
+ }
+
+ public List<String> getSelectedNames() {
+ return selectedNames;
+ }
+
+ public StreamPartitioner<?> getOutputPartitioner() {
+ return outputPartitioner;
+ }
+
+ @Override
+ public String toString() {
+ return "StreamGraphEdge{" +
+ "sourceVertex=" + sourceVertex +
+ ", targetVertex=" + targetVertex +
+ ", typeNumber=" + typeNumber +
+ ", selectedNames=" + selectedNames +
+ ", outputPartitioner=" + outputPartitioner +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/29a66155/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
new file mode 100644
index 0000000..85202ab
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdgeList.java
@@ -0,0 +1,112 @@
+/*
+ * 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.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+public class StreamEdgeList {
+
+ private Map<Integer, List<StreamEdge>> outEdgeLists;
+ private Map<Integer, List<StreamEdge>> inEdgeLists;
+
+ public StreamEdgeList() {
+ outEdgeLists = new HashMap<Integer, List<StreamEdge>>();
+ inEdgeLists = new HashMap<Integer, List<StreamEdge>>();
+ }
+
+ public void addVertex(int vertexId) {
+ outEdgeLists.put(vertexId, new ArrayList<StreamEdge>());
+ inEdgeLists.put(vertexId, new ArrayList<StreamEdge>());
+ }
+
+ public void removeVertex(int vertexId) {
+ ArrayList<StreamEdge> toRemove = new ArrayList<StreamEdge>();
+
+ for (StreamEdge edge : outEdgeLists.get(vertexId)) {
+ toRemove.add(edge);
+ }
+
+ for (StreamEdge edge : inEdgeLists.get(vertexId)) {
+ toRemove.add(edge);
+ }
+
+ for (StreamEdge edge : toRemove) {
+ removeEdge(edge);
+ }
+
+ outEdgeLists.remove(vertexId);
+ inEdgeLists.remove(vertexId);
+ }
+
+ public void addEdge(StreamEdge edge) {
+ int sourceId = edge.getSourceVertex();
+ int targetId = edge.getTargetVertex();
+ outEdgeLists.get(sourceId).add(edge);
+ inEdgeLists.get(targetId).add(edge);
+ }
+
+ public void removeEdge(StreamEdge edge) {
+ int sourceId = edge.getSourceVertex();
+ int targetId = edge.getTargetVertex();
+ removeEdge(sourceId, targetId);
+ }
+
+ public void removeEdge(int sourceId, int targetId) {
+ Iterator<StreamEdge> outIterator = outEdgeLists.get(sourceId).iterator();
+ while (outIterator.hasNext()) {
+ StreamEdge edge = outIterator.next();
+
+ if (edge.getTargetVertex() == targetId) {
+ outIterator.remove();
+ }
+ }
+
+ Iterator<StreamEdge> inIterator = inEdgeLists.get(targetId).iterator();
+ while (inIterator.hasNext()) {
+ StreamEdge edge = inIterator.next();
+
+ if (edge.getSourceVertex() == sourceId) {
+ inIterator.remove();
+ }
+ }
+ }
+
+ public List<StreamEdge> getOutEdges(int i) {
+ List<StreamEdge> outEdges = outEdgeLists.get(i);
+
+ if (outEdges == null) {
+ throw new RuntimeException("No such vertex in stream graph: " + i);
+ }
+
+ return outEdges;
+ }
+
+ public List<StreamEdge> getInEdges(int i) {
+ List<StreamEdge> inEdges = inEdgeLists.get(i);
+
+ if (inEdges == null) {
+ throw new RuntimeException("No such vertex in stream graph: " + i);
+ }
+
+ return inEdges;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/29a66155/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/StreamEdgeListTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/StreamEdgeListTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/StreamEdgeListTest.java
new file mode 100644
index 0000000..c4ba987
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/StreamEdgeListTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+
+public class StreamEdgeListTest {
+
+ private StreamEdgeList edgeList;
+
+ @Before
+ public void init() {
+ edgeList = new StreamEdgeList();
+ }
+
+ @Test
+ public void test() {
+ edgeList.addVertex(1);
+ edgeList.addVertex(2);
+ edgeList.addVertex(3);
+
+
+ // add edges
+ StreamEdge edge1 = new StreamEdge(1, 2, -1, null, null);
+ StreamEdge edge2 = new StreamEdge(2, 3, -1, null, null);
+ StreamEdge edge3 = new StreamEdge(1, 3, -1, null, null);
+
+ edgeList.addEdge(edge1);
+ edgeList.addEdge(edge2);
+ edgeList.addEdge(edge3);
+
+ // check adding
+ checkIfSameElements(edgeList.getOutEdges(1), Arrays.asList(edge1, edge3));
+ checkIfSameElements(edgeList.getOutEdges(2), Arrays.asList(edge2));
+ checkIfSameElements(edgeList.getOutEdges(3), new ArrayList<StreamEdge>());
+
+ checkIfSameElements(edgeList.getInEdges(1), new ArrayList<StreamEdge>());
+ checkIfSameElements(edgeList.getInEdges(2), Arrays.asList(edge1));
+ checkIfSameElements(edgeList.getInEdges(3), Arrays.asList(edge2, edge3));
+
+ // add duplicate edges
+ StreamEdge edge1new = new StreamEdge(1, 2, -2, null, null);
+ StreamEdge edge2new = new StreamEdge(2, 3, -2, null, null);
+
+ edgeList.addEdge(edge1new);
+ edgeList.addEdge(edge2new);
+
+ // check adding
+ checkIfSameElements(edgeList.getOutEdges(1), Arrays.asList(edge1, edge1new, edge3));
+ checkIfSameElements(edgeList.getOutEdges(2), Arrays.asList(edge2, edge2new));
+ checkIfSameElements(edgeList.getOutEdges(3), new ArrayList<StreamEdge>());
+
+ checkIfSameElements(edgeList.getInEdges(1), new ArrayList<StreamEdge>());
+ checkIfSameElements(edgeList.getInEdges(2), Arrays.asList(edge1, edge1new));
+ checkIfSameElements(edgeList.getInEdges(3), Arrays.asList(edge2, edge2new, edge3));
+
+ // remove a duplicate edge
+ edgeList.removeEdge(1, 2);
+
+ // check removing
+ checkIfSameElements(edgeList.getOutEdges(1), Arrays.asList(edge3));
+ checkIfSameElements(edgeList.getOutEdges(2), Arrays.asList(edge2, edge2new));
+
+ checkIfSameElements(edgeList.getInEdges(1), new ArrayList<StreamEdge>());
+ checkIfSameElements(edgeList.getInEdges(2), new ArrayList<StreamEdge>());
+
+ // add back an edge and delete a vertex
+ edgeList.addEdge(edge1);
+ edgeList.removeVertex(2);
+
+ // check removing
+ checkIfSameElements(edgeList.getOutEdges(1), Arrays.asList(edge3));
+ try {
+ checkIfSameElements(edgeList.getOutEdges(2), null);
+ fail();
+ } catch (RuntimeException e) {
+ }
+ checkIfSameElements(edgeList.getOutEdges(3), new ArrayList<StreamEdge>());
+
+ checkIfSameElements(edgeList.getInEdges(1), new ArrayList<StreamEdge>());
+ try {
+ checkIfSameElements(edgeList.getInEdges(2), null);
+ fail();
+ } catch (RuntimeException e) {
+ }
+ checkIfSameElements(edgeList.getInEdges(3), Arrays.asList(edge3));
+ }
+
+ private <T> void checkIfSameElements(List<T> expected, List<T> result) {
+ assertEquals(new HashSet<T>(expected), new HashSet<T>(result));
+ }
+}
\ No newline at end of file
[02/10] flink git commit: [FLINK-1594] [streaming] Fixed co-tasks
input handling
Posted by gy...@apache.org.
[FLINK-1594] [streaming] Fixed co-tasks input handling
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3158d1d4
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3158d1d4
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3158d1d4
Branch: refs/heads/master
Commit: 3158d1d46d92c166ffd3c7c8eafd899460315016
Parents: a8ba72b
Author: Gábor Hermann <re...@gmail.com>
Authored: Fri Feb 27 11:40:01 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:03 2015 +0100
----------------------------------------------------------------------
.../flink/streaming/api/StreamConfig.java | 73 ++++++++++++++------
.../apache/flink/streaming/api/StreamEdge.java | 3 +-
.../api/StreamingJobGraphGenerator.java | 43 ++++++------
.../api/streamvertex/CoStreamVertex.java | 7 +-
.../api/streamvertex/OutputHandler.java | 8 ++-
5 files changed, 89 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3158d1d4/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
index ea19a44..5b6de85 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamConfig.java
@@ -60,7 +60,11 @@ public class StreamConfig implements Serializable {
private static final String TYPE_SERIALIZER_OUT_2 = "typeSerializer_out_2";
private static final String ITERATON_WAIT = "iterationWait";
private static final String OUTPUTS = "outvertexIDs";
+ private static final String NONCHAINED_OUTPUTS = "NONCHAINED_OUTPUTS";
+ private static final String CHAINED_OUTPUT_EDGES = "CHAINED_OUTPUTS";
private static final String EDGES_IN_ORDER = "rwOrder";
+ private static final String OUT_STREAM_EDGES = "out stream edges";
+ private static final String IN_STREAM_EDGES = "out stream edges";
// DEFAULT VALUES
@@ -281,19 +285,60 @@ public class StreamConfig implements Serializable {
return config.getInteger(NUMBER_OF_OUTPUTS, 0);
}
- public void setOutputs(List<Integer> outputvertexIDs) {
- config.setBytes(OUTPUTS, SerializationUtils.serialize((Serializable) outputvertexIDs));
+ public void setNonChainedOutputs(List<StreamEdge> outputvertexIDs) {
+ config.setBytes(NONCHAINED_OUTPUTS, SerializationUtils.serialize((Serializable) outputvertexIDs));
}
@SuppressWarnings("unchecked")
- public List<Integer> getOutputs(ClassLoader cl) {
+ public List<StreamEdge> getNonChainedOutputs(ClassLoader cl) {
try {
- return (List<Integer>) InstantiationUtil.readObjectFromConfig(this.config, OUTPUTS, cl);
+ return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(this.config, NONCHAINED_OUTPUTS, cl);
} catch (Exception e) {
throw new RuntimeException("Could not instantiate outputs.");
}
}
+ public void setChainedOutputs(List<StreamEdge> chainedOutputs) {
+ config.setBytes(CHAINED_OUTPUTS,
+ SerializationUtils.serialize((Serializable) chainedOutputs));
+ }
+
+ @SuppressWarnings("unchecked")
+ public List<StreamEdge> getChainedOutputs(ClassLoader cl) {
+ try {
+ return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(this.config,
+ CHAINED_OUTPUTS, cl);
+ } catch (Exception e) {
+ throw new RuntimeException("Could not instantiate chained outputs.");
+ }
+ }
+
+ public void setOutEdges(List<StreamEdge> outEdges) {
+ config.setBytes(OUT_STREAM_EDGES, SerializationUtils.serialize((Serializable) outEdges));
+ }
+
+ public List<StreamEdge> getOutEdges(ClassLoader cl) {
+ try {
+ return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(
+ this.config, OUT_STREAM_EDGES, cl);
+ } catch (Exception e) {
+ throw new RuntimeException("Could not instantiate outputs.");
+ }
+ }
+
+ public void setInEdges(List<StreamEdge> inEdges) {
+ config.setBytes(IN_STREAM_EDGES, SerializationUtils.serialize((Serializable) inEdges));
+ }
+
+ public List<StreamEdge> getInEdges(ClassLoader cl) {
+ try {
+ return (List<StreamEdge>) InstantiationUtil.readObjectFromConfig(
+ this.config, IN_STREAM_EDGES, cl);
+ } catch (Exception e) {
+ throw new RuntimeException("Could not instantiate inputs.");
+ }
+ }
+
public void setOutEdgesInOrder(List<Tuple2<Integer, Integer>> outEdgeList) {
config.setBytes(EDGES_IN_ORDER, SerializationUtils.serialize((Serializable) outEdgeList));
@@ -329,21 +374,6 @@ public class StreamConfig implements Serializable {
return config.getInteger(INPUT_TYPE + inputNumber, 0);
}
- public void setChainedOutputs(List<Integer> chainedOutputs) {
- config.setBytes(CHAINED_OUTPUTS,
- SerializationUtils.serialize((Serializable) chainedOutputs));
- }
-
- @SuppressWarnings("unchecked")
- public List<Integer> getChainedOutputs(ClassLoader cl) {
- try {
- return (List<Integer>) InstantiationUtil.readObjectFromConfig(this.config,
- CHAINED_OUTPUTS, cl);
- } catch (Exception e) {
- throw new RuntimeException("Could not instantiate chained outputs.");
- }
- }
-
public void setTransitiveChainedTaskConfigs(Map<Integer, StreamConfig> chainedTaskConfigs) {
config.setBytes(CHAINED_TASK_CONFIG,
SerializationUtils.serialize((Serializable) chainedTaskConfigs));
@@ -382,9 +412,10 @@ public class StreamConfig implements Serializable {
builder.append("\nTask name: " + getVertexID());
builder.append("\nNumber of non-chained inputs: " + getNumberOfInputs());
builder.append("\nNumber of non-chained outputs: " + getNumberOfOutputs());
- builder.append("\nOutput names: " + getOutputs(cl));
+ builder.append("\nOutput names: " + getNonChainedOutputs(cl));
builder.append("\nPartitioning:");
- for (Integer outputname : getOutputs(cl)) {
+ for (StreamEdge output : getNonChainedOutputs(cl)) {
+ int outputname = output.getTargetVertex();
builder.append("\n\t" + outputname + ": " + getPartitioner(cl, outputname));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3158d1d4/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
index 479ae93..7363e08 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/StreamEdge.java
@@ -17,11 +17,12 @@
package org.apache.flink.streaming.api;
+import java.io.Serializable;
import java.util.List;
import org.apache.flink.streaming.partitioner.StreamPartitioner;
-public class StreamEdge {
+public class StreamEdge implements Serializable {
final private int sourceVertex;
final private int targetVertex;
http://git-wip-us.apache.org/repos/asf/flink/blob/3158d1d4/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 607d041..3ff64ca 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
@@ -101,25 +101,24 @@ public class StreamingJobGraphGenerator {
if (!builtVertices.contains(startNode)) {
List<Tuple2<Integer, Integer>> transitiveOutEdges = new ArrayList<Tuple2<Integer, Integer>>();
- List<Integer> chainableOutputs = new ArrayList<Integer>();
- List<Integer> nonChainableOutputs = new ArrayList<Integer>();
+ List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
+ List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();
for (StreamEdge outEdge : streamGraph.getOutEdges(current)) {
- Integer outID = outEdge.getTargetVertex();
- if (isChainable(current, outID)) {
- chainableOutputs.add(outID);
+ if (isChainable(outEdge)) {
+ chainableOutputs.add(outEdge);
} else {
- nonChainableOutputs.add(outID);
+ nonChainableOutputs.add(outEdge);
}
}
- for (Integer chainable : chainableOutputs) {
- transitiveOutEdges.addAll(createChain(startNode, chainable));
+ for (StreamEdge chainable : chainableOutputs) {
+ transitiveOutEdges.addAll(createChain(startNode, chainable.getTargetVertex()));
}
- for (Integer nonChainable : nonChainableOutputs) {
- transitiveOutEdges.add(new Tuple2<Integer, Integer>(current, nonChainable));
- createChain(nonChainable, nonChainable);
+ for (StreamEdge nonChainable : nonChainableOutputs) {
+ transitiveOutEdges.add(new Tuple2<Integer, Integer>(current, nonChainable.getTargetVertex()));
+ createChain(nonChainable.getTargetVertex(), nonChainable.getTargetVertex());
}
chainedNames.put(current, createChainedName(current, chainableOutputs));
@@ -133,6 +132,8 @@ public class StreamingJobGraphGenerator {
config.setChainStart();
config.setOutEdgesInOrder(transitiveOutEdges);
+ config.setOutEdges(streamGraph.getOutEdges(current));
+ config.setInEdges(streamGraph.getInEdges(current));
for (Tuple2<Integer, Integer> edge : transitiveOutEdges) {
connect(startNode, edge);
@@ -157,12 +158,12 @@ public class StreamingJobGraphGenerator {
}
}
- private String createChainedName(Integer vertexID, List<Integer> chainedOutputs) {
+ private String createChainedName(Integer vertexID, List<StreamEdge> chainedOutputs) {
String operatorName = streamGraph.getOperatorName(vertexID);
if (chainedOutputs.size() > 1) {
List<String> outputChainedNames = new ArrayList<String>();
- for (Integer chainable : chainedOutputs) {
- outputChainedNames.add(chainedNames.get(chainable));
+ for (StreamEdge chainable : chainedOutputs) {
+ outputChainedNames.add(chainedNames.get(chainable.getTargetVertex()));
}
return operatorName + " -> (" + StringUtils.join(outputChainedNames, ", ") + ")";
} else if (chainedOutputs.size() == 1) {
@@ -201,7 +202,7 @@ public class StreamingJobGraphGenerator {
}
private void setVertexConfig(Integer vertexID, StreamConfig config,
- List<Integer> chainableOutputs, List<Integer> nonChainableOutputs) {
+ List<StreamEdge> chainableOutputs, List<StreamEdge> nonChainableOutputs) {
config.setVertexID(vertexID);
config.setBufferTimeout(streamGraph.getBufferTimeout(vertexID));
@@ -215,7 +216,7 @@ public class StreamingJobGraphGenerator {
config.setOutputSelectors(streamGraph.getOutputSelector(vertexID));
config.setNumberOfOutputs(nonChainableOutputs.size());
- config.setOutputs(nonChainableOutputs);
+ config.setNonChainedOutputs(nonChainableOutputs);
config.setChainedOutputs(chainableOutputs);
config.setStateMonitoring(streamGraph.isMonitoringEnabled());
@@ -227,11 +228,11 @@ public class StreamingJobGraphGenerator {
config.setIterationWaitTime(streamGraph.getIterationTimeout(vertexID));
}
- List<Integer> allOutputs = new ArrayList<Integer>(chainableOutputs);
+ List<StreamEdge> allOutputs = new ArrayList<StreamEdge>(chainableOutputs);
allOutputs.addAll(nonChainableOutputs);
- for (Integer output : allOutputs) {
- config.setSelectedNames(output, streamGraph.getEdge(vertexID, output).getSelectedNames());
+ for (StreamEdge output : allOutputs) {
+ config.setSelectedNames(output.getTargetVertex(), streamGraph.getEdge(vertexID, output.getTargetVertex()).getSelectedNames());
}
vertexConfigs.put(vertexID, config);
@@ -274,7 +275,9 @@ public class StreamingJobGraphGenerator {
}
}
- private boolean isChainable(Integer vertexID, Integer outName) {
+ private boolean isChainable(StreamEdge edge) {
+ int vertexID = edge.getSourceVertex();
+ int outName = edge.getTargetVertex();
StreamInvokable<?, ?> headInvokable = streamGraph.getInvokable(vertexID);
StreamInvokable<?, ?> outInvokable = streamGraph.getInvokable(outName);
http://git-wip-us.apache.org/repos/asf/flink/blob/3158d1d4/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
index d794a35..b919e0f 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/CoStreamVertex.java
@@ -20,6 +20,8 @@ package org.apache.flink.streaming.api.streamvertex;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
+import org.apache.flink.streaming.api.StreamEdge;
+import org.apache.flink.streaming.api.invokable.operator.co.CoInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.io.CoReaderIterator;
@@ -29,6 +31,7 @@ import org.apache.flink.util.MutableObjectIterator;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.List;
public class CoStreamVertex<IN1, IN2, OUT> extends StreamVertex<IN1, OUT> {
@@ -78,8 +81,10 @@ public class CoStreamVertex<IN1, IN2, OUT> extends StreamVertex<IN1, OUT> {
ArrayList<InputGate> inputList1 = new ArrayList<InputGate>();
ArrayList<InputGate> inputList2 = new ArrayList<InputGate>();
+ List<StreamEdge> inEdges = configuration.getInEdges(userClassLoader);
+
for (int i = 0; i < numberOfInputs; i++) {
- int inputType = configuration.getInputIndex(i);
+ int inputType = inEdges.get(i).getTypeNumber();
InputGate reader = getEnvironment().getInputGate(i);
switch (inputType) {
case 1:
http://git-wip-us.apache.org/repos/asf/flink/blob/3158d1d4/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
index fd375f6..ca6b34d 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/OutputHandler.java
@@ -28,6 +28,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.StreamConfig;
+import org.apache.flink.streaming.api.StreamEdge;
import org.apache.flink.streaming.api.collector.CollectorWrapper;
import org.apache.flink.streaming.api.collector.DirectedCollectorWrapper;
import org.apache.flink.streaming.api.collector.StreamOutput;
@@ -117,7 +118,8 @@ public class OutputHandler<OUT> {
chainedTaskConfig.getOutputSelectors(cl)) : new CollectorWrapper<OUT>();
// Create collectors for the network outputs
- for (Integer output : chainedTaskConfig.getOutputs(cl)) {
+ for (StreamEdge outputEdge : chainedTaskConfig.getNonChainedOutputs(cl)) {
+ Integer output = outputEdge.getTargetVertex();
Collector<?> outCollector = outputMap.get(output);
@@ -130,7 +132,9 @@ public class OutputHandler<OUT> {
}
// Create collectors for the chained outputs
- for (Integer output : chainedTaskConfig.getChainedOutputs(cl)) {
+ for (StreamEdge outputEdge : chainedTaskConfig.getChainedOutputs(cl)) {
+ Integer output = outputEdge.getTargetVertex();
+
Collector<?> outCollector = createChainedCollector(chainedConfigs.get(output));
if (isDirectEmit) {
((DirectedCollectorWrapper<OUT>) wrapper).addCollector(outCollector,
[09/10] flink git commit: [FLINK-1618] [streaming] Parallel time
reduce
Posted by gy...@apache.org.
[FLINK-1618] [streaming] Parallel time reduce
Closes #485
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1377ca97
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1377ca97
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1377ca97
Branch: refs/heads/master
Commit: 1377ca97dd0d8d1bbb7224f562dfc4f68226e02b
Parents: 221e5e6
Author: Gyula Fora <gy...@apache.org>
Authored: Mon Mar 16 09:52:41 2015 +0100
Committer: Gyula Fora <gy...@apache.org>
Committed: Fri Mar 20 11:25:04 2015 +0100
----------------------------------------------------------------------
.../api/StreamingJobGraphGenerator.java | 28 +--
.../api/datastream/DiscretizedStream.java | 91 +++++++-
.../api/datastream/WindowedDataStream.java | 94 ++++++--
.../api/invokable/ChainableInvokable.java | 10 +
.../streaming/api/invokable/SinkInvokable.java | 2 +-
.../api/invokable/operator/FilterInvokable.java | 2 +-
.../invokable/operator/FlatMapInvokable.java | 2 +-
.../api/invokable/operator/MapInvokable.java | 2 +-
.../operator/StreamReduceInvokable.java | 2 +-
.../operator/windowing/EmptyWindowFilter.java | 32 +++
.../windowing/ParallelGroupedMerge.java | 41 ++++
.../operator/windowing/ParallelMerge.java | 142 +++++++++++
.../windowing/ParallelWindowPartitioner.java | 84 -------
.../operator/windowing/StreamDiscretizer.java | 13 +-
.../windowing/WindowBufferInvokable.java | 3 +-
.../operator/windowing/WindowMapper.java | 1 +
.../operator/windowing/WindowPartExtractor.java | 55 +++++
.../operator/windowing/WindowPartitioner.java | 6 +-
.../streaming/api/windowing/StreamWindow.java | 39 ++--
.../api/windowing/StreamWindowSerializer.java | 33 +--
.../streaming/api/windowing/WindowUtils.java | 7 +-
.../windowbuffer/BasicWindowBuffer.java | 13 +-
.../windowbuffer/CompletePreAggregator.java | 27 ---
.../windowing/windowbuffer/PreAggregator.java | 27 +++
.../windowbuffer/SlidingPreReducer.java | 16 +-
.../windowbuffer/TumblingGroupedPreReducer.java | 16 +-
.../windowbuffer/TumblingPreReducer.java | 17 +-
.../windowing/windowbuffer/WindowBuffer.java | 33 ++-
.../operator/windowing/ParallelMergeTest.java | 119 ++++++++++
.../windowing/WindowIntegrationTest.java | 234 +++++++++++++------
.../operator/windowing/WindowMergerTest.java | 6 +-
.../windowing/WindowPartitionerTest.java | 8 +-
.../api/windowing/StreamWindowTest.java | 8 +-
.../TumblingGroupedPreReducerTest.java | 26 +++
34 files changed, 915 insertions(+), 324 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/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 8a110bf..0146448 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
@@ -194,10 +194,12 @@ public class StreamingJobGraphGenerator {
for (StreamEdge chainable : chainedOutputs) {
outputChainedNames.add(chainedNames.get(chainable.getTargetVertex()));
}
- String returnOperatorName = operatorName + " -> (" + StringUtils.join(outputChainedNames, ", ") + ")";
+ String returnOperatorName = operatorName + " -> ("
+ + StringUtils.join(outputChainedNames, ", ") + ")";
return returnOperatorName;
} else if (chainedOutputs.size() == 1) {
- String returnOperatorName = operatorName + " -> " + chainedNames.get(chainedOutputs.get(0).getTargetVertex());
+ String returnOperatorName = operatorName + " -> "
+ + chainedNames.get(chainedOutputs.get(0).getTargetVertex());
return returnOperatorName;
} else {
return operatorName;
@@ -215,8 +217,7 @@ public class StreamingJobGraphGenerator {
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Parallelism set: {} for {}", streamGraph.getParallelism(vertexID),
- vertexID);
+ LOG.debug("Parallelism set: {} for {}", streamGraph.getParallelism(vertexID), vertexID);
}
if (streamGraph.getInputFormat(vertexID) != null) {
@@ -263,7 +264,8 @@ public class StreamingJobGraphGenerator {
allOutputs.addAll(nonChainableOutputs);
for (StreamEdge output : allOutputs) {
- config.setSelectedNames(output.getTargetVertex(), streamGraph.getEdge(vertexID, output.getTargetVertex()).getSelectedNames());
+ config.setSelectedNames(output.getTargetVertex(),
+ streamGraph.getEdge(vertexID, output.getTargetVertex()).getSelectedNames());
}
vertexConfigs.put(vertexID, config);
@@ -302,15 +304,15 @@ public class StreamingJobGraphGenerator {
StreamInvokable<?, ?> headInvokable = streamGraph.getInvokable(vertexID);
StreamInvokable<?, ?> outInvokable = streamGraph.getInvokable(outName);
- return
- streamGraph.getInEdges(outName).size() == 1
- && outInvokable != null
- && outInvokable.getChainingStrategy() == ChainingStrategy.ALWAYS
- && (headInvokable.getChainingStrategy() == ChainingStrategy.HEAD || headInvokable
+ return streamGraph.getInEdges(outName).size() == 1
+ && outInvokable != null
+ && outInvokable.getChainingStrategy() == ChainingStrategy.ALWAYS
+ && (headInvokable.getChainingStrategy() == ChainingStrategy.HEAD || headInvokable
.getChainingStrategy() == ChainingStrategy.ALWAYS)
- && edge.getPartitioner().getStrategy() == PartitioningStrategy.FORWARD
- && streamGraph.getParallelism(vertexID) == streamGraph.getParallelism(outName)
- && streamGraph.chaining;
+ && (edge.getPartitioner().getStrategy() == PartitioningStrategy.FORWARD || streamGraph
+ .getParallelism(outName) == 1)
+ && streamGraph.getParallelism(vertexID) == streamGraph.getParallelism(outName)
+ && streamGraph.chaining;
}
private void setSlotSharing() {
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/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 6526aa6..7597b47 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
@@ -18,21 +18,31 @@
package org.apache.flink.streaming.api.datastream;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.common.functions.FoldFunction;
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.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.api.function.WindowMapFunction;
+import org.apache.flink.streaming.api.function.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.invokable.StreamInvokable;
+import org.apache.flink.streaming.api.invokable.operator.FilterInvokable;
+import org.apache.flink.streaming.api.invokable.operator.FlatMapInvokable;
+import org.apache.flink.streaming.api.invokable.operator.co.CoFlatMapInvokable;
+import org.apache.flink.streaming.api.invokable.operator.windowing.EmptyWindowFilter;
+import org.apache.flink.streaming.api.invokable.operator.windowing.ParallelGroupedMerge;
+import org.apache.flink.streaming.api.invokable.operator.windowing.ParallelMerge;
import org.apache.flink.streaming.api.invokable.operator.windowing.WindowFlattener;
import org.apache.flink.streaming.api.invokable.operator.windowing.WindowFolder;
import org.apache.flink.streaming.api.invokable.operator.windowing.WindowMapper;
import org.apache.flink.streaming.api.invokable.operator.windowing.WindowMerger;
+import org.apache.flink.streaming.api.invokable.operator.windowing.WindowPartExtractor;
import org.apache.flink.streaming.api.invokable.operator.windowing.WindowPartitioner;
import org.apache.flink.streaming.api.invokable.operator.windowing.WindowReducer;
import org.apache.flink.streaming.api.windowing.StreamWindow;
@@ -91,6 +101,56 @@ public class DiscretizedStream<OUT> extends WindowedDataStream<OUT> {
}
}
+ /**
+ * This method implements the parallel time reduce logic for time windows
+ *
+ * @param reduceFunction
+ * The reduce function to be applied on the windows
+ * @param isPreAggregated
+ * Flag whether the window buffer was a pre-aggregator or not
+ * @return
+ */
+ protected DiscretizedStream<OUT> timeReduce(ReduceFunction<OUT> reduceFunction,
+ boolean isPreAggregated) {
+
+ // We partition the windowed stream if it is not already pre-aggregated
+ DiscretizedStream<OUT> partitioned = isPreAggregated ? this : partition(transformation);
+
+ // Since we also emit the empty windows for bookkeeping, we need to
+ // filter them out
+ DiscretizedStream<OUT> nonEmpty = filterEmpty(partitioned);
+
+ // We extract the number of parts from each window we will merge using
+ // this afterwards
+ DataStream<Tuple2<Integer, Integer>> numOfParts = extractPartsByID(partitioned);
+
+ // We reduce the windows if not pre-aggregated
+ DiscretizedStream<OUT> reduced = isPreAggregated ? nonEmpty : nonEmpty.transform(
+ WindowTransformation.REDUCEWINDOW, "Window Reduce", nonEmpty.getType(),
+ new WindowReducer<OUT>(reduceFunction));
+
+ // We merge the windows by the number of parts
+ return wrap(parallelMerge(numOfParts, reduced, reduceFunction), false);
+
+ }
+
+ private SingleOutputStreamOperator<StreamWindow<OUT>, ?> parallelMerge(
+ DataStream<Tuple2<Integer, Integer>> numOfParts, DiscretizedStream<OUT> reduced,
+ ReduceFunction<OUT> reduceFunction) {
+
+ CoFlatMapFunction<StreamWindow<OUT>, Tuple2<Integer, Integer>, StreamWindow<OUT>> parallelMerger = isGrouped() ? new ParallelGroupedMerge<OUT>()
+ : new ParallelMerge<OUT>(reduceFunction);
+
+ return reduced.discretizedStream
+ .groupBy(new WindowKey<OUT>())
+ .connect(numOfParts.groupBy(0))
+ .addCoFunction(
+ "CoFlatMap",
+ reduced.discretizedStream.getType(),
+ new CoFlatMapInvokable<StreamWindow<OUT>, Tuple2<Integer, Integer>, StreamWindow<OUT>>(
+ parallelMerger));
+ }
+
@Override
public <R> DiscretizedStream<R> mapWindow(WindowMapFunction<OUT, R> windowMapFunction) {
@@ -128,6 +188,23 @@ public class DiscretizedStream<OUT> extends WindowedDataStream<OUT> {
invokable), transformation);
}
+ private DiscretizedStream<OUT> filterEmpty(DiscretizedStream<OUT> input) {
+ return wrap(input.discretizedStream.transform("Filter", input.discretizedStream.getType(),
+ new FilterInvokable<StreamWindow<OUT>>(new EmptyWindowFilter<OUT>())
+ .withoutInputCopy()), input.isPartitioned);
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ private DataStream<Tuple2<Integer, Integer>> extractPartsByID(DiscretizedStream<OUT> input) {
+ return input.discretizedStream
+ .transform(
+ "ExtractParts",
+ new TupleTypeInfo(Tuple2.class, BasicTypeInfo.INT_TYPE_INFO,
+ BasicTypeInfo.INT_TYPE_INFO),
+ new FlatMapInvokable<StreamWindow<OUT>, Tuple2<Integer, Integer>>(
+ new WindowPartExtractor<OUT>()).withoutInputCopy());
+ }
+
private DiscretizedStream<OUT> partition(WindowTransformation transformation) {
int parallelism = discretizedStream.getParallelism();
@@ -173,13 +250,15 @@ public class DiscretizedStream<OUT> extends WindowedDataStream<OUT> {
}
- @SuppressWarnings("rawtypes")
- private <R> DiscretizedStream<R> wrap(SingleOutputStreamOperator stream, boolean isPartitioned) {
- return wrap(stream, transformation);
+ @SuppressWarnings("unchecked")
+ private <R> DiscretizedStream<R> wrap(SingleOutputStreamOperator<StreamWindow<R>, ?> stream,
+ boolean isPartitioned) {
+ return new DiscretizedStream<R>(stream, (KeySelector<R, ?>) this.groupByKey,
+ transformation, isPartitioned);
}
- @SuppressWarnings({ "unchecked", "rawtypes" })
- private <R> DiscretizedStream<R> wrap(SingleOutputStreamOperator stream,
+ @SuppressWarnings("unchecked")
+ private <R> DiscretizedStream<R> wrap(SingleOutputStreamOperator<StreamWindow<R>, ?> stream,
WindowTransformation transformation) {
return new DiscretizedStream<R>(stream, (KeySelector<R, ?>) this.groupByKey,
transformation, isPartitioned);
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/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 c80546f..73cbdfd 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
@@ -58,7 +58,7 @@ import org.apache.flink.streaming.api.windowing.policy.TimeTriggerPolicy;
import org.apache.flink.streaming.api.windowing.policy.TriggerPolicy;
import org.apache.flink.streaming.api.windowing.policy.TumblingEvictionPolicy;
import org.apache.flink.streaming.api.windowing.windowbuffer.BasicWindowBuffer;
-import org.apache.flink.streaming.api.windowing.windowbuffer.CompletePreAggregator;
+import org.apache.flink.streaming.api.windowing.windowbuffer.PreAggregator;
import org.apache.flink.streaming.api.windowing.windowbuffer.SlidingCountGroupedPreReducer;
import org.apache.flink.streaming.api.windowing.windowbuffer.SlidingCountPreReducer;
import org.apache.flink.streaming.api.windowing.windowbuffer.SlidingTimeGroupedPreReducer;
@@ -262,17 +262,25 @@ public class WindowedDataStream<OUT> {
*/
public DiscretizedStream<OUT> reduceWindow(ReduceFunction<OUT> reduceFunction) {
- WindowTransformation transformation = WindowTransformation.REDUCEWINDOW
- .with(clean(reduceFunction));
+ // We check whether we should apply parallel time discretization, which
+ // is a more complex exploiting the monotonic properties of time
+ // policies
+ if (WindowUtils.isTimeOnly(getTrigger(), getEviction()) && discretizerKey == null
+ && dataStream.getParallelism() > 1) {
+ return timeReduce(reduceFunction);
+ } else {
+ WindowTransformation transformation = WindowTransformation.REDUCEWINDOW
+ .with(clean(reduceFunction));
- WindowBuffer<OUT> windowBuffer = getWindowBuffer(transformation);
+ WindowBuffer<OUT> windowBuffer = getWindowBuffer(transformation);
- DiscretizedStream<OUT> discretized = discretize(transformation, windowBuffer);
+ DiscretizedStream<OUT> discretized = discretize(transformation, windowBuffer);
- if (windowBuffer instanceof CompletePreAggregator) {
- return discretized;
- } else {
- return discretized.reduceWindow(reduceFunction);
+ if (windowBuffer instanceof PreAggregator) {
+ return discretized;
+ } else {
+ return discretized.reduceWindow(reduceFunction);
+ }
}
}
@@ -377,25 +385,76 @@ public class WindowedDataStream<OUT> {
int parallelism = getDiscretizerParallelism(transformation);
return new DiscretizedStream<OUT>(dataStream
- .transform("Stream Discretizer", bufferEventType, discretizer)
+ .transform(discretizer.getClass().getSimpleName(), bufferEventType, discretizer)
.setParallelism(parallelism)
- .transform("WindowBuffer", new StreamWindowTypeInfo<OUT>(getType()),
- bufferInvokable).setParallelism(parallelism), groupByKey, transformation,
- false);
+ .transform(windowBuffer.getClass().getSimpleName(),
+ new StreamWindowTypeInfo<OUT>(getType()), bufferInvokable)
+ .setParallelism(parallelism), groupByKey, transformation, false);
}
+ /**
+ * Returns the degree of parallelism for the stream discretizer. The
+ * returned parallelism is either 1 for for non-parallel global policies (or
+ * when the input stream is non-parallel), environment parallelism for the
+ * policies that can run in parallel (such as, any ditributed policy, reduce
+ * by count or time).
+ *
+ * @param transformation
+ * The applied transformation
+ * @return The parallelism for the stream discretizer
+ */
private int getDiscretizerParallelism(WindowTransformation transformation) {
return isLocal
|| (transformation == WindowTransformation.REDUCEWINDOW && WindowUtils
.isParallelPolicy(getTrigger(), getEviction(), dataStream.getParallelism()))
|| (discretizerKey != null) ? dataStream.environment.getDegreeOfParallelism() : 1;
+
+ }
+
+ /**
+ * Dedicated method for applying parallel time reduce transformations on
+ * windows
+ *
+ * @param reduceFunction
+ * Reduce function to apply
+ * @return The transformed stream
+ */
+ protected DiscretizedStream<OUT> timeReduce(ReduceFunction<OUT> reduceFunction) {
+
+ WindowTransformation transformation = WindowTransformation.REDUCEWINDOW
+ .with(clean(reduceFunction));
+
+ // We get the windowbuffer and set it to emit empty windows with
+ // sequential IDs. This logic is necessarry to merge windows created in
+ // parallel.
+ WindowBuffer<OUT> windowBuffer = getWindowBuffer(transformation).emitEmpty().sequentialID();
+
+ // If there is a groupby for the reduce operation we apply it before the
+ // discretizers, because we will forward everything afterwards to
+ // exploit task chaining
+ if (groupByKey != null) {
+ dataStream = dataStream.groupBy(groupByKey);
+ }
+
+ // We discretize the stream and call the timeReduce function of the
+ // discretized stream, we also pass the type of the windowbuffer
+ DiscretizedStream<OUT> discretized = discretize(transformation, windowBuffer);
+
+ return discretized
+ .timeReduce(reduceFunction, windowBuffer instanceof PreAggregator);
+
}
+ /**
+ * Based on the defined policies, returns the stream discretizer to be used
+ */
private StreamInvokable<OUT, WindowEvent<OUT>> getDiscretizer() {
if (discretizerKey == null) {
return new StreamDiscretizer<OUT>(getTrigger(), getEviction());
} else if (WindowUtils.isSystemTimeTrigger(getTrigger())) {
+ // We return a special more efficient grouped discretizer for system
+ // time policies to avoid lunching multiple threads
return new GroupedTimeDiscretizer<OUT>(discretizerKey,
(TimeTriggerPolicy<OUT>) getTrigger(),
(CloneableEvictionPolicy<OUT>) getEviction());
@@ -416,6 +475,13 @@ public class WindowedDataStream<OUT> {
}
}
+ /**
+ * Based on the given policies returns the WindowBuffer used to store the
+ * elements in the window. This is the module that also encapsulates the
+ * pre-aggregator logic when it is applicable, reducing the space cost, and
+ * trigger latency.
+ *
+ */
@SuppressWarnings("unchecked")
private WindowBuffer<OUT> getWindowBuffer(WindowTransformation transformation) {
TriggerPolicy<OUT> trigger = getTrigger();
@@ -704,7 +770,7 @@ public class WindowedDataStream<OUT> {
if (evictionHelper != null) {
return evictionHelper.toEvict();
- } else if (userEvicter == null) {
+ } else if (userEvicter == null || userEvicter instanceof TumblingEvictionPolicy) {
if (triggerHelper instanceof Time) {
return triggerHelper.toEvict();
} else {
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java
index 24c0319..470fc81 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/ChainableInvokable.java
@@ -25,6 +25,7 @@ public abstract class ChainableInvokable<IN, OUT> extends StreamInvokable<IN, OU
Collector<IN> {
private static final long serialVersionUID = 1L;
+ private boolean copyInput = true;
public ChainableInvokable(Function userFunction) {
super(userFunction);
@@ -36,4 +37,13 @@ public abstract class ChainableInvokable<IN, OUT> extends StreamInvokable<IN, OU
this.inSerializer = inSerializer;
this.objectSerializer = inSerializer.getObjectSerializer();
}
+
+ public ChainableInvokable<IN, OUT> withoutInputCopy() {
+ copyInput = false;
+ return this;
+ }
+
+ protected IN copyInput(IN input) {
+ return copyInput ? copy(input) : input;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
index 35060fd..2c6b6e6 100755
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
@@ -43,7 +43,7 @@ public class SinkInvokable<IN> extends ChainableInvokable<IN, IN> {
@Override
public void collect(IN record) {
- nextObject = copy(record);
+ nextObject = copyInput(record);
callUserFunctionAndLogException();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
index ab3f147..610fa53 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
@@ -50,7 +50,7 @@ public class FilterInvokable<IN> extends ChainableInvokable<IN, IN> {
@Override
public void collect(IN record) {
if (isRunning) {
- nextObject = copy(record);
+ nextObject = copyInput(record);
callUserFunctionAndLogException();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
index 025bd32..436cf4e 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
@@ -45,7 +45,7 @@ public class FlatMapInvokable<IN, OUT> extends ChainableInvokable<IN, OUT> {
@Override
public void collect(IN record) {
if (isRunning) {
- nextObject = copy(record);
+ nextObject = copyInput(record);
callUserFunctionAndLogException();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
index 8fc1f13..9647144 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
@@ -45,7 +45,7 @@ public class MapInvokable<IN, OUT> extends ChainableInvokable<IN, OUT> {
@Override
public void collect(IN record) {
if (isRunning) {
- nextObject = copy(record);
+ nextObject = copyInput(record);
callUserFunctionAndLogException();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java
index e7fa2b1..fe58105 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/StreamReduceInvokable.java
@@ -63,7 +63,7 @@ public class StreamReduceInvokable<IN> extends ChainableInvokable<IN, IN> {
@Override
public void collect(IN record) {
if (isRunning) {
- nextObject = copy(record);
+ nextObject = copyInput(record);
callUserFunctionAndLogException();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/EmptyWindowFilter.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/EmptyWindowFilter.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/EmptyWindowFilter.java
new file mode 100644
index 0000000..0f2ee31
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/EmptyWindowFilter.java
@@ -0,0 +1,32 @@
+/*
+ * 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.api.common.functions.FilterFunction;
+import org.apache.flink.streaming.api.windowing.StreamWindow;
+
+public class EmptyWindowFilter<OUT> implements FilterFunction<StreamWindow<OUT>> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean filter(StreamWindow<OUT> value) throws Exception {
+ return !value.isEmpty();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelGroupedMerge.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelGroupedMerge.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelGroupedMerge.java
new file mode 100644
index 0000000..737485f
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelGroupedMerge.java
@@ -0,0 +1,41 @@
+/*
+ * 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.windowing.StreamWindow;
+
+/**
+ * The version of the ParallelMerge CoFlatMap that does not reduce the incoming
+ * elements only appends them to the current window. This is necessary for
+ * grouped reduces.
+ */
+public class ParallelGroupedMerge<OUT> extends ParallelMerge<OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ public ParallelGroupedMerge() {
+ super(null);
+ }
+
+ @Override
+ protected void updateCurrent(StreamWindow<OUT> current, StreamWindow<OUT> nextWindow)
+ throws Exception {
+ current.addAll(nextWindow);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMerge.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMerge.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMerge.java
new file mode 100644
index 0000000..8ffca91
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMerge.java
@@ -0,0 +1,142 @@
+/*
+ * 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.common.functions.ReduceFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.function.co.RichCoFlatMapFunction;
+import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.util.Collector;
+
+/**
+ * Class that encapsulates the functionality necessary to merge windows created
+ * in parallel. This CoFlatMap uses the information received on the number of
+ * parts for each window to merge the different parts. It waits until it
+ * receives an indication on the number of parts from all the discretizers
+ * before producing any output.
+ */
+public class ParallelMerge<OUT> extends
+ RichCoFlatMapFunction<StreamWindow<OUT>, Tuple2<Integer, Integer>, StreamWindow<OUT>> {
+
+ private static final long serialVersionUID = 1L;
+
+ protected Integer numberOfDiscretizers;
+ private ReduceFunction<OUT> reducer;
+
+ private Map<Integer, Integer> availableNumberOfParts = new HashMap<Integer, Integer>();
+ private Map<Integer, Tuple2<StreamWindow<OUT>, Integer>> receivedWindows = new HashMap<Integer, Tuple2<StreamWindow<OUT>, Integer>>();
+ private Map<Integer, Tuple2<Integer, Integer>> receivedNumberOfParts = new HashMap<Integer, Tuple2<Integer, Integer>>();
+
+ public ParallelMerge(ReduceFunction<OUT> reducer) {
+ this.reducer = reducer;
+ }
+
+ @Override
+ public void flatMap1(StreamWindow<OUT> nextWindow, Collector<StreamWindow<OUT>> out)
+ throws Exception {
+
+ Integer id = nextWindow.windowID;
+
+ Tuple2<StreamWindow<OUT>, Integer> current = receivedWindows.get(id);
+
+ if (current == null) {
+ current = new Tuple2<StreamWindow<OUT>, Integer>(nextWindow, 1);
+ } else {
+ updateCurrent(current.f0, nextWindow);
+ current.f1++;
+ }
+
+ Integer count = current.f1;
+
+ if (availableNumberOfParts.containsKey(id) && availableNumberOfParts.get(id) <= count) {
+ out.collect(current.f0);
+ receivedWindows.remove(id);
+ availableNumberOfParts.remove(id);
+
+ checkOld(id);
+
+ } else {
+ receivedWindows.put(id, (Tuple2<StreamWindow<OUT>, Integer>) current);
+ }
+ }
+
+ private void checkOld(Integer id) {
+ // In case we have remaining partial windows (which indicates errors in
+ // processing), output and log them
+ if (receivedWindows.containsKey(id - 1)) {
+ throw new RuntimeException("Error in processing logic, window with id " + id
+ + " should have already been processed");
+ }
+
+ }
+
+ @Override
+ public void flatMap2(Tuple2<Integer, Integer> partInfo, Collector<StreamWindow<OUT>> out)
+ throws Exception {
+
+ Integer id = partInfo.f0;
+ Integer numOfParts = partInfo.f1;
+
+ Tuple2<Integer, Integer> currentPartInfo = receivedNumberOfParts.get(id);
+ if (currentPartInfo != null) {
+ currentPartInfo.f0 += numOfParts;
+ currentPartInfo.f1++;
+ } else {
+ currentPartInfo = new Tuple2<Integer, Integer>(numOfParts, 1);
+ receivedNumberOfParts.put(id, currentPartInfo);
+ }
+
+ if (currentPartInfo.f1 >= numberOfDiscretizers) {
+ receivedNumberOfParts.remove(id);
+
+ Tuple2<StreamWindow<OUT>, Integer> current = receivedWindows.get(id);
+
+ Integer count = current != null ? current.f1 : -1;
+
+ if (count >= currentPartInfo.f0) {
+ out.collect(current.f0);
+ receivedWindows.remove(id);
+ checkOld(id);
+ } else if (currentPartInfo.f0 > 0) {
+ availableNumberOfParts.put(id, currentPartInfo.f1);
+ }
+ }
+
+ }
+
+ protected void updateCurrent(StreamWindow<OUT> current, StreamWindow<OUT> nextWindow)
+ throws Exception {
+ if (current.size() != 1 || nextWindow.size() != 1) {
+ throw new RuntimeException(
+ "Error in parallel merge logic. Current window should contain only one element.");
+ }
+ OUT currentReduced = current.remove(0);
+ currentReduced = reducer.reduce(currentReduced, nextWindow.get(0));
+ current.add(currentReduced);
+ }
+
+ @Override
+ public void open(Configuration conf) {
+ this.numberOfDiscretizers = getRuntimeContext().getNumberOfParallelSubtasks();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelWindowPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelWindowPartitioner.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelWindowPartitioner.java
deleted file mode 100644
index 32778da..0000000
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelWindowPartitioner.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.invokable.operator.windowing;
-
-import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.streaming.api.invokable.ChainableInvokable;
-import org.apache.flink.streaming.api.windowing.StreamWindow;
-
-/**
- * This invokable applies either split or key partitioning depending on the
- * transformation.
- */
-public class ParallelWindowPartitioner<T> extends
- ChainableInvokable<StreamWindow<T>, StreamWindow<T>> {
-
- private KeySelector<T, ?> keySelector;
- private int numberOfSplits;
- private int currentWindowID = 0;
-
- public ParallelWindowPartitioner(KeySelector<T, ?> keySelector) {
- super(null);
- this.keySelector = keySelector;
- }
-
- public ParallelWindowPartitioner(int numberOfSplits) {
- super(null);
- this.numberOfSplits = numberOfSplits;
- }
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public void invoke() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
- }
-
- @Override
- protected void callUserFunction() throws Exception {
- StreamWindow<T> currentWindow = nextObject;
- currentWindow.setID(++currentWindowID);
-
- if (keySelector == null) {
- if (numberOfSplits <= 1) {
- collector.collect(currentWindow);
- } else {
- for (StreamWindow<T> window : currentWindow.split(numberOfSplits)) {
- collector.collect(window);
- }
- }
- } else {
-
- for (StreamWindow<T> window : currentWindow.partitionBy(keySelector)) {
- collector.collect(window);
- }
-
- }
- }
-
- @Override
- public void collect(StreamWindow<T> record) {
- if (isRunning) {
- nextObject = record;
- callUserFunctionAndLogException();
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/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 e668b66..30512e6 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
@@ -25,13 +25,11 @@ import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerCallback;
import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerPolicy;
import org.apache.flink.streaming.api.windowing.policy.EvictionPolicy;
import org.apache.flink.streaming.api.windowing.policy.TriggerPolicy;
-import org.apache.flink.streaming.api.windowing.windowbuffer.WindowBuffer;
/**
* This invokable represents the discretization step of a window transformation.
* The user supplied eviction and trigger policies are applied to create the
* {@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, WindowEvent<IN>> {
@@ -129,11 +127,6 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, WindowEvent<IN>>
emitWindow();
}
- protected synchronized void externalTriggerOnFakeElement(Object input) {
- emitWindow();
- activeEvict(input);
- }
-
/**
* This method emits the content of the buffer as a new {@link StreamWindow}
* if not empty
@@ -175,7 +168,7 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, WindowEvent<IN>>
ActiveTriggerPolicy<IN> tp = (ActiveTriggerPolicy<IN>) triggerPolicy;
Runnable runnable = tp.createActiveTriggerRunnable(new WindowingCallback());
- if (activePolicyThread != null) {
+ if (runnable != null) {
activePolicyThread = new Thread(runnable);
activePolicyThread.start();
}
@@ -216,7 +209,7 @@ public class StreamDiscretizer<IN> extends StreamInvokable<IN, WindowEvent<IN>>
@Override
public String toString() {
- return "Discretizer(Trigger: " + triggerPolicy.toString() + ", Eviction: " + evictionPolicy.toString()
- + ")";
+ return "Discretizer(Trigger: " + triggerPolicy.toString() + ", Eviction: "
+ + evictionPolicy.toString() + ")";
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/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
index 75f7d9d..475611f 100644
--- 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
@@ -23,8 +23,7 @@ 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
+ * This invokable manages the window buffers attached to the discretizers.
*/
public class WindowBufferInvokable<T> extends ChainableInvokable<WindowEvent<T>, StreamWindow<T>> {
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMapper.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMapper.java
index 3dfd59d..9578a70 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMapper.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowMapper.java
@@ -50,6 +50,7 @@ public class WindowMapper<IN, OUT> extends MapInvokable<StreamWindow<IN>, Stream
@Override
public StreamWindow<R> map(StreamWindow<T> window) throws Exception {
StreamWindow<R> outputWindow = new StreamWindow<R>(window.windowID);
+
outputWindow.numberOfParts = window.numberOfParts;
mapper.mapWindow(window, outputWindow);
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartExtractor.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartExtractor.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartExtractor.java
new file mode 100644
index 0000000..416b915
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartExtractor.java
@@ -0,0 +1,55 @@
+/*
+ * 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.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.util.Collector;
+
+/**
+ * This FlatMapFunction is used to send the number of parts for each window ID
+ * (for each parallel discretizer) to the parallel merger that will use is to
+ * merge parallel discretized windows
+ */
+public class WindowPartExtractor<OUT> implements FlatMapFunction<StreamWindow<OUT>, Tuple2<Integer, Integer>> {
+
+ private static final long serialVersionUID = 1L;
+
+ Integer lastIndex = -1;
+
+ @Override
+ public void flatMap(StreamWindow<OUT> value, Collector<Tuple2<Integer, Integer>> out)
+ throws Exception {
+
+ // We dont emit new values for the same index, this avoids sending the
+ // same information for the same partitioned window multiple times
+ if (value.windowID != lastIndex) {
+
+ // For empty windows we send 0 since these windows will be filtered
+ // out
+ if (value.isEmpty()) {
+ out.collect(new Tuple2<Integer, Integer>(value.windowID, 0));
+ } else {
+ out.collect(new Tuple2<Integer, Integer>(value.windowID, value.numberOfParts));
+ }
+ lastIndex = value.windowID;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitioner.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitioner.java
index 846650d..0a28d99 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitioner.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/windowing/WindowPartitioner.java
@@ -56,13 +56,14 @@ public class WindowPartitioner<T> extends ChainableInvokable<StreamWindow<T>, St
if (numberOfSplits <= 1) {
collector.collect(currentWindow);
} else {
- for (StreamWindow<T> window : currentWindow.split(numberOfSplits)) {
+ for (StreamWindow<T> window : StreamWindow.split(currentWindow, numberOfSplits)) {
collector.collect(window);
}
}
} else {
- for (StreamWindow<T> window : currentWindow.partitionBy(keySelector)) {
+ for (StreamWindow<T> window : StreamWindow
+ .partitionBy(currentWindow, keySelector, true)) {
collector.collect(window);
}
@@ -76,5 +77,4 @@ public class WindowPartitioner<T> extends ChainableInvokable<StreamWindow<T>, St
callUserFunctionAndLogException();
}
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindow.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindow.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindow.java
index b45babb..ee2ea06 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindow.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/StreamWindow.java
@@ -42,7 +42,6 @@ public class StreamWindow<T> extends ArrayList<T> implements Collector<T> {
private static Random rnd = new Random();
public int windowID;
-
public int numberOfParts;
/**
@@ -104,27 +103,33 @@ public class StreamWindow<T> extends ArrayList<T> implements Collector<T> {
* Partitions the window using the given keyselector. A subwindow will be
* created for each key.
*
+ * @param streamWindow
+ * StreamWindow instance to partition
* @param keySelector
* The keyselector used for extracting keys.
+ * @param withKey
+ * Flag to decide whether the key object should be included in
+ * the created window
* @return A list of the subwindows
*/
- public List<StreamWindow<T>> partitionBy(KeySelector<T, ?> keySelector) throws Exception {
- Map<Object, StreamWindow<T>> partitions = new HashMap<Object, StreamWindow<T>>();
+ public static <X> List<StreamWindow<X>> partitionBy(StreamWindow<X> streamWindow,
+ KeySelector<X, ?> keySelector, boolean withKey) throws Exception {
+ Map<Object, StreamWindow<X>> partitions = new HashMap<Object, StreamWindow<X>>();
- for (T value : this) {
+ for (X value : streamWindow) {
Object key = keySelector.getKey(value);
- StreamWindow<T> window = partitions.get(key);
+ StreamWindow<X> window = partitions.get(key);
if (window == null) {
- window = new StreamWindow<T>(this.windowID, 0);
+ window = new StreamWindow<X>(streamWindow.windowID, 0);
partitions.put(key, window);
}
window.add(value);
}
- List<StreamWindow<T>> output = new ArrayList<StreamWindow<T>>();
+ List<StreamWindow<X>> output = new ArrayList<StreamWindow<X>>();
int numkeys = partitions.size();
- for (StreamWindow<T> window : partitions.values()) {
+ for (StreamWindow<X> window : partitions.values()) {
output.add(window.setNumberOfParts(numkeys));
}
@@ -134,30 +139,32 @@ public class StreamWindow<T> extends ArrayList<T> implements Collector<T> {
/**
* Splits the window into n equal (if possible) sizes.
*
+ * @param window
+ * Window to split
* @param n
* Number of desired partitions
* @return The list of subwindows.
*/
- public List<StreamWindow<T>> split(int n) {
- int numElements = size();
+ public static <X> List<StreamWindow<X>> split(StreamWindow<X> window, int n) {
+ int numElements = window.size();
if (n == 0) {
- return new ArrayList<StreamWindow<T>>();
+ return new ArrayList<StreamWindow<X>>();
}
if (n > numElements) {
- return split(numElements);
+ return split(window, numElements);
} else {
- List<StreamWindow<T>> split = new ArrayList<StreamWindow<T>>();
+ List<StreamWindow<X>> split = new ArrayList<StreamWindow<X>>();
int splitSize = numElements / n;
int index = -1;
- StreamWindow<T> currentSubWindow = new StreamWindow<T>(windowID, n);
+ StreamWindow<X> currentSubWindow = new StreamWindow<X>(window.windowID, n);
split.add(currentSubWindow);
- for (T element : this) {
+ for (X element : window) {
index++;
if (index == splitSize && split.size() < n) {
- currentSubWindow = new StreamWindow<T>(windowID, n);
+ currentSubWindow = new StreamWindow<X>(window.windowID, n);
split.add(currentSubWindow);
index = 0;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/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 e8945f1..229cb4a 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
@@ -23,6 +23,7 @@ import java.io.IOException;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.BooleanSerializer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
@@ -33,6 +34,7 @@ public final class StreamWindowSerializer<T> extends TypeSerializer<StreamWindow
private final TypeSerializer<T> typeSerializer;
TypeSerializer<Integer> intSerializer = IntSerializer.INSTANCE;
+ TypeSerializer<Boolean> boolSerializer = BooleanSerializer.INSTANCE;
public StreamWindowSerializer(TypeInformation<T> typeInfo, ExecutionConfig conf) {
this.typeSerializer = typeInfo.createSerializer(conf);
@@ -62,7 +64,6 @@ public final class StreamWindowSerializer<T> extends TypeSerializer<StreamWindow
reuse.clear();
reuse.windowID = from.windowID;
reuse.numberOfParts = from.numberOfParts;
-
for (T element : from) {
reuse.add(typeSerializer.copy(element));
}
@@ -75,31 +76,21 @@ public final class StreamWindowSerializer<T> extends TypeSerializer<StreamWindow
}
@Override
- public void serialize(StreamWindow<T> value, DataOutputView target) throws IOException {
+ public void serialize(StreamWindow<T> window, DataOutputView target) throws IOException {
+
+ intSerializer.serialize(window.windowID, target);
+ intSerializer.serialize(window.numberOfParts, target);
- intSerializer.serialize(value.windowID, target);
- intSerializer.serialize(value.numberOfParts, target);
- intSerializer.serialize(value.size(), target);
+ intSerializer.serialize(window.size(), target);
- for (T element : value) {
+ for (T element : window) {
typeSerializer.serialize(element, target);
}
}
@Override
public StreamWindow<T> deserialize(DataInputView source) throws IOException {
- StreamWindow<T> window = createInstance();
-
- window.windowID = intSerializer.deserialize(source);
- window.numberOfParts = intSerializer.deserialize(source);
-
- int size = intSerializer.deserialize(source);
-
- for (int i = 0; i < size; i++) {
- window.add(typeSerializer.deserialize(source));
- }
-
- return window;
+ return deserialize(createInstance(), source);
}
@Override
@@ -109,10 +100,10 @@ public final class StreamWindowSerializer<T> extends TypeSerializer<StreamWindow
StreamWindow<T> window = reuse;
window.clear();
- window.windowID = source.readInt();
- window.numberOfParts = source.readInt();
+ window.windowID = intSerializer.deserialize(source);
+ window.numberOfParts = intSerializer.deserialize(source);
- int size = source.readInt();
+ int size = intSerializer.deserialize(source);
for (int i = 0; i < size; i++) {
window.add(typeSerializer.deserialize(source));
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowUtils.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowUtils.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowUtils.java
index 8411d31..034c7d7 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowUtils.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/WindowUtils.java
@@ -45,9 +45,10 @@ public class WindowUtils {
}
public static boolean isParallelPolicy(TriggerPolicy<?> trigger, EvictionPolicy<?> eviction,
- int inputParallelism) {
- return inputParallelism != 1
- && ((eviction instanceof CountEvictionPolicy && (trigger instanceof CountTriggerPolicy || trigger instanceof TimeTriggerPolicy)) || (eviction instanceof TumblingEvictionPolicy && trigger instanceof CountTriggerPolicy));
+ int parallelism) {
+ return ((eviction instanceof CountEvictionPolicy && (trigger instanceof CountTriggerPolicy || trigger instanceof TimeTriggerPolicy))
+ || (eviction instanceof TumblingEvictionPolicy && trigger instanceof CountTriggerPolicy) || (WindowUtils
+ .isTimeOnly(trigger, eviction) && parallelism > 1));
}
public static boolean isSlidingTimePolicy(TriggerPolicy<?> trigger, EvictionPolicy<?> eviction) {
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/BasicWindowBuffer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/BasicWindowBuffer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/BasicWindowBuffer.java
index 8e39398..371e20d 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/BasicWindowBuffer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/BasicWindowBuffer.java
@@ -27,7 +27,7 @@ import org.apache.flink.util.Collector;
* Basic window buffer that stores the elements in a simple list without any
* pre-aggregation.
*/
-public class BasicWindowBuffer<T> implements WindowBuffer<T> {
+public class BasicWindowBuffer<T> extends WindowBuffer<T> {
private static final long serialVersionUID = 1L;
protected LinkedList<T> buffer;
@@ -36,15 +36,12 @@ public class BasicWindowBuffer<T> implements WindowBuffer<T> {
this.buffer = new LinkedList<T>();
}
- public boolean emitWindow(Collector<StreamWindow<T>> collector) {
- if (!buffer.isEmpty()) {
- StreamWindow<T> currentWindow = new StreamWindow<T>();
+ public void emitWindow(Collector<StreamWindow<T>> collector) {
+ if (emitEmpty || !buffer.isEmpty()) {
+ StreamWindow<T> currentWindow = createEmptyWindow();
currentWindow.addAll(buffer);
collector.collect(currentWindow);
- return true;
- } else {
- return false;
- }
+ }
}
public void store(T element) throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/CompletePreAggregator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/CompletePreAggregator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/CompletePreAggregator.java
deleted file mode 100644
index 59bd974..0000000
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/CompletePreAggregator.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUStreamRecord<?>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.windowbuffer;
-
-/**
- * Interface for marking window pre-aggregators that fully process the window so
- * that no further reduce step is necessary afterwards.
- */
-public interface CompletePreAggregator {
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/PreAggregator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/PreAggregator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/PreAggregator.java
new file mode 100644
index 0000000..1b95248
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/PreAggregator.java
@@ -0,0 +1,27 @@
+/*
+ * 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,
+ * WITHOUStreamRecord<?>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.windowbuffer;
+
+/**
+ * Interface for marking window pre-aggregators that fully process the window so
+ * that no further reduce step is necessary afterwards.
+ */
+public interface PreAggregator {
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java
index 27f7ff5..8b9558f 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java
@@ -24,7 +24,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.streaming.api.windowing.StreamWindow;
import org.apache.flink.util.Collector;
-public abstract class SlidingPreReducer<T> implements WindowBuffer<T>, CompletePreAggregator {
+public abstract class SlidingPreReducer<T> extends WindowBuffer<T> implements PreAggregator {
private static final long serialVersionUID = 1L;
@@ -45,18 +45,14 @@ public abstract class SlidingPreReducer<T> implements WindowBuffer<T>, CompleteP
this.serializer = serializer;
}
- public boolean emitWindow(Collector<StreamWindow<T>> collector) {
- StreamWindow<T> currentWindow = new StreamWindow<T>();
+ public void emitWindow(Collector<StreamWindow<T>> collector) {
+ StreamWindow<T> currentWindow = createEmptyWindow();
try {
- if (addFinalAggregate(currentWindow)) {
+ if (addFinalAggregate(currentWindow) || emitEmpty) {
collector.collect(currentWindow);
- afterEmit();
- return true;
- } else {
- afterEmit();
- return false;
- }
+ }
+ afterEmit();
} catch (Exception e) {
throw new RuntimeException(e);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducer.java
index 9431a99..68f9837 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingGroupedPreReducer.java
@@ -29,7 +29,7 @@ import org.apache.flink.util.Collector;
/**
* Grouped pre-reducer for tumbling eviction polciy.
*/
-public class TumblingGroupedPreReducer<T> implements WindowBuffer<T>, CompletePreAggregator {
+public class TumblingGroupedPreReducer<T> extends WindowBuffer<T> implements PreAggregator {
private static final long serialVersionUID = 1L;
@@ -37,6 +37,7 @@ public class TumblingGroupedPreReducer<T> implements WindowBuffer<T>, CompletePr
private KeySelector<T, ?> keySelector;
private Map<Object, T> reducedValues;
+ private Map<Object, T> keyInstancePerKey = new HashMap<Object, T>();
private TypeSerializer<T> serializer;
@@ -48,16 +49,15 @@ public class TumblingGroupedPreReducer<T> implements WindowBuffer<T>, CompletePr
this.reducedValues = new HashMap<Object, T>();
}
- public boolean emitWindow(Collector<StreamWindow<T>> collector) {
+ public void emitWindow(Collector<StreamWindow<T>> collector) {
if (!reducedValues.isEmpty()) {
- StreamWindow<T> currentWindow = new StreamWindow<T>();
+ StreamWindow<T> currentWindow = createEmptyWindow();
currentWindow.addAll(reducedValues.values());
collector.collect(currentWindow);
reducedValues.clear();
- return true;
- } else {
- return false;
+ } else if (emitEmpty) {
+ collector.collect(createEmptyWindow());
}
}
@@ -74,6 +74,10 @@ public class TumblingGroupedPreReducer<T> implements WindowBuffer<T>, CompletePr
}
reducedValues.put(key, reduced);
+
+ if (emitPerGroup && !keyInstancePerKey.containsKey(key)) {
+ keyInstancePerKey.put(key, element);
+ }
}
public void evict(int n) {
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingPreReducer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingPreReducer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingPreReducer.java
index 58b30a6..e56e556 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingPreReducer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/TumblingPreReducer.java
@@ -25,7 +25,7 @@ import org.apache.flink.util.Collector;
/**
* Non-grouped pre-reducer for tumbling eviction policy.
*/
-public class TumblingPreReducer<T> implements WindowBuffer<T>, CompletePreAggregator {
+public class TumblingPreReducer<T> extends WindowBuffer<T> implements PreAggregator {
private static final long serialVersionUID = 1L;
@@ -39,15 +39,14 @@ public class TumblingPreReducer<T> implements WindowBuffer<T>, CompletePreAggreg
this.serializer = serializer;
}
- public boolean emitWindow(Collector<StreamWindow<T>> collector) {
+ public void emitWindow(Collector<StreamWindow<T>> collector) {
if (reduced != null) {
- StreamWindow<T> currentWindow = new StreamWindow<T>();
+ StreamWindow<T> currentWindow = createEmptyWindow();
currentWindow.add(reduced);
collector.collect(currentWindow);
reduced = null;
- return true;
- } else {
- return false;
+ } else if (emitEmpty) {
+ collector.collect(createEmptyWindow());
}
}
@@ -72,4 +71,10 @@ public class TumblingPreReducer<T> implements WindowBuffer<T>, CompletePreAggreg
return reduced.toString();
}
+ @Override
+ public WindowBuffer<T> emitEmpty() {
+ emitEmpty = true;
+ return this;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/WindowBuffer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/WindowBuffer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/WindowBuffer.java
index 2dd50db..5c5ea52 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/WindowBuffer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/WindowBuffer.java
@@ -23,17 +23,38 @@ import org.apache.flink.streaming.api.windowing.StreamWindow;
import org.apache.flink.util.Collector;
/**
- * Interface for defining specialized buffers to store/emit window data.
+ * Class for defining specialized buffers to store/emit window data.
* Pre-aggregators should be implemented using this interface.
*/
-public interface WindowBuffer<T> extends Serializable, Cloneable {
+public abstract class WindowBuffer<T> implements Serializable, Cloneable {
- public void store(T element) throws Exception;
+ private static final long serialVersionUID = 1L;
- public void evict(int n);
+ protected Integer nextID = 1;
+ protected boolean sequentialID = false;
+ protected boolean emitEmpty = false;
+ protected boolean emitPerGroup = false;
- public boolean emitWindow(Collector<StreamWindow<T>> collector);
+ public abstract void store(T element) throws Exception;
- public WindowBuffer<T> clone();
+ public abstract void evict(int n);
+
+ public abstract void emitWindow(Collector<StreamWindow<T>> collector);
+
+ public abstract WindowBuffer<T> clone();
+
+ public WindowBuffer<T> emitEmpty() {
+ emitEmpty = true;
+ return this;
+ }
+
+ public WindowBuffer<T> sequentialID() {
+ sequentialID = true;
+ return this;
+ }
+
+ protected StreamWindow<T> createEmptyWindow() {
+ return sequentialID ? new StreamWindow<T>(nextID++) : new StreamWindow<T>();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1377ca97/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMergeTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMergeTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMergeTest.java
new file mode 100644
index 0000000..d892c48
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/windowing/ParallelMergeTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.List;
+
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.streaming.api.windowing.windowbuffer.BasicWindowBufferTest.TestCollector;
+import org.junit.Test;
+
+public class ParallelMergeTest {
+
+ @Test
+ public void nonGroupedTest() throws Exception {
+
+ ReduceFunction<Integer> reducer = new ReduceFunction<Integer>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Integer reduce(Integer a, Integer b) throws Exception {
+ return a + b;
+ }
+ };
+
+ TestCollector<StreamWindow<Integer>> out = new TestCollector<StreamWindow<Integer>>();
+ List<StreamWindow<Integer>> output = out.getCollected();
+
+ ParallelMerge<Integer> merger = new ParallelMerge<Integer>(reducer);
+ merger.numberOfDiscretizers = 2;
+
+ merger.flatMap1(createTestWindow(1), out);
+ merger.flatMap1(createTestWindow(1), out);
+ merger.flatMap2(new Tuple2<Integer, Integer>(1, 1), out);
+ assertTrue(output.isEmpty());
+ merger.flatMap2(new Tuple2<Integer, Integer>(1, 1), out);
+ assertEquals(StreamWindow.fromElements(2), output.get(0));
+
+ merger.flatMap2(new Tuple2<Integer, Integer>(2, 2), out);
+ merger.flatMap1(createTestWindow(2), out);
+ merger.flatMap1(createTestWindow(2), out);
+ merger.flatMap2(new Tuple2<Integer, Integer>(2, 1), out);
+ assertEquals(1, output.size());
+ merger.flatMap1(createTestWindow(2), out);
+ assertEquals(StreamWindow.fromElements(3), output.get(1));
+
+ // check error handling
+ merger.flatMap1(createTestWindow(3), out);
+ merger.flatMap2(new Tuple2<Integer, Integer>(3, 1), out);
+ merger.flatMap2(new Tuple2<Integer, Integer>(3, 1), out);
+
+ merger.flatMap2(new Tuple2<Integer, Integer>(4, 1), out);
+ merger.flatMap2(new Tuple2<Integer, Integer>(4, 1), out);
+ merger.flatMap1(createTestWindow(4), out);
+ try {
+ merger.flatMap1(createTestWindow(4), out);
+ fail();
+ } catch (RuntimeException e) {
+ // Do nothing
+ }
+
+ ParallelMerge<Integer> merger2 = new ParallelMerge<Integer>(reducer);
+ merger2.numberOfDiscretizers = 2;
+ merger2.flatMap1(createTestWindow(0), out);
+ merger2.flatMap1(createTestWindow(1), out);
+ merger2.flatMap1(createTestWindow(1), out);
+ merger2.flatMap2(new Tuple2<Integer, Integer>(1, 1), out);
+ try {
+ merger2.flatMap2(new Tuple2<Integer, Integer>(1, 1), out);
+ fail();
+ } catch (RuntimeException e) {
+ // Do nothing
+ }
+
+ }
+
+ @Test
+ public void groupedTest() throws Exception {
+
+ TestCollector<StreamWindow<Integer>> out = new TestCollector<StreamWindow<Integer>>();
+ List<StreamWindow<Integer>> output = out.getCollected();
+
+ ParallelMerge<Integer> merger = new ParallelGroupedMerge<Integer>();
+ merger.numberOfDiscretizers = 2;
+
+ merger.flatMap1(createTestWindow(1), out);
+ merger.flatMap1(createTestWindow(1), out);
+ merger.flatMap2(new Tuple2<Integer, Integer>(1, 1), out);
+ assertTrue(output.isEmpty());
+ merger.flatMap2(new Tuple2<Integer, Integer>(1, 1), out);
+ assertEquals(StreamWindow.fromElements(1, 1), output.get(0));
+ }
+
+ private StreamWindow<Integer> createTestWindow(Integer id) {
+ StreamWindow<Integer> ret = new StreamWindow<Integer>(id);
+ ret.add(1);
+ return ret;
+ }
+}