You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2014/08/18 19:25:44 UTC
[07/51] [abbrv] git commit: [streaming] Serialization rework to reuse
objects
[streaming] Serialization rework to reuse objects
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/0c1ef6d2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/0c1ef6d2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/0c1ef6d2
Branch: refs/heads/master
Commit: 0c1ef6d2dcb4b0cb9effc602411ec53fd9bb9ada
Parents: f54d7b9
Author: gyfora <gy...@gmail.com>
Authored: Mon Jul 21 09:28:53 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Aug 18 16:14:13 2014 +0200
----------------------------------------------------------------------
.../api/collector/DirectedStreamCollector.java | 9 +-
.../api/collector/StreamCollector.java | 24 +--
.../api/invokable/DefaultSinkInvokable.java | 7 +-
.../api/invokable/DefaultTaskInvokable.java | 6 +-
.../streaming/api/invokable/SinkInvokable.java | 13 +-
.../api/invokable/StreamRecordInvokable.java | 20 ++-
.../operator/BatchReduceInvokable.java | 54 ++++--
.../api/invokable/operator/FilterInvokable.java | 17 +-
.../invokable/operator/FlatMapInvokable.java | 12 +-
.../api/invokable/operator/MapInvokable.java | 11 +-
.../AbstractStreamComponent.java | 93 +++++-----
.../streamcomponent/StreamIterationSink.java | 32 ++--
.../streamcomponent/StreamIterationSource.java | 16 +-
.../api/streamcomponent/StreamRecordReader.java | 169 -------------------
.../api/streamcomponent/StreamSink.java | 15 +-
.../api/streamcomponent/StreamSource.java | 9 +-
.../api/streamcomponent/StreamTask.java | 41 ++---
.../UnionStreamRecordReader.java | 91 ----------
.../api/streamrecord/StreamRecord.java | 36 +---
.../streamrecord/StreamRecordSerializer.java | 119 +++++++++++++
.../partitioner/BroadcastPartitioner.java | 4 +-
.../partitioner/FieldsPartitioner.java | 10 +-
.../partitioner/ForwardPartitioner.java | 4 +-
.../partitioner/GlobalPartitioner.java | 4 +-
.../partitioner/ShufflePartitioner.java | 6 +-
.../partitioner/StreamPartitioner.java | 3 +-
.../apache/flink/streaming/api/PrintTest.java | 19 +--
.../api/invokable/operator/BatchReduceTest.java | 5 +-
.../api/streamcomponent/MockRecordWriter.java | 0
29 files changed, 357 insertions(+), 492 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedStreamCollector.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedStreamCollector.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedStreamCollector.java
index b059041..f968b83 100755
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedStreamCollector.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/DirectedStreamCollector.java
@@ -51,7 +51,8 @@ public class DirectedStreamCollector<T extends Tuple> extends StreamCollector<T>
* @param outputSelector
* User defined {@link OutputSelector}
*/
- public DirectedStreamCollector(int channelID, SerializationDelegate<T> serializationDelegate,
+ public DirectedStreamCollector(int channelID,
+ SerializationDelegate<StreamRecord<T>> serializationDelegate,
OutputSelector<T> outputSelector) {
super(channelID, serializationDelegate);
this.outputSelector = outputSelector;
@@ -81,10 +82,12 @@ public class DirectedStreamCollector<T extends Tuple> extends StreamCollector<T>
private void emit(StreamRecord<T> streamRecord) {
Collection<String> outputNames = outputSelector.getOutputs(streamRecord.getTuple());
streamRecord.setId(channelID);
+ serializationDelegate.setInstance(streamRecord);
for (String outputName : outputNames) {
try {
- for (RecordWriter<StreamRecord<T>> output : outputMap.get(outputName)) {
- output.emit(streamRecord);
+ for (RecordWriter<SerializationDelegate<StreamRecord<T>>> output : outputMap
+ .get(outputName)) {
+ output.emit(serializationDelegate);
}
} catch (Exception e) {
if (log.isErrorEnabled()) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/StreamCollector.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/StreamCollector.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/StreamCollector.java
index 47ec8f0..544a695 100755
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/StreamCollector.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/collector/StreamCollector.java
@@ -47,8 +47,9 @@ public class StreamCollector<T extends Tuple> implements Collector<T> {
protected StreamRecord<T> streamRecord;
protected int channelID;
- private List<RecordWriter<StreamRecord<T>>> outputs;
- protected Map<String, List<RecordWriter<StreamRecord<T>>>> outputMap;
+ private List<RecordWriter<SerializationDelegate<StreamRecord<T>>>> outputs;
+ protected Map<String, List<RecordWriter<SerializationDelegate<StreamRecord<T>>>>> outputMap;
+ protected SerializationDelegate<StreamRecord<T>> serializationDelegate;
/**
* Creates a new StreamCollector
@@ -58,13 +59,13 @@ public class StreamCollector<T extends Tuple> implements Collector<T> {
* @param serializationDelegate
* Serialization delegate used for tuple serialization
*/
- public StreamCollector(int channelID, SerializationDelegate<T> serializationDelegate) {
+ public StreamCollector(int channelID, SerializationDelegate<StreamRecord<T>> serializationDelegate) {
+ this.serializationDelegate = serializationDelegate;
this.streamRecord = new StreamRecord<T>();
- this.streamRecord.setSeralizationDelegate(serializationDelegate);
this.channelID = channelID;
- this.outputs = new ArrayList<RecordWriter<StreamRecord<T>>>();
- this.outputMap = new HashMap<String, List<RecordWriter<StreamRecord<T>>>>();
+ this.outputs = new ArrayList<RecordWriter<SerializationDelegate<StreamRecord<T>>>>();
+ this.outputMap = new HashMap<String, List<RecordWriter<SerializationDelegate<StreamRecord<T>>>>>();
}
/**
@@ -75,13 +76,15 @@ public class StreamCollector<T extends Tuple> implements Collector<T> {
* @param outputName
* User defined name of the output.
*/
- public void addOutput(RecordWriter<StreamRecord<T>> output, String outputName) {
+ public void addOutput(RecordWriter<SerializationDelegate<StreamRecord<T>>> output,
+ String outputName) {
outputs.add(output);
if (outputName != null) {
if (outputMap.containsKey(outputName)) {
outputMap.get(outputName).add(output);
} else {
- outputMap.put(outputName,new ArrayList<RecordWriter<StreamRecord<T>>>());
+ outputMap.put(outputName,
+ new ArrayList<RecordWriter<SerializationDelegate<StreamRecord<T>>>>());
outputMap.get(outputName).add(output);
}
@@ -109,9 +112,10 @@ public class StreamCollector<T extends Tuple> implements Collector<T> {
*/
private void emit(StreamRecord<T> streamRecord) {
streamRecord.setId(channelID);
- for (RecordWriter<StreamRecord<T>> output : outputs) {
+ serializationDelegate.setInstance(streamRecord);
+ for (RecordWriter<SerializationDelegate<StreamRecord<T>>> output : outputs) {
try {
- output.emit(streamRecord);
+ output.emit(serializationDelegate);
} catch (Exception e) {
if (LOG.isErrorEnabled()) {
LOG.error(String.format("Emit failed due to: %s",
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultSinkInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultSinkInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultSinkInvokable.java
index 91c6959..09e766b 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultSinkInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultSinkInvokable.java
@@ -19,17 +19,12 @@
package org.apache.flink.streaming.api.invokable;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.util.Collector;
public class DefaultSinkInvokable<T extends Tuple> extends UserSinkInvokable<T> {
private static final long serialVersionUID = 1L;
@Override
- public void invoke(StreamRecord<T> record, Collector<T> collector) throws Exception {
- String value = (String) record.getTuple().getField(0);
- System.out.println(value);
+ public void invoke() throws Exception {
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultTaskInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultTaskInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultTaskInvokable.java
index abf392c..c86ffcb 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultTaskInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/DefaultTaskInvokable.java
@@ -19,17 +19,13 @@
package org.apache.flink.streaming.api.invokable;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.util.Collector;
public class DefaultTaskInvokable<T extends Tuple> extends UserTaskInvokable<T, T> {
private static final long serialVersionUID = 1L;
@Override
- public void invoke(StreamRecord<T> record, Collector<T> collector) throws Exception {
- collector.collect(record.getTuple());
+ public void invoke() throws Exception {
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
index cb16307..c6e4e36 100755
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/SinkInvokable.java
@@ -19,10 +19,8 @@
package org.apache.flink.streaming.api.invokable;
-import org.apache.flink.streaming.api.function.sink.SinkFunction;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.util.Collector;
+import org.apache.flink.streaming.api.function.sink.SinkFunction;
public class SinkInvokable<IN extends Tuple> extends UserSinkInvokable<IN> {
private static final long serialVersionUID = 1L;
@@ -32,10 +30,11 @@ public class SinkInvokable<IN extends Tuple> extends UserSinkInvokable<IN> {
public SinkInvokable(SinkFunction<IN> sinkFunction) {
this.sinkFunction = sinkFunction;
}
-
+
@Override
- public void invoke(StreamRecord<IN> record, Collector<IN> collector) throws Exception {
- IN tuple = record.getTuple();
- sinkFunction.invoke(tuple);
+ public void invoke() throws Exception {
+ while ((reuse = recordIterator.next(reuse)) != null) {
+ sinkFunction.invoke((IN) reuse.getTuple());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamRecordInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamRecordInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamRecordInvokable.java
index 229d409..98eb679 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamRecordInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/StreamRecordInvokable.java
@@ -19,16 +19,28 @@
package org.apache.flink.streaming.api.invokable;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.streaming.api.streamrecord.StreamRecord;
+import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
public abstract class StreamRecordInvokable<IN extends Tuple, OUT extends Tuple> extends
StreamComponentInvokable {
private static final long serialVersionUID = 1L;
- public abstract void invoke(StreamRecord<IN> record, Collector<OUT> collector)
- throws Exception;
+ protected Collector<OUT> collector;
+ protected MutableObjectIterator<StreamRecord<IN>> recordIterator;
+ protected StreamRecord<IN> reuse;
+
+ public void initialize(Collector<OUT> collector,
+ MutableObjectIterator<StreamRecord<IN>> recordIterator,
+ StreamRecordSerializer<IN> serializer) {
+ this.collector = collector;
+ this.recordIterator = recordIterator;
+ this.reuse = serializer.createInstance();
+ }
+
+ public abstract void invoke() throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceInvokable.java
index 0b009e9..8e35af2 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceInvokable.java
@@ -19,19 +19,16 @@
package org.apache.flink.streaming.api.invokable.operator;
-import java.util.ArrayList;
-import java.util.List;
+import java.io.IOException;
+import java.util.Iterator;
import org.apache.flink.api.java.functions.GroupReduceFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-import org.apache.flink.util.Collector;
public class BatchReduceInvokable<IN extends Tuple, OUT extends Tuple> extends
UserTaskInvokable<IN, OUT> {
private static final long serialVersionUID = 1L;
- private List<IN> tupleBatch;
private int counter;
private int batchSize;
@@ -39,20 +36,53 @@ public class BatchReduceInvokable<IN extends Tuple, OUT extends Tuple> extends
public BatchReduceInvokable(GroupReduceFunction<IN, OUT> reduceFunction, int batchSize) {
this.reducer = reduceFunction;
- this.tupleBatch = new ArrayList<IN>();
this.counter = 0;
this.batchSize = batchSize;
}
@Override
- public void invoke(StreamRecord<IN> record, Collector<OUT> collector) throws Exception {
+ public void invoke() throws Exception {
+ MyIterator it = new MyIterator();
+ do {
+ reducer.reduce(it, collector);
+ it.reset();
+ } while (reuse != null);
+ }
+
+ public class MyIterator implements Iterator<IN> {
+
+ public MyIterator() {
+ reset();
+ }
+
+ @Override
+ public boolean hasNext() {
+
+ if (counter >= batchSize) {
+ return false;
+ } else {
+ try {
+ reuse = recordIterator.next(reuse);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ return (reuse != null);
+ }
+ }
+
+ @Override
+ public IN next() {
+ counter++;
+ return reuse.getTuple();
+ }
- tupleBatch.add(record.getTuple());
- counter++;
- if (counter >= batchSize) {
+ public void reset() {
counter = 0;
- reducer.reduce(tupleBatch.iterator(), collector);
- tupleBatch.clear();
+ }
+
+ @Override
+ public void remove() {
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
index 0d135b5..1dba6b8 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FilterInvokable.java
@@ -19,14 +19,11 @@
package org.apache.flink.streaming.api.invokable.operator;
-import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.api.java.functions.FilterFunction;
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.util.Collector;
+import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
-public class FilterInvokable<IN extends Tuple> extends
- UserTaskInvokable<IN, IN> {
+public class FilterInvokable<IN extends Tuple> extends UserTaskInvokable<IN, IN> {
private static final long serialVersionUID = 1L;
@@ -37,11 +34,11 @@ public class FilterInvokable<IN extends Tuple> extends
}
@Override
- public void invoke(StreamRecord<IN> record, Collector<IN> collector)
- throws Exception {
- IN tuple = record.getTuple();
- if (filterFunction.filter(tuple)) {
- collector.collect(tuple);
+ public void invoke() throws Exception {
+ while ((reuse = recordIterator.next(reuse)) != null) {
+ if (filterFunction.filter(reuse.getTuple())) {
+ collector.collect(reuse.getTuple());
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
index 5696b40..36ab1bf 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/FlatMapInvokable.java
@@ -19,11 +19,9 @@
package org.apache.flink.streaming.api.invokable.operator;
-import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.api.java.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.util.Collector;
+import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
public class FlatMapInvokable<IN extends Tuple, OUT extends Tuple> extends
UserTaskInvokable<IN, OUT> {
@@ -35,9 +33,9 @@ public class FlatMapInvokable<IN extends Tuple, OUT extends Tuple> extends
this.flatMapper = flatMapper;
}
- @Override
- public void invoke(StreamRecord<IN> record, Collector<OUT> collector) throws Exception {
- IN tuple = record.getTuple();
- flatMapper.flatMap(tuple, collector);
+ public void invoke() throws Exception {
+ while ((reuse = recordIterator.next(reuse)) != null) {
+ flatMapper.flatMap(reuse.getTuple(), collector);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
index 378c898..59b9f8f 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/MapInvokable.java
@@ -19,11 +19,9 @@
package org.apache.flink.streaming.api.invokable.operator;
-import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
import org.apache.flink.api.java.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.util.Collector;
+import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
public class MapInvokable<IN extends Tuple, OUT extends Tuple> extends UserTaskInvokable<IN, OUT> {
private static final long serialVersionUID = 1L;
@@ -35,8 +33,9 @@ public class MapInvokable<IN extends Tuple, OUT extends Tuple> extends UserTaskI
}
@Override
- public void invoke(StreamRecord<IN> record, Collector<OUT> collector) throws Exception {
- IN tuple = record.getTuple();
- collector.collect(mapper.map(tuple));
+ public void invoke() throws Exception {
+ while ((reuse = recordIterator.next(reuse)) != null) {
+ collector.collect(mapper.map(reuse.getTuple()));
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/AbstractStreamComponent.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/AbstractStreamComponent.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/AbstractStreamComponent.java
index 1132276..207d269 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/AbstractStreamComponent.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/AbstractStreamComponent.java
@@ -28,6 +28,7 @@ import org.apache.commons.lang.SerializationUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.flink.api.common.functions.AbstractFunction;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.FilterFunction;
import org.apache.flink.api.java.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.GroupReduceFunction;
@@ -35,12 +36,14 @@ import org.apache.flink.api.java.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.io.network.api.AbstractRecordReader;
+import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.io.network.api.MutableReader;
import org.apache.flink.runtime.io.network.api.MutableRecordReader;
+import org.apache.flink.runtime.io.network.api.MutableUnionRecordReader;
import org.apache.flink.runtime.io.network.api.RecordWriter;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.operators.util.ReaderIterator;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.collector.DirectedStreamCollector;
@@ -48,24 +51,25 @@ import org.apache.flink.streaming.api.collector.OutputSelector;
import org.apache.flink.streaming.api.collector.StreamCollector;
import org.apache.flink.streaming.api.function.sink.SinkFunction;
import org.apache.flink.streaming.api.invokable.StreamComponentInvokable;
-import org.apache.flink.streaming.api.invokable.StreamRecordInvokable;
import org.apache.flink.streaming.api.invokable.UserSourceInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
+import org.apache.flink.streaming.api.streamrecord.StreamRecordSerializer;
import org.apache.flink.streaming.partitioner.ShufflePartitioner;
import org.apache.flink.streaming.partitioner.StreamPartitioner;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tuple> extends
AbstractInvokable {
private static final Log LOG = LogFactory.getLog(AbstractStreamComponent.class);
protected TupleTypeInfo<IN> inTupleTypeInfo = null;
- protected TupleSerializer<IN> inTupleSerializer = null;
- protected DeserializationDelegate<IN> inDeserializationDelegate = null;
+ protected StreamRecordSerializer<IN> inTupleSerializer = null;
+ protected DeserializationDelegate<StreamRecord<IN>> inDeserializationDelegate = null;
protected TupleTypeInfo<OUT> outTupleTypeInfo = null;
- protected TupleSerializer<OUT> outTupleSerializer = null;
- protected SerializationDelegate<OUT> outSerializationDelegate = null;
+ protected StreamRecordSerializer<OUT> outTupleSerializer = null;
+ protected SerializationDelegate<StreamRecord<OUT>> outSerializationDelegate = null;
protected Configuration configuration;
protected StreamCollector<OUT> collector;
@@ -126,16 +130,18 @@ public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tupl
} else if (operatorName.equals("elements")) {
outTupleTypeInfo = new TupleTypeInfo<OUT>(TypeExtractor.getForObject(function));
- outTupleSerializer = outTupleTypeInfo.createSerializer();
- outSerializationDelegate = new SerializationDelegate<OUT>(outTupleSerializer);
+ outTupleSerializer = new StreamRecordSerializer<OUT>(outTupleTypeInfo.createSerializer());
+ outSerializationDelegate = new SerializationDelegate<StreamRecord<OUT>>(
+ outTupleSerializer);
} else {
throw new Exception("Wrong operator name!");
}
} catch (Exception e) {
throw new StreamComponentException(e);
-// throw new StreamComponentException("Nonsupported object (named " + operatorName
-// + ") passed as operator");
+ // throw new StreamComponentException("Nonsupported object (named "
+ // + operatorName
+ // + ") passed as operator");
}
}
@@ -149,8 +155,8 @@ public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tupl
inTupleTypeInfo = (TupleTypeInfo) TypeExtractor.createTypeInfo(clazz, function.getClass(),
0, null, null);
- inTupleSerializer = inTupleTypeInfo.createSerializer();
- inDeserializationDelegate = new DeserializationDelegate<IN>(inTupleSerializer);
+ inTupleSerializer = new StreamRecordSerializer(inTupleTypeInfo.createSerializer());
+ inDeserializationDelegate = new DeserializationDelegate<StreamRecord<IN>>(inTupleSerializer);
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@@ -158,8 +164,8 @@ public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tupl
outTupleTypeInfo = (TupleTypeInfo) TypeExtractor.createTypeInfo(clazz, function.getClass(),
typeParameter, null, null);
- outTupleSerializer = outTupleTypeInfo.createSerializer();
- outSerializationDelegate = new SerializationDelegate<OUT>(outTupleSerializer);
+ outTupleSerializer = new StreamRecordSerializer(outTupleTypeInfo.createSerializer());
+ outSerializationDelegate = new SerializationDelegate<StreamRecord<OUT>>(outTupleSerializer);
}
@SuppressWarnings("unchecked")
@@ -167,32 +173,42 @@ public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tupl
if (outSerializationDelegate != null) {
inTupleTypeInfo = (TupleTypeInfo<IN>) outTupleTypeInfo;
- inTupleSerializer = inTupleTypeInfo.createSerializer();
- inDeserializationDelegate = new DeserializationDelegate<IN>(inTupleSerializer);
+ inTupleSerializer = new StreamRecordSerializer<IN>(inTupleTypeInfo.createSerializer());
+ inDeserializationDelegate = new DeserializationDelegate<StreamRecord<IN>>(inTupleSerializer);
}
}
@SuppressWarnings("unchecked")
- protected AbstractRecordReader getConfigInputs() throws StreamComponentException {
+ protected MutableReader<IOReadableWritable> getConfigInputs() throws StreamComponentException {
int numberOfInputs = configuration.getInteger("numberOfInputs", 0);
if (numberOfInputs < 2) {
- return new StreamRecordReader<IN>(this, StreamRecord.class, inDeserializationDelegate,
- inTupleSerializer);
+ return new MutableRecordReader<IOReadableWritable>(this);
} else {
- MutableRecordReader<StreamRecord<IN>>[] recordReaders = (MutableRecordReader<StreamRecord<IN>>[]) new MutableRecordReader<?>[numberOfInputs];
+ MutableRecordReader<IOReadableWritable>[] recordReaders = (MutableRecordReader<IOReadableWritable>[]) new MutableRecordReader<?>[numberOfInputs];
for (int i = 0; i < numberOfInputs; i++) {
- recordReaders[i] = new MutableRecordReader<StreamRecord<IN>>(this);
+ recordReaders[i] = new MutableRecordReader<IOReadableWritable>(this);
}
- return new UnionStreamRecordReader<IN>(recordReaders, StreamRecord.class,
- inDeserializationDelegate, inTupleSerializer);
+ return new MutableUnionRecordReader<IOReadableWritable>(recordReaders);
}
}
- protected void setConfigOutputs(List<RecordWriter<StreamRecord<OUT>>> outputs) {
+ protected MutableObjectIterator<StreamRecord<IN>> createInputIterator(MutableReader<?> inputReader,
+ TypeSerializer<?> serializer) {
+
+ // generic data type serialization
+ @SuppressWarnings("unchecked")
+ MutableReader<DeserializationDelegate<?>> reader = (MutableReader<DeserializationDelegate<?>>) inputReader;
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ final MutableObjectIterator<StreamRecord<IN>> iter = new ReaderIterator(reader, serializer);
+ return iter;
+
+ }
+
+ protected void setConfigOutputs(List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> outputs) {
int numberOfOutputs = configuration.getInteger("numberOfOutputs", 0);
@@ -201,7 +217,8 @@ public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tupl
}
}
- private void setPartitioner(int outputNumber, List<RecordWriter<StreamRecord<OUT>>> outputs) {
+ private void setPartitioner(int outputNumber,
+ List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> outputs) {
byte[] serializedPartitioner = configuration.getBytes("partitionerObject_" + outputNumber,
SerializationUtils.serialize((new ShufflePartitioner<OUT>())));
@@ -210,8 +227,8 @@ public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tupl
try {
outputPartitioner = deserializeObject(serializedPartitioner);
- RecordWriter<StreamRecord<OUT>> output = new RecordWriter<StreamRecord<OUT>>(this,
- outputPartitioner);
+ RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output = new RecordWriter<SerializationDelegate<StreamRecord<OUT>>>(
+ this, outputPartitioner);
outputs.add(output);
String outputName = configuration.getString("outputName_" + outputNumber, null);
@@ -230,26 +247,6 @@ public abstract class AbstractStreamComponent<IN extends Tuple, OUT extends Tupl
}
}
- @SuppressWarnings("unchecked")
- protected void invokeRecords(StreamRecordInvokable<IN, OUT> userFunction,
- AbstractRecordReader inputs) throws Exception {
- if (inputs instanceof UnionStreamRecordReader) {
- UnionStreamRecordReader<IN> recordReader = (UnionStreamRecordReader<IN>) inputs;
- while (recordReader.hasNext()) {
- StreamRecord<IN> record = recordReader.next();
- userFunction.invoke(record, collector);
- }
-
- } else if (inputs instanceof StreamRecordReader) {
- StreamRecordReader<IN> recordReader = (StreamRecordReader<IN>) inputs;
-
- while (recordReader.hasNext()) {
- StreamRecord<IN> record = recordReader.next();
- userFunction.invoke(record, collector);
- }
- }
- }
-
/**
* Reads and creates a StreamComponent from the config.
*
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSink.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSink.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSink.java
index c2e214d..b9d44c3 100755
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSink.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSink.java
@@ -25,15 +25,18 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.runtime.io.network.api.AbstractRecordReader;
+import org.apache.flink.runtime.io.network.api.MutableReader;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
import org.apache.flink.util.StringUtils;
public class StreamIterationSink<IN extends Tuple> extends AbstractStreamComponent<IN, IN> {
private static final Log LOG = LogFactory.getLog(StreamIterationSink.class);
- private AbstractRecordReader inputs;
+ @SuppressWarnings("rawtypes")
+ private MutableReader inputs;
+ MutableObjectIterator<StreamRecord<IN>> inputIter;
private String iterationId;
@SuppressWarnings("rawtypes")
private BlockingQueue<StreamRecord> dataChannel;
@@ -49,6 +52,7 @@ public class StreamIterationSink<IN extends Tuple> extends AbstractStreamCompone
setSerializers();
setSinkSerializer();
inputs = getConfigInputs();
+ inputIter = createInputIterator(inputs, inTupleSerializer);
iterationId = configuration.getString("iteration-id", "iteration-0");
dataChannel = BlockingQueueBroker.instance().get(iterationId);
} catch (Exception e) {
@@ -63,29 +67,19 @@ public class StreamIterationSink<IN extends Tuple> extends AbstractStreamCompone
LOG.debug("SINK " + name + " invoked");
}
- forwardRecords(inputs);
+ forwardRecords();
if (LOG.isDebugEnabled()) {
LOG.debug("SINK " + name + " invoke finished");
}
}
- @SuppressWarnings("unchecked")
- protected void forwardRecords(AbstractRecordReader inputs) throws Exception {
- if (inputs instanceof UnionStreamRecordReader) {
- UnionStreamRecordReader<IN> recordReader = (UnionStreamRecordReader<IN>) inputs;
- while (recordReader.hasNext()) {
- StreamRecord<IN> record = recordReader.next();
- pushToQueue(record);
- }
-
- } else if (inputs instanceof StreamRecordReader) {
- StreamRecordReader<IN> recordReader = (StreamRecordReader<IN>) inputs;
-
- while (recordReader.hasNext()) {
- StreamRecord<IN> record = recordReader.next();
- pushToQueue(record);
- }
+ protected void forwardRecords() throws Exception {
+ StreamRecord<IN> reuse = inTupleSerializer.createInstance().setId(0);
+ while ((reuse = inputIter.next(reuse)) != null) {
+ pushToQueue(reuse);
+ // TODO: Fix object reuse for iteration
+ reuse = inTupleSerializer.createInstance().setId(0);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSource.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSource.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSource.java
index 7f4da1f..9f58842 100755
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSource.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamIterationSource.java
@@ -29,13 +29,14 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.runtime.io.network.api.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
public class StreamIterationSource<OUT extends Tuple> extends AbstractStreamComponent<Tuple, OUT> {
private static final Log LOG = LogFactory.getLog(StreamIterationSource.class);
- private List<RecordWriter<StreamRecord<OUT>>> outputs;
+ private List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> outputs;
private static int numSources;
private int[] numberOfOutputChannels;
private String iterationId;
@@ -45,7 +46,7 @@ public class StreamIterationSource<OUT extends Tuple> extends AbstractStreamComp
@SuppressWarnings("rawtypes")
public StreamIterationSource() {
- outputs = new LinkedList<RecordWriter<StreamRecord<OUT>>>();
+ outputs = new LinkedList<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>>();
numSources = newComponent();
instanceID = numSources;
dataChannel = new ArrayBlockingQueue<StreamRecord>(1);
@@ -81,7 +82,7 @@ public class StreamIterationSource<OUT extends Tuple> extends AbstractStreamComp
LOG.debug("SOURCE " + name + " invoked with instance id " + instanceID);
}
- for (RecordWriter<StreamRecord<OUT>> output : outputs) {
+ for (RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output : outputs) {
output.initializeSerializers();
}
@@ -91,14 +92,13 @@ public class StreamIterationSource<OUT extends Tuple> extends AbstractStreamComp
if (nextRecord == null) {
break;
}
- nextRecord.setSeralizationDelegate(this.outSerializationDelegate);
- for (RecordWriter<StreamRecord<OUT>> output : outputs) {
- output.emit(nextRecord);
- // output.flush();
+ for (RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output : outputs) {
+ outSerializationDelegate.setInstance(nextRecord);
+ output.emit(outSerializationDelegate);
}
}
- for (RecordWriter<StreamRecord<OUT>> output : outputs) {
+ for (RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output : outputs) {
output.flush();
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamRecordReader.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamRecordReader.java
deleted file mode 100755
index d1cc1e0..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamRecordReader.java
+++ /dev/null
@@ -1,169 +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.streamcomponent;
-
-import java.io.IOException;
-
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.plugable.DeserializationDelegate;
-import org.apache.flink.runtime.io.network.api.AbstractSingleGateRecordReader;
-import org.apache.flink.runtime.io.network.api.Reader;
-import org.apache.flink.runtime.io.network.gates.InputChannelResult;
-
-/**
- * A record writer connects an input gate to an application. It allows the
- * application query for incoming records and read them from input gate.
- *
- */
-public class StreamRecordReader<T extends Tuple> extends AbstractSingleGateRecordReader<StreamRecord<T>> implements
- Reader<StreamRecord<T>> {
-
- @SuppressWarnings("rawtypes")
- private final Class<? extends StreamRecord> recordType;
- private DeserializationDelegate<T> deserializationDelegate;
- private TupleSerializer<T> tupleSerializer;
- /**
- * Stores the last read record.
- */
- private StreamRecord<T> lookahead;
-
- /**
- * Stores if more no more records will be received from the assigned input
- * gate.
- */
- private boolean noMoreRecordsWillFollow;
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Constructs a new record reader and registers a new input gate with the
- * application's environment.
- *
- * @param taskBase
- * The application that instantiated the record reader.
- * @param recordType
- * The class of records that can be read from the record reader.
- * @param deserializationDelegate
- * deserializationDelegate
- * @param tupleSerializer
- * tupleSerializer
- */
- @SuppressWarnings("rawtypes")
- public StreamRecordReader(AbstractInvokable taskBase, Class<? extends StreamRecord> recordType,
- DeserializationDelegate<T> deserializationDelegate,
- TupleSerializer<T> tupleSerializer) {
- super(taskBase);
- this.recordType = recordType;
- this.deserializationDelegate = deserializationDelegate;
- this.tupleSerializer = tupleSerializer;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Checks if at least one more record can be read from the associated input
- * gate. This method may block until the associated input gate is able to
- * read the record from one of its input channels.
- *
- * @return <code>true</code>it at least one more record can be read from the
- * associated input gate, otherwise <code>false</code>
- */
- @Override
- public boolean hasNext() throws IOException, InterruptedException {
- if (this.lookahead != null) {
- return true;
- } else {
- if (this.noMoreRecordsWillFollow) {
- return false;
- }
-
- StreamRecord<T> record = instantiateRecordType();
- record.setDeseralizationDelegate(deserializationDelegate, tupleSerializer);
-
- while (true) {
- InputChannelResult result = this.inputGate.readRecord(record);
- switch (result) {
- case INTERMEDIATE_RECORD_FROM_BUFFER:
- case LAST_RECORD_FROM_BUFFER:
- this.lookahead = record;
- return true;
-
- case END_OF_SUPERSTEP:
- if (incrementEndOfSuperstepEventAndCheck()) {
- return false;
- } else {
- break; // fall through and wait for next record/event
- }
-
- case TASK_EVENT:
- handleEvent(this.inputGate.getCurrentEvent());
- break;
-
- case END_OF_STREAM:
- this.noMoreRecordsWillFollow = true;
- return false;
-
- default:
- ; // fall through the loop
- }
- }
- }
- }
-
- /**
- * Reads the current record from the associated input gate.
- *
- * @return the current record from the associated input gate.
- * @throws IOException
- * thrown if any error occurs while reading the record from the
- * input gate
- */
- @Override
- public StreamRecord<T> next() throws IOException, InterruptedException {
- if (hasNext()) {
- StreamRecord<T> tmp = this.lookahead;
- this.lookahead = null;
- return tmp;
- } else {
- return null;
- }
- }
-
- @Override
- public boolean isInputClosed() {
- return this.noMoreRecordsWillFollow;
- }
-
- @SuppressWarnings("unchecked")
- private StreamRecord<T> instantiateRecordType() {
- try {
- return this.recordType.newInstance();
- } catch (InstantiationException e) {
- throw new RuntimeException("Cannot instantiate class '" + this.recordType.getName()
- + "'.", e);
- } catch (IllegalAccessException e) {
- throw new RuntimeException("Cannot instantiate class '" + this.recordType.getName()
- + "'.", e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSink.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSink.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSink.java
index f27282a..508a10e9 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSink.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSink.java
@@ -21,18 +21,21 @@ package org.apache.flink.streaming.api.streamcomponent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.runtime.io.network.api.MutableReader;
import org.apache.flink.streaming.api.invokable.DefaultSinkInvokable;
import org.apache.flink.streaming.api.invokable.StreamRecordInvokable;
import org.apache.flink.streaming.api.invokable.UserSinkInvokable;
-
-import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.runtime.io.network.api.AbstractRecordReader;
+import org.apache.flink.streaming.api.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
public class StreamSink<IN extends Tuple> extends AbstractStreamComponent<IN, IN> {
private static final Log LOG = LogFactory.getLog(StreamSink.class);
- private AbstractRecordReader inputs;
+ @SuppressWarnings("rawtypes")
+ private MutableReader inputs;
+ private MutableObjectIterator<StreamRecord<IN>> inputIter;
private StreamRecordInvokable<IN, IN> userFunction;
public StreamSink() {
@@ -47,6 +50,7 @@ public class StreamSink<IN extends Tuple> extends AbstractStreamComponent<IN, IN
setSerializers();
setSinkSerializer();
inputs = getConfigInputs();
+ inputIter = createInputIterator(inputs, inTupleSerializer);
} catch (Exception e) {
throw new StreamComponentException("Cannot register inputs for "
+ getClass().getSimpleName(), e);
@@ -61,6 +65,7 @@ public class StreamSink<IN extends Tuple> extends AbstractStreamComponent<IN, IN
Class<? extends UserSinkInvokable> userFunctionClass = configuration.getClass(
"userfunction", DefaultSinkInvokable.class, UserSinkInvokable.class);
userFunction = (UserSinkInvokable<IN>) getInvokable(userFunctionClass);
+ userFunction.initialize(collector, inputIter, inTupleSerializer);
}
@Override
@@ -69,7 +74,7 @@ public class StreamSink<IN extends Tuple> extends AbstractStreamComponent<IN, IN
LOG.debug("SINK " + name + " invoked");
}
- invokeRecords(userFunction, inputs);
+ userFunction.invoke();
if (LOG.isDebugEnabled()) {
LOG.debug("SINK " + name + " invoke finished");
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSource.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSource.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSource.java
index 3936ce0..4ac7100 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSource.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamSource.java
@@ -26,6 +26,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.runtime.io.network.api.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.invokable.DefaultSourceInvokable;
import org.apache.flink.streaming.api.invokable.UserSourceInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
@@ -34,14 +35,14 @@ public class StreamSource<OUT extends Tuple> extends AbstractStreamComponent<Tup
private static final Log LOG = LogFactory.getLog(StreamSource.class);
- private List<RecordWriter<StreamRecord<OUT>>> outputs;
+ private List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> outputs;
private UserSourceInvokable<OUT> userFunction;
private static int numSources;
private int[] numberOfOutputChannels;
public StreamSource() {
- outputs = new LinkedList<RecordWriter<StreamRecord<OUT>>>();
+ outputs = new LinkedList<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>>();
userFunction = null;
numSources = newComponent();
instanceID = numSources;
@@ -83,7 +84,7 @@ public class StreamSource<OUT extends Tuple> extends AbstractStreamComponent<Tup
LOG.debug("SOURCE " + name + " invoked with instance id " + instanceID);
}
- for (RecordWriter<StreamRecord<OUT>> output : outputs) {
+ for (RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output : outputs) {
output.initializeSerializers();
}
@@ -93,7 +94,7 @@ public class StreamSource<OUT extends Tuple> extends AbstractStreamComponent<Tup
LOG.debug("SOURCE " + name + " invoke finished with instance id " + instanceID);
}
- for (RecordWriter<StreamRecord<OUT>> output : outputs) {
+ for (RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output : outputs) {
output.flush();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamTask.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamTask.java
index 98d6d97..44f93b3 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamTask.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/StreamTask.java
@@ -25,26 +25,31 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.runtime.io.network.api.AbstractRecordReader;
+import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.io.network.api.MutableReader;
import org.apache.flink.runtime.io.network.api.RecordWriter;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.invokable.DefaultTaskInvokable;
import org.apache.flink.streaming.api.invokable.StreamRecordInvokable;
import org.apache.flink.streaming.api.invokable.UserTaskInvokable;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
-public class StreamTask<IN extends Tuple, OUT extends Tuple> extends AbstractStreamComponent<IN, OUT> {
+public class StreamTask<IN extends Tuple, OUT extends Tuple> extends
+ AbstractStreamComponent<IN, OUT> {
private static final Log LOG = LogFactory.getLog(StreamTask.class);
- private AbstractRecordReader inputs;
- private List<RecordWriter<StreamRecord<OUT>>> outputs;
+ private MutableReader<IOReadableWritable> inputs;
+ MutableObjectIterator<StreamRecord<IN>> inputIter;
+ private List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> outputs;
private StreamRecordInvokable<IN, OUT> userFunction;
private int[] numberOfOutputChannels;
private static int numTasks;
public StreamTask() {
-
- outputs = new LinkedList<RecordWriter<StreamRecord<OUT>>>();
+
+ outputs = new LinkedList<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>>();
userFunction = null;
numTasks = newComponent();
instanceID = numTasks;
@@ -54,15 +59,12 @@ public class StreamTask<IN extends Tuple, OUT extends Tuple> extends AbstractStr
public void registerInputOutput() {
initialize();
-// try {
- setSerializers();
- setCollector();
- inputs = getConfigInputs();
- setConfigOutputs(outputs);
-// } catch (StreamComponentException e) {
-// throw new StreamComponentException("Cannot register inputs/outputs for "
-// + getClass().getSimpleName(), e);
-// }
+ setSerializers();
+ setCollector();
+ inputs = getConfigInputs();
+ setConfigOutputs(outputs);
+
+ inputIter = createInputIterator(inputs, inTupleSerializer);
numberOfOutputChannels = new int[outputs.size()];
for (int i = 0; i < numberOfOutputChannels.length; i++) {
@@ -79,6 +81,7 @@ public class StreamTask<IN extends Tuple, OUT extends Tuple> extends AbstractStr
Class<? extends UserTaskInvokable> userFunctionClass = configuration.getClass(
"userfunction", DefaultTaskInvokable.class, UserTaskInvokable.class);
userFunction = (UserTaskInvokable<IN, OUT>) getInvokable(userFunctionClass);
+ userFunction.initialize(collector, inputIter, inTupleSerializer);
}
@Override
@@ -87,17 +90,17 @@ public class StreamTask<IN extends Tuple, OUT extends Tuple> extends AbstractStr
LOG.debug("TASK " + name + " invoked with instance id " + instanceID);
}
- for (RecordWriter<StreamRecord<OUT>> output : outputs) {
+ for (RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output : outputs) {
output.initializeSerializers();
}
- invokeRecords(userFunction, inputs);
+ userFunction.invoke();
if (LOG.isDebugEnabled()) {
LOG.debug("TASK " + name + " invoke finished with instance id " + instanceID);
}
-
- for (RecordWriter<StreamRecord<OUT>> output : outputs){
+
+ for (RecordWriter<SerializationDelegate<StreamRecord<OUT>>> output : outputs) {
output.flush();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/UnionStreamRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/UnionStreamRecordReader.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/UnionStreamRecordReader.java
deleted file mode 100755
index 90fba85..0000000
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamcomponent/UnionStreamRecordReader.java
+++ /dev/null
@@ -1,91 +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.streamcomponent;
-
-import java.io.IOException;
-
-import org.apache.flink.streaming.api.streamrecord.StreamRecord;
-import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
-import org.apache.flink.runtime.plugable.DeserializationDelegate;
-import org.apache.flink.runtime.io.network.api.AbstractUnionRecordReader;
-import org.apache.flink.runtime.io.network.api.MutableRecordReader;
-import org.apache.flink.runtime.io.network.api.Reader;
-
-public final class UnionStreamRecordReader<T extends Tuple> extends AbstractUnionRecordReader<StreamRecord<T>>
- implements Reader<StreamRecord<T>> {
-
- @SuppressWarnings("rawtypes")
- private final Class<? extends StreamRecord> recordType;
-
- private StreamRecord<T> lookahead;
- private DeserializationDelegate<T> deserializationDelegate;
- private TupleSerializer<T> tupleSerializer;
-
- @SuppressWarnings("rawtypes")
- public UnionStreamRecordReader(MutableRecordReader<StreamRecord<T>>[] recordReaders, Class<? extends StreamRecord> class1,
- DeserializationDelegate<T> deserializationDelegate,
- TupleSerializer<T> tupleSerializer) {
- super(recordReaders);
- this.recordType = class1;
- this.deserializationDelegate = deserializationDelegate;
- this.tupleSerializer = tupleSerializer;
- }
-
- @Override
- public boolean hasNext() throws IOException, InterruptedException {
- if (this.lookahead != null) {
- return true;
- } else {
- StreamRecord<T> record = instantiateRecordType();
- record.setDeseralizationDelegate(deserializationDelegate, tupleSerializer);
- if (getNextRecord(record)) {
- this.lookahead = record;
- return true;
- } else {
- return false;
- }
- }
- }
-
- @Override
- public StreamRecord<T> next() throws IOException, InterruptedException {
- if (hasNext()) {
- StreamRecord<T> tmp = this.lookahead;
- this.lookahead = null;
- return tmp;
- } else {
- return null;
- }
- }
-
- @SuppressWarnings("unchecked")
- private StreamRecord<T> instantiateRecordType() {
- try {
- return this.recordType.newInstance();
- } catch (InstantiationException e) {
- throw new RuntimeException("Cannot instantiate class '" + this.recordType.getName()
- + "'.", e);
- } catch (IllegalAccessException e) {
- throw new RuntimeException("Cannot instantiate class '" + this.recordType.getName()
- + "'.", e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecord.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecord.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecord.java
index a8a6012..5021013 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecord.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecord.java
@@ -19,45 +19,27 @@
package org.apache.flink.streaming.api.streamrecord;
-import java.io.IOException;
import java.io.Serializable;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
-import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.runtime.plugable.DeserializationDelegate;
-import org.apache.flink.runtime.plugable.SerializationDelegate;
/**
* Object for wrapping a tuple with ID used for sending records between
* streaming task in Apache Flink stream processing.
*/
-public class StreamRecord<T extends Tuple> implements IOReadableWritable, Serializable {
+public class StreamRecord<T extends Tuple> implements Serializable {
private static final long serialVersionUID = 1L;
protected UID uid;
private T tuple;
- protected SerializationDelegate<T> serializationDelegate;
- protected DeserializationDelegate<T> deserializationDelegate;
protected TupleSerializer<T> tupleSerializer;
public StreamRecord() {
}
- public void setSeralizationDelegate(SerializationDelegate<T> serializationDelegate) {
- this.serializationDelegate = serializationDelegate;
- }
-
- public void setDeseralizationDelegate(DeserializationDelegate<T> deserializationDelegate,
- TupleSerializer<T> tupleSerializer) {
- this.deserializationDelegate = deserializationDelegate;
- this.tupleSerializer = tupleSerializer;
- }
-
/**
* @return The ID of the object
*/
@@ -98,22 +80,6 @@ public class StreamRecord<T extends Tuple> implements IOReadableWritable, Serial
}
@Override
- public void read(DataInputView in) throws IOException {
- uid = new UID();
- uid.read(in);
- deserializationDelegate.setInstance(tupleSerializer.createInstance());
- deserializationDelegate.read(in);
- tuple = deserializationDelegate.getInstance();
- }
-
- @Override
- public void write(DataOutputView out) throws IOException {
- uid.write(out);
- serializationDelegate.setInstance(tuple);
- serializationDelegate.write(out);
- }
-
- @Override
public String toString() {
return tuple.toString();
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecordSerializer.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecordSerializer.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecordSerializer.java
new file mode 100755
index 0000000..b367333
--- /dev/null
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamrecord/StreamRecordSerializer.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,
+ * 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.streamrecord;
+
+import java.io.IOException;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+public final class StreamRecordSerializer<T extends Tuple> extends TypeSerializer<StreamRecord<T>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final TupleSerializer<T> tupleSerializer;
+
+ public StreamRecordSerializer(TupleSerializer<T> tupleSerializer) {
+ this.tupleSerializer = tupleSerializer;
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public boolean isStateful() {
+ return false;
+ }
+
+ @Override
+ public StreamRecord<T> createInstance() {
+ try {
+ @SuppressWarnings("unchecked")
+ StreamRecord<T> t = StreamRecord.class.newInstance().setId(0);
+ t.setTuple(tupleSerializer.createInstance());
+
+ return t;
+ } catch (Exception e) {
+ throw new RuntimeException("Cannot instantiate StreamRecord.", e);
+ }
+ }
+
+ @Override
+ public StreamRecord<T> copy(StreamRecord<T> from, StreamRecord<T> reuse) {
+
+ return null;
+ // for (int i = 0; i < arity; i++) {
+ // Object copy = fieldSerializers[i].copy(from.getField(i),
+ // reuse.getField(i));
+ // reuse.setField(copy, i);
+ // }
+ //
+ // return reuse;
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(StreamRecord<T> value, DataOutputView target) throws IOException {
+ value.getId().write(target);
+ tupleSerializer.serialize(value.getTuple(), target);
+ }
+
+ @Override
+ public StreamRecord<T> deserialize(StreamRecord<T> reuse, DataInputView source)
+ throws IOException {
+ reuse.getId().read(source);
+ tupleSerializer.deserialize(reuse.getTuple(), source);
+ return reuse;
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+
+ }
+
+ // @Override
+ // public int hashCode() {
+ // int hashCode = arity * 47;
+ // for (TypeSerializer<?> ser : this.fieldSerializers) {
+ // hashCode = (hashCode << 7) | (hashCode >>> -7);
+ // hashCode += ser.hashCode();
+ // }
+ // return hashCode;
+ // }
+
+ // @Override
+ // public boolean equals(Object obj) {
+ // if (obj != null && obj instanceof StreamRecordSerializer) {
+ // StreamRecordSerializer otherTS = (StreamRecordSerializer) obj;
+ // return (otherTS.tupleClass == this.tupleClass)
+ // && Arrays.deepEquals(this.fieldSerializers, otherTS.fieldSerializers);
+ // } else {
+ // return false;
+ // }
+ // }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java
index 24893c2..06f059d 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/BroadcastPartitioner.java
@@ -20,6 +20,7 @@
package org.apache.flink.streaming.partitioner;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
/**
@@ -35,7 +36,8 @@ public class BroadcastPartitioner<T extends Tuple> implements StreamPartitioner<
boolean set;
@Override
- public int[] selectChannels(StreamRecord<T> record, int numberOfOutputChannels) {
+ public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
+ int numberOfOutputChannels) {
if (set) {
return returnArray;
} else {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java
index 3ce9d19..51bb54e 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/FieldsPartitioner.java
@@ -20,11 +20,12 @@
package org.apache.flink.streaming.partitioner;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
/**
- * Partitioner that selects the same (one) channel for two Tuples having a specified
- * fields equal.
+ * Partitioner that selects the same (one) channel for two Tuples having a
+ * specified fields equal.
*
* @param <T>
* Type of the Tuple
@@ -41,8 +42,9 @@ public class FieldsPartitioner<T extends Tuple> implements StreamPartitioner<T>
}
@Override
- public int[] selectChannels(StreamRecord<T> record, int numberOfOutputChannels) {
- returnArray[0] = Math.abs(record.getTuple().getField(keyPosition).hashCode())
+ public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
+ int numberOfOutputChannels) {
+ returnArray[0] = Math.abs(record.getInstance().getTuple().getField(keyPosition).hashCode())
% numberOfOutputChannels;
return returnArray;
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ForwardPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ForwardPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ForwardPartitioner.java
index 7b26570..88da063 100755
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ForwardPartitioner.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ForwardPartitioner.java
@@ -20,6 +20,7 @@
package org.apache.flink.streaming.partitioner;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
/**
@@ -38,7 +39,8 @@ public class ForwardPartitioner<T extends Tuple> implements StreamPartitioner<T>
}
@Override
- public int[] selectChannels(StreamRecord<T> record, int numberOfOutputChannels) {
+ public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
+ int numberOfOutputChannels) {
return returnArray;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java
index 7cad3dd..649f4b7 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/GlobalPartitioner.java
@@ -20,6 +20,7 @@
package org.apache.flink.streaming.partitioner;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
//Group to the partitioner with the lowest id
@@ -33,7 +34,8 @@ public class GlobalPartitioner<T extends Tuple> implements StreamPartitioner<T>
}
@Override
- public int[] selectChannels(StreamRecord<T> record, int numberOfOutputChannels) {
+ public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
+ int numberOfOutputChannels) {
return returnArray;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java
index 579250e..8d292e7 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/ShufflePartitioner.java
@@ -22,12 +22,13 @@ package org.apache.flink.streaming.partitioner;
import java.util.Random;
import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
/**
* Partitioner that distributes the data equally by selecting one output channel
* randomly.
- *
+ *
* @param <T>
* Type of the Tuple
*/
@@ -44,7 +45,8 @@ public class ShufflePartitioner<T extends Tuple> implements StreamPartitioner<T>
}
@Override
- public int[] selectChannels(StreamRecord<T> record, int numberOfOutputChannels) {
+ public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
+ int numberOfOutputChannels) {
returnArray[0] = random.nextInt(numberOfOutputChannels);
return returnArray;
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java
index 236e6b5..c0e9c1a 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/partitioner/StreamPartitioner.java
@@ -22,6 +22,7 @@ import java.io.Serializable;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.runtime.io.network.api.ChannelSelector;
+import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.streamrecord.StreamRecord;
/**
@@ -30,6 +31,6 @@ import org.apache.flink.streaming.api.streamrecord.StreamRecord;
* @param <T>
* Type of the Tuple
*/
-public interface StreamPartitioner<T extends Tuple> extends ChannelSelector<StreamRecord<T>>,
+public interface StreamPartitioner<T extends Tuple> extends ChannelSelector<SerializationDelegate<StreamRecord<T>>>,
Serializable {
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/PrintTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/PrintTest.java b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/PrintTest.java
index 923e4ac..c70cca3 100755
--- a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/PrintTest.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/PrintTest.java
@@ -19,18 +19,13 @@
package org.apache.flink.streaming.api;
-import java.util.HashSet;
-import java.util.Set;
-
-import org.apache.flink.streaming.api.LocalStreamEnvironment;
-import org.apache.flink.streaming.api.StreamExecutionEnvironment;
-import org.apache.flink.streaming.util.LogUtils;
-import org.junit.Test;
import org.apache.flink.api.java.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.util.LogUtils;
import org.apache.flink.util.Collector;
import org.apache.log4j.Level;
+import org.junit.Test;
public class PrintTest {
@@ -83,11 +78,11 @@ public class PrintTest {
LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
env.generateSequence(1, 10).print();
- Set<Integer> a = new HashSet<Integer>();
- a.add(-2);
- a.add(-100);
- env.fromCollection(a).print();
- env.executeTest(MEMORYSIZE);
+ env.execute();
+
}
+
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceTest.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceTest.java b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceTest.java
index 012c89a..d8c7213 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceTest.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/invokable/operator/BatchReduceTest.java
@@ -57,8 +57,9 @@ public class BatchReduceTest {
sum += values.next().f0;
count++;
}
-
- out.collect(new Tuple1<Double>(sum / count));
+ if(count>0){
+ out.collect(new Tuple1<Double>(sum / count));
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/0c1ef6d2/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamcomponent/MockRecordWriter.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamcomponent/MockRecordWriter.java b/flink-addons/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/streamcomponent/MockRecordWriter.java
old mode 100644
new mode 100755