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/05/19 22:23:52 UTC
[5/6] flink git commit: [FLINK-1986] [streaming] Iterative stream
creation bugfix
[FLINK-1986] [streaming] Iterative stream creation bugfix
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2d3e69a2
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2d3e69a2
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2d3e69a2
Branch: refs/heads/master
Commit: 2d3e69a25e753b2eb825b589fe2550d2b968e161
Parents: 6edf31a
Author: Gyula Fora <gy...@apache.org>
Authored: Mon May 18 16:36:14 2015 +0200
Committer: Gyula Fora <gy...@apache.org>
Committed: Tue May 19 18:33:14 2015 +0200
----------------------------------------------------------------------
.../streaming/api/datastream/DataStream.java | 16 ++++++++
.../api/datastream/IterativeDataStream.java | 43 ++------------------
2 files changed, 19 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2d3e69a2/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 03cdaa5..dbb9b05 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
@@ -113,6 +113,9 @@ public class DataStream<OUT> {
@SuppressWarnings("rawtypes")
protected TypeInformation typeInfo;
protected List<DataStream<OUT>> mergedStreams;
+
+ protected Integer iterationID = null;
+ protected Long iterationWaitTime = null;
protected final StreamGraph streamGraph;
private boolean typeUsed;
@@ -160,6 +163,8 @@ public class DataStream<OUT> {
this.partitioner = dataStream.partitioner.copy();
this.streamGraph = dataStream.streamGraph;
this.typeInfo = dataStream.typeInfo;
+ this.iterationID = dataStream.iterationID;
+ this.iterationWaitTime = dataStream.iterationWaitTime;
this.mergedStreams = new ArrayList<DataStream<OUT>>();
this.mergedStreams.add(this);
if (dataStream.mergedStreams.size() > 1) {
@@ -1224,9 +1229,20 @@ public class DataStream<OUT> {
operatorName);
connectGraph(inputStream, returnStream.getId(), 0);
+
+ if (iterationID != null) {
+ //This data stream is an input to some iteration
+ addIterationSource(returnStream);
+ }
return returnStream;
}
+
+ private <X> void addIterationSource(DataStream<X> dataStream) {
+ Integer id = ++counter;
+ streamGraph.addIterationHead(id, dataStream.getId(), iterationID, iterationWaitTime);
+ streamGraph.setParallelism(id, dataStream.getParallelism());
+ }
/**
* Internal function for setting the partitioner for the DataStream
http://git-wip-us.apache.org/repos/asf/flink/blob/2d3e69a2/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/IterativeDataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/IterativeDataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/IterativeDataStream.java
index b1bdb85..178f2eb 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/IterativeDataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/IterativeDataStream.java
@@ -17,9 +17,6 @@
package org.apache.flink.streaming.api.datastream;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-
/**
* The iterative data stream represents the start of an iteration in a
* {@link DataStream}.
@@ -31,21 +28,13 @@ public class IterativeDataStream<IN> extends
SingleOutputStreamOperator<IN, IterativeDataStream<IN>> {
static Integer iterationCount = 0;
- protected Integer iterationID;
- protected long waitTime;
-
+
protected IterativeDataStream(DataStream<IN> dataStream, long maxWaitTime) {
super(dataStream);
setBufferTimeout(dataStream.environment.getBufferTimeout());
iterationID = iterationCount;
iterationCount++;
- waitTime = maxWaitTime;
- }
-
- protected IterativeDataStream(DataStream<IN> dataStream, Integer iterationID, long waitTime) {
- super(dataStream);
- this.iterationID = iterationID;
- this.waitTime = waitTime;
+ iterationWaitTime = maxWaitTime;
}
/**
@@ -70,35 +59,9 @@ public class IterativeDataStream<IN> extends
// We add an iteration sink to the tail which will send tuples to the
// iteration head
streamGraph.addIterationTail(iterationSink.getId(), iterationTail.getId(), iterationID,
- waitTime);
+ iterationWaitTime);
connectGraph(iterationTail.forward(), iterationSink.getId(), 0);
return iterationTail;
}
-
- @Override
- public <R> SingleOutputStreamOperator<R, ?> transform(String operatorName,
- TypeInformation<R> outTypeInfo, OneInputStreamOperator<IN, R> operator) {
-
- // We call the superclass tranform method
- SingleOutputStreamOperator<R, ?> returnStream = super.transform(operatorName, outTypeInfo,
- operator);
-
- // Then we add a source that will take care of receiving feedback tuples
- // from the tail
- addIterationSource(returnStream);
-
- return returnStream;
- }
-
- private <X> void addIterationSource(DataStream<X> dataStream) {
- Integer id = ++counter;
- streamGraph.addIterationHead(id, dataStream.getId(), iterationID, waitTime);
- streamGraph.setParallelism(id, dataStream.getParallelism());
- }
-
- @Override
- public IterativeDataStream<IN> copy() {
- return new IterativeDataStream<IN>(this, iterationID, waitTime);
- }
}