flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [44/51] [abbrv] git commit: [streaming] DataStream fix to return DataStreamSink on closing operators
Date Mon, 18 Aug 2014 17:26:21 GMT
[streaming] DataStream fix to return DataStreamSink on closing operators


Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/74d3742b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/74d3742b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/74d3742b

Branch: refs/heads/master
Commit: 74d3742b6e39c174111d3b0a6177c25b93f9c926
Parents: 1b31f4d
Author: gyfora <gyula.fora@gmail.com>
Authored: Wed Aug 6 17:22:06 2014 +0200
Committer: Stephan Ewen <sewen@apache.org>
Committed: Mon Aug 18 16:23:43 2014 +0200

----------------------------------------------------------------------
 .../streaming/api/datastream/DataStream.java    | 48 ++++++++++----------
 1 file changed, 24 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/74d3742b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index 7cab2df..86b3322 100644
--- a/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-addons/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -353,10 +353,10 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream.
 	 */
-	public DataStream<OUT> print() {
+	public DataStreamSink<OUT> print() {
 		DataStream<OUT> inputStream = this.copy();
 		PrintSinkFunction<OUT> printFunction = new PrintSinkFunction<OUT>();
-		DataStream<OUT> returnStream = addSink(inputStream, printFunction, null);
+		DataStreamSink<OUT> returnStream = addSink(inputStream, printFunction, null);
 
 		jobGraphBuilder.setBytesFrom(inputStream.getId(), returnStream.getId());
 
@@ -373,7 +373,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsText(String path) {
+	public DataStreamSink<OUT> writeAsText(String path) {
 		return writeAsText(this, path, new WriteFormatAsText<OUT>(), 1, null);
 	}
 
@@ -390,7 +390,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsText(String path, long millis) {
+	public DataStreamSink<OUT> writeAsText(String path, long millis) {
 		return writeAsText(this, path, new WriteFormatAsText<OUT>(), millis, null);
 	}
 
@@ -408,7 +408,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsText(String path, int batchSize) {
+	public DataStreamSink<OUT> writeAsText(String path, int batchSize) {
 		return writeAsText(this, path, new WriteFormatAsText<OUT>(), batchSize, null);
 	}
 
@@ -430,7 +430,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsText(String path, long millis, OUT endTuple) {
+	public DataStreamSink<OUT> writeAsText(String path, long millis, OUT endTuple) {
 		return writeAsText(this, path, new WriteFormatAsText<OUT>(), millis, endTuple);
 	}
 
@@ -453,7 +453,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsText(String path, int batchSize, OUT endTuple) {
+	public DataStreamSink<OUT> writeAsText(String path, int batchSize, OUT endTuple) {
 		return writeAsText(this, path, new WriteFormatAsText<OUT>(), batchSize, endTuple);
 	}
 
@@ -475,9 +475,9 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return the data stream constructed
 	 */
-	private DataStream<OUT> writeAsText(DataStream<OUT> inputStream, String path,
+	private DataStreamSink<OUT> writeAsText(DataStream<OUT> inputStream, String
path,
 			WriteFormatAsText<OUT> format, long millis, OUT endTuple) {
-		DataStream<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByMillis<OUT>(
+		DataStreamSink<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByMillis<OUT>(
 				path, format, millis, endTuple), null);
 		jobGraphBuilder.setBytesFrom(inputStream.getId(), returnStream.getId());
 		jobGraphBuilder.setMutability(returnStream.getId(), false);
@@ -503,9 +503,9 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return the data stream constructed
 	 */
-	private DataStream<OUT> writeAsText(DataStream<OUT> inputStream, String path,
+	private DataStreamSink<OUT> writeAsText(DataStream<OUT> inputStream, String
path,
 			WriteFormatAsText<OUT> format, int batchSize, OUT endTuple) {
-		DataStream<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByBatches<OUT>(
+		DataStreamSink<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByBatches<OUT>(
 				path, format, batchSize, endTuple), null);
 		jobGraphBuilder.setBytesFrom(inputStream.getId(), returnStream.getId());
 		jobGraphBuilder.setMutability(returnStream.getId(), false);
@@ -522,7 +522,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsCsv(String path) {
+	public DataStreamSink<OUT> writeAsCsv(String path) {
 		return writeAsCsv(this, path, new WriteFormatAsCsv<OUT>(), 1, null);
 	}
 
@@ -539,7 +539,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsCsv(String path, long millis) {
+	public DataStreamSink<OUT> writeAsCsv(String path, long millis) {
 		return writeAsCsv(this, path, new WriteFormatAsCsv<OUT>(), millis, null);
 	}
 
@@ -557,7 +557,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsCsv(String path, int batchSize) {
+	public DataStreamSink<OUT> writeAsCsv(String path, int batchSize) {
 		return writeAsCsv(this, path, new WriteFormatAsCsv<OUT>(), batchSize, null);
 	}
 
@@ -579,7 +579,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsCsv(String path, long millis, OUT endTuple) {
+	public DataStreamSink<OUT> writeAsCsv(String path, long millis, OUT endTuple) {
 		return writeAsCsv(this, path, new WriteFormatAsCsv<OUT>(), millis, endTuple);
 	}
 
@@ -602,7 +602,7 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return The closed DataStream
 	 */
-	public DataStream<OUT> writeAsCsv(String path, int batchSize, OUT endTuple) {
+	public DataStreamSink<OUT> writeAsCsv(String path, int batchSize, OUT endTuple) {
 		if (this instanceof SingleOutputStreamOperator) {
 			((SingleOutputStreamOperator<?, ?>) this).setMutability(false);
 		}
@@ -627,9 +627,9 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return the data stream constructed
 	 */
-	private DataStream<OUT> writeAsCsv(DataStream<OUT> inputStream, String path,
+	private DataStreamSink<OUT> writeAsCsv(DataStream<OUT> inputStream, String path,
 			WriteFormatAsCsv<OUT> format, long millis, OUT endTuple) {
-		DataStream<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByMillis<OUT>(
+		DataStreamSink<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByMillis<OUT>(
 				path, format, millis, endTuple));
 		jobGraphBuilder.setBytesFrom(inputStream.getId(), returnStream.getId());
 		jobGraphBuilder.setMutability(returnStream.getId(), false);
@@ -655,9 +655,9 @@ public abstract class DataStream<OUT> {
 	 * 
 	 * @return the data stream constructed
 	 */
-	private DataStream<OUT> writeAsCsv(DataStream<OUT> inputStream, String path,
+	private DataStreamSink<OUT> writeAsCsv(DataStream<OUT> inputStream, String path,
 			WriteFormatAsCsv<OUT> format, int batchSize, OUT endTuple) {
-		DataStream<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByBatches<OUT>(
+		DataStreamSink<OUT> returnStream = addSink(inputStream, new WriteSinkFunctionByBatches<OUT>(
 				path, format, batchSize, endTuple), null);
 		jobGraphBuilder.setBytesFrom(inputStream.getId(), returnStream.getId());
 		jobGraphBuilder.setMutability(returnStream.getId(), false);
@@ -792,18 +792,18 @@ public abstract class DataStream<OUT> {
 	 *            The object containing the sink's invoke function.
 	 * @return The closed DataStream.
 	 */
-	public DataStream<OUT> addSink(SinkFunction<OUT> sinkFunction) {
+	public DataStreamSink<OUT> addSink(SinkFunction<OUT> sinkFunction) {
 		return addSink(this.copy(), sinkFunction);
 	}
 
-	private DataStream<OUT> addSink(DataStream<OUT> inputStream, SinkFunction<OUT>
sinkFunction) {
+	private DataStreamSink<OUT> addSink(DataStream<OUT> inputStream, SinkFunction<OUT>
sinkFunction) {
 		return addSink(inputStream, sinkFunction, new FunctionTypeWrapper<OUT, Tuple, OUT>(
 				sinkFunction, SinkFunction.class, 0, -1, 0));
 	}
 
-	private DataStream<OUT> addSink(DataStream<OUT> inputStream, SinkFunction<OUT>
sinkFunction,
+	private DataStreamSink<OUT> addSink(DataStream<OUT> inputStream, SinkFunction<OUT>
sinkFunction,
 			TypeSerializerWrapper<OUT, Tuple, OUT> typeWrapper) {
-		DataStream<OUT> returnStream = new DataStreamSink<OUT>(environment, "sink");
+		DataStreamSink<OUT> returnStream = new DataStreamSink<OUT>(environment, "sink");
 
 		try {
 			jobGraphBuilder.addSink(returnStream.getId(), new SinkInvokable<OUT>(sinkFunction),


Mime
View raw message