You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2015/05/19 16:38:03 UTC
[6/7] flink git commit: [FLINK-1977] Rework Stream Operators to
always be push based
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
index 68ff532..1a177c6 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
@@ -21,18 +21,20 @@ import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Queue;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class FileMonitoringFunction implements SourceFunction<Tuple3<String, Long, Long>> {
+public class FileMonitoringFunction extends RichSourceFunction<Tuple3<String, Long, Long>> {
private static final long serialVersionUID = 1L;
private static final Logger LOG = LoggerFactory.getLogger(FileMonitoringFunction.class);
@@ -55,42 +57,21 @@ public class FileMonitoringFunction implements SourceFunction<Tuple3<String, Lon
private volatile boolean isRunning = false;
+ private Queue<Tuple3<String, Long, Long>> pendingFiles;
+
public FileMonitoringFunction(String path, long interval, WatchType watchType) {
this.path = path;
this.interval = interval;
this.watchType = watchType;
- this.modificationTimes = new HashMap<String, Long>();
- this.offsetOfFiles = new HashMap<String, Long>();
}
@Override
- public void run(Collector<Tuple3<String, Long, Long>> collector) throws Exception {
- isRunning = true;
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ this.modificationTimes = new HashMap<String, Long>();
+ this.offsetOfFiles = new HashMap<String, Long>();
+ this.pendingFiles = new LinkedList<Tuple3<String, Long, Long>>();
fileSystem = FileSystem.get(new URI(path));
-
- while (isRunning) {
- List<String> files = listNewFiles();
- for (String filePath : files) {
- if (watchType == WatchType.ONLY_NEW_FILES
- || watchType == WatchType.REPROCESS_WITH_APPENDED) {
- collector.collect(new Tuple3<String, Long, Long>(filePath, 0L, -1L));
- offsetOfFiles.put(filePath, -1L);
- } else if (watchType == WatchType.PROCESS_ONLY_APPENDED) {
- long offset = 0;
- long fileSize = fileSystem.getFileStatus(new Path(filePath)).getLen();
- if (offsetOfFiles.containsKey(filePath)) {
- offset = offsetOfFiles.get(filePath);
- }
-
- collector.collect(new Tuple3<String, Long, Long>(filePath, offset, fileSize));
- offsetOfFiles.put(filePath, fileSize);
-
- LOG.info("File processed: {}, {}, {}", filePath, offset, fileSize);
- }
- }
-
- Thread.sleep(interval);
- }
}
private List<String> listNewFiles() throws IOException {
@@ -126,8 +107,44 @@ public class FileMonitoringFunction implements SourceFunction<Tuple3<String, Lon
}
}
+
+ @Override
+ public boolean reachedEnd() throws Exception {
+ return false;
+ }
+
@Override
- public void cancel() {
- isRunning = false;
+ public Tuple3<String, Long, Long> next() throws Exception {
+ if (pendingFiles.size() > 0) {
+ return pendingFiles.poll();
+ } else {
+ while (true) {
+ List<String> files = listNewFiles();
+ for (String filePath : files) {
+ if (watchType == WatchType.ONLY_NEW_FILES
+ || watchType == WatchType.REPROCESS_WITH_APPENDED) {
+ pendingFiles.add(new Tuple3<String, Long, Long>(filePath, 0L, -1L));
+ offsetOfFiles.put(filePath, -1L);
+ } else if (watchType == WatchType.PROCESS_ONLY_APPENDED) {
+ long offset = 0;
+ long fileSize = fileSystem.getFileStatus(new Path(filePath)).getLen();
+ if (offsetOfFiles.containsKey(filePath)) {
+ offset = offsetOfFiles.get(filePath);
+ }
+
+ pendingFiles.add(new Tuple3<String, Long, Long>(filePath, offset, fileSize));
+ offsetOfFiles.put(filePath, fileSize);
+
+ LOG.info("File added to queue: {}, {}, {}", filePath, offset, fileSize);
+ }
+ }
+ if (files.size() > 0) {
+ break;
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ return pendingFiles.poll();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
index 9fa4d94..f1dcd2b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
@@ -27,7 +27,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputSplit;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
-import org.apache.flink.util.Collector;
public class FileSourceFunction extends RichParallelSourceFunction<String> {
private static final long serialVersionUID = 1L;
@@ -37,8 +36,13 @@ public class FileSourceFunction extends RichParallelSourceFunction<String> {
private InputFormat<String, ?> inputFormat;
private TypeInformation<String> typeInfo;
+ private transient TypeSerializer<String> serializer;
- private volatile boolean isRunning;
+ private InputFormat<String, InputSplit> format;
+
+ private Iterator<InputSplit> splitIterator;
+
+ private transient String nextElement;
public FileSourceFunction(InputFormat<String, ?> format, TypeInformation<String> typeInfo) {
this.inputFormat = format;
@@ -46,39 +50,25 @@ public class FileSourceFunction extends RichParallelSourceFunction<String> {
}
@Override
+ @SuppressWarnings("unchecked")
public void open(Configuration parameters) throws Exception {
StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();
this.provider = context.getInputSplitProvider();
inputFormat.configure(context.getTaskStubParameters());
+ serializer = typeInfo.createSerializer(getRuntimeContext().getExecutionConfig());
+
+ format = (InputFormat<String, InputSplit>) this.inputFormat;
+ splitIterator = getInputSplits();
+ if (splitIterator.hasNext()) {
+ format.open(splitIterator.next());
+ }
+
}
@Override
- public void run(Collector<String> collector) throws Exception {
- isRunning = true;
- final TypeSerializer<String> serializer = typeInfo.createSerializer(getRuntimeContext()
- .getExecutionConfig());
- final Iterator<InputSplit> splitIterator = getInputSplits();
- @SuppressWarnings("unchecked")
- final InputFormat<String, InputSplit> format = (InputFormat<String, InputSplit>) this.inputFormat;
- try {
- while (isRunning && splitIterator.hasNext()) {
-
- final InputSplit split = splitIterator.next();
- String record = serializer.createInstance();
-
- format.open(split);
- while (isRunning && !format.reachedEnd()) {
- if ((record = format.nextRecord(record)) != null) {
- collector.collect(record);
- }
- }
-
- }
- collector.close();
- } finally {
- format.close();
- }
- isRunning = false;
+ public void close() throws Exception {
+ super.close();
+ format.close();
}
private Iterator<InputSplit> getInputSplits() {
@@ -129,7 +119,30 @@ public class FileSourceFunction extends RichParallelSourceFunction<String> {
}
@Override
- public void cancel() {
- isRunning = false;
+ public boolean reachedEnd() throws Exception {
+ if (nextElement != null) {
+ return false;
+ }
+ nextElement = serializer.createInstance();
+ nextElement = format.nextRecord(nextElement);
+ if (nextElement == null && splitIterator.hasNext()) {
+ format.open(splitIterator.next());
+ return reachedEnd();
+ } else if (nextElement == null) {
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public String next() throws Exception {
+ if (reachedEnd()) {
+ throw new RuntimeException("End of FileSource reached.");
+ }
+
+ String result = nextElement;
+ nextElement = null;
+ return result;
}
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
index 5bc1eb0..6654361 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
@@ -19,15 +19,16 @@ package org.apache.flink.streaming.api.functions.source;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Iterator;
-import org.apache.flink.util.Collector;
+import org.apache.flink.configuration.Configuration;
-public class FromElementsFunction<T> implements SourceFunction<T> {
+public class FromElementsFunction<T> extends RichSourceFunction<T> {
private static final long serialVersionUID = 1L;
- Iterable<T> iterable;
+ private transient Iterator<T> iterator;
- private volatile boolean isRunning;
+ private Iterable<T> iterable;
public FromElementsFunction(T... elements) {
this.iterable = Arrays.asList(elements);
@@ -42,20 +43,19 @@ public class FromElementsFunction<T> implements SourceFunction<T> {
}
@Override
- public void run(Collector<T> collector) throws Exception {
- isRunning = true;
- for (T element : iterable) {
- if (isRunning) {
- collector.collect(element);
- } else {
- break;
- }
- }
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ iterator = iterable.iterator();
}
@Override
- public void cancel() {
- isRunning = false;
+ public boolean reachedEnd() throws Exception {
+ return !iterator.hasNext();
+ }
+
+ @Override
+ public T next() throws Exception {
+ return iterator.next();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/GenSequenceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/GenSequenceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/GenSequenceFunction.java
index 4878c1b..7d302d2 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/GenSequenceFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/GenSequenceFunction.java
@@ -18,7 +18,6 @@
package org.apache.flink.streaming.api.functions.source;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.util.Collector;
import org.apache.flink.util.NumberSequenceIterator;
/**
@@ -32,21 +31,11 @@ public class GenSequenceFunction extends RichParallelSourceFunction<Long> {
private NumberSequenceIterator fullIterator;
private NumberSequenceIterator splitIterator;
- private volatile boolean isRunning;
-
public GenSequenceFunction(long from, long to) {
fullIterator = new NumberSequenceIterator(from, to);
}
@Override
- public void run(Collector<Long> collector) throws Exception {
- isRunning = true;
- while (splitIterator.hasNext() && isRunning) {
- collector.collect(splitIterator.next());
- }
- }
-
- @Override
public void open(Configuration config) {
int splitNumber = getRuntimeContext().getIndexOfThisSubtask();
int numOfSubTasks = getRuntimeContext().getNumberOfParallelSubtasks();
@@ -54,8 +43,13 @@ public class GenSequenceFunction extends RichParallelSourceFunction<Long> {
}
@Override
- public void cancel() {
- isRunning = false;
+ public boolean reachedEnd() throws Exception {
+ return !splitIterator.hasNext();
+ }
+
+ @Override
+ public Long next() throws Exception {
+ return splitIterator.next();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
index b36ae39..6a1f930 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
@@ -26,7 +26,6 @@ import java.net.Socket;
import java.net.SocketException;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,7 +44,11 @@ public class SocketTextStreamFunction extends RichSourceFunction<String> {
private static final int CONNECTION_TIMEOUT_TIME = 0;
private static final int CONNECTION_RETRY_SLEEP = 1000;
- private volatile boolean isRunning = false;
+ private transient StringBuffer buffer;
+ private transient BufferedReader reader;
+
+ private boolean socketClosed;
+ private transient String nextElement;
public SocketTextStreamFunction(String hostname, int port, char delimiter, long maxRetry) {
this.hostname = hostname;
@@ -60,89 +63,102 @@ public class SocketTextStreamFunction extends RichSourceFunction<String> {
super.open(parameters);
socket = new Socket();
socket.connect(new InetSocketAddress(hostname, port), CONNECTION_TIMEOUT_TIME);
+ buffer = new StringBuffer();
+ reader = new BufferedReader(new InputStreamReader(
+ socket.getInputStream()));
+ socketClosed = false;
}
@Override
- public void run(Collector<String> collector) throws Exception {
- streamFromSocket(collector, socket);
+ public void close() throws Exception {
+ super.close();
+ if (reader != null) {
+ reader.close();
+ }
+ if (socket != null && !socket.isClosed()) {
+ try {
+ socket.close();
+ } catch (IOException e) {
+ if (LOG.isErrorEnabled()) {
+ LOG.error("Could not close open socket");
+ }
+ }
+ }
+
}
- public void streamFromSocket(Collector<String> collector, Socket socket) throws Exception {
- isRunning = true;
- try {
- StringBuffer buffer = new StringBuffer();
- BufferedReader reader = new BufferedReader(new InputStreamReader(
- socket.getInputStream()));
-
- while (isRunning) {
- int data;
- try {
- data = reader.read();
- } catch (SocketException e) {
- if (!isRunning) {
- break;
- } else {
- throw e;
- }
- }
+ public String blockingRead(Socket socket) throws Exception {
- if (data == -1) {
- socket.close();
- long retry = 0;
- boolean success = false;
- while (retry < maxRetry && !success) {
- if (!retryForever) {
- retry++;
- }
- LOG.warn("Lost connection to server socket. Retrying in "
- + (CONNECTION_RETRY_SLEEP / 1000) + " seconds...");
- try {
- socket = new Socket();
- socket.connect(new InetSocketAddress(hostname, port),
- CONNECTION_TIMEOUT_TIME);
- success = true;
- } catch (ConnectException ce) {
- Thread.sleep(CONNECTION_RETRY_SLEEP);
- }
- }
+ while (true) {
+ int data;
+ try {
+ data = reader.read();
+ } catch (SocketException e) {
+ socketClosed = true;
+ break;
+ }
- if (success) {
- LOG.info("Server socket is reconnected.");
- } else {
- LOG.error("Could not reconnect to server socket.");
- break;
+ if (data == -1) {
+ socket.close();
+ long retry = 0;
+ boolean success = false;
+ while (retry < maxRetry && !success) {
+ if (!retryForever) {
+ retry++;
+ }
+ LOG.warn("Lost connection to server socket. Retrying in "
+ + (CONNECTION_RETRY_SLEEP / 1000) + " seconds...");
+ try {
+ socket = new Socket();
+ socket.connect(new InetSocketAddress(hostname, port),
+ CONNECTION_TIMEOUT_TIME);
+ success = true;
+ } catch (ConnectException ce) {
+ Thread.sleep(CONNECTION_RETRY_SLEEP);
}
- reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
- continue;
}
- if (data == delimiter) {
- collector.collect(buffer.toString());
- buffer = new StringBuffer();
- } else if (data != '\r') { // ignore carriage return
- buffer.append((char) data);
+ if (success) {
+ LOG.info("Server socket is reconnected.");
+ } else {
+ LOG.error("Could not reconnect to server socket.");
+ break;
}
+ reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+ continue;
}
- if (buffer.length() > 0) {
- collector.collect(buffer.toString());
+ if (data == delimiter) {
+ String result = buffer.toString();
+ buffer = new StringBuffer();
+ return result;
+ } else if (data != '\r') { // ignore carriage return
+ buffer.append((char) data);
}
- } finally {
- socket.close();
}
+
+ return null;
}
+
@Override
- public void cancel() {
- isRunning = false;
- if (socket != null && !socket.isClosed()) {
- try {
- socket.close();
- } catch (IOException e) {
- if (LOG.isErrorEnabled()) {
- LOG.error("Could not close open socket");
- }
- }
+ public boolean reachedEnd() throws Exception {
+ if (socketClosed) {
+ return false;
}
+
+ nextElement = blockingRead(socket);
+
+ return nextElement == null;
}
+
+ @Override
+ public String next() throws Exception {
+ if (nextElement == null) {
+ reachedEnd();
+ }
+
+ return nextElement;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
index 9f0602f..e6ffdc1 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
@@ -1,56 +1,84 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.api.functions.source;
-
-import java.io.Serializable;
-
-import org.apache.flink.api.common.functions.Function;
-import org.apache.flink.util.Collector;
-
-/**
- * Interface for a stream data source.
- *
- * <p>Sources implementing this specific interface are executed with
- * parallelism 1. To execute your sources in parallel
- * see {@link ParallelSourceFunction}.</p>
- *
- * @param <OUT> The type of the records produced by this source.
- */
-public interface SourceFunction<OUT> extends Function, Serializable {
-
- /**
- * Main work method of the source. This function is invoked at the beginning of the
- * source's life and is expected to produce its data py "pushing" the records into
- * the given collector.
- *
- * @param collector The collector that forwards records to the source's consumers.
- *
- * @throws Exception Throwing any type of exception will cause the source to be considered
- * failed. When fault tolerance is enabled, recovery will be triggered,
- * which may create a new instance of this source.
- */
- public void run(Collector<OUT> collector) throws Exception;
-
- /**
- * This method signals the source function to cancel its operation
- * The method is called by the framework if the task is to be aborted prematurely.
- * This happens when the user cancels the job, or when the task is canceled as
- * part of a program failure and cleanup.
- */
- public void cancel();
-}
+/*
+ * 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.functions.source;
+
+import org.apache.flink.api.common.functions.Function;
+
+import java.io.Serializable;
+
+/**
+ * Base interface for all stream data sources in Flink. The contract of a stream source
+ * is similar to an iterator - it is consumed as in the following pseudo code:
+ *
+ * <pre>{@code
+ * StreamSource<T> source = ...;
+ * Collector<T> out = ...;
+ * while (!source.reachedEnd()) {
+ * out.collect(source.next());
+ * }
+ * }
+ * </pre>
+ *
+ * <b>Note about blocking behavior</b>
+ * <p>This implementations of the methods in the stream sources must have certain guarantees about
+ * blocking behavior. One of the two characteristics must be fulfilled.</p>
+ * <ul>
+ * <li>The methods must react to thread interrupt calls and break out of blocking calls with
+ * an {@link InterruptedException}.</li>
+ * <li>The method may ignore interrupt calls and/or swallow InterruptedExceptions, if it is guaranteed
+ * that the method returns quasi immediately irrespectively of the input. This is true for example
+ * for file streams, where the call is guaranteed to return after a very short I/O delay in
+ * the order of milliseconds.</li>
+ * </ul>
+ *
+ * @param <T> The type of the records produced by this source.
+ */
+public interface SourceFunction<T> extends Function, Serializable {
+
+ /**
+ * Checks whether the stream has reached its end.
+ *
+ * <p>This method must obey the contract about blocking behavior declared in the
+ * description of this class.</p>
+ *
+ * @return True, if the end of the stream has been reached, false if more data is available.
+ *
+ * @throws InterruptedException The calling thread may be interrupted to pull the function out of this
+ * method during checkpoints.
+ * @throws Exception Any other exception that is thrown causes the source to fail and results in failure of
+ * the streaming program, or triggers recovery, depending on the program setup.
+ */
+ boolean reachedEnd() throws Exception;
+
+
+ /**
+ * Produces the next record.
+ *
+ * <p>This method must obey the contract about blocking behavior declared in the
+ * description of this class.</p>
+ *
+ * @return The next record produced by this stream source.
+ *
+ * @throws InterruptedException The calling thread may be interrupted to pull the function out of this
+ * method during checkpoints.
+ * @throws Exception Any other exception that is thrown causes the source to fail and results in failure of
+ * the streaming program, or triggers recovery, depending on the program setup.
+ */
+ T next() throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/JSONGenerator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/JSONGenerator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/JSONGenerator.java
index a0e0a36..2c536d8 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/JSONGenerator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/JSONGenerator.java
@@ -165,16 +165,11 @@ public class JSONGenerator {
node.put(PACT, "Data Stream");
}
- StreamOperator<?, ?> operator = streamGraph.getStreamNode(vertexID).getOperator();
+ StreamOperator<?> operator = streamGraph.getStreamNode(vertexID).getOperator();
- if (operator != null && operator.getUserFunction() != null) {
- node.put(CONTENTS, vertex.getOperatorName() + " at "
- + operator.getUserFunction().getClass().getSimpleName());
- } else {
- node.put(CONTENTS, vertex.getOperatorName());
- }
+ node.put(CONTENTS, vertex.getOperatorName());
node.put(PARALLELISM, streamGraph.getStreamNode(vertexID).getParallelism());
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
index fdfec00..a1047df 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
@@ -162,7 +162,7 @@ public class StreamConfig implements Serializable {
return config.getLong(BUFFER_TIMEOUT, DEFAULT_TIMEOUT);
}
- public void setStreamOperator(StreamOperator<?, ?> operator) {
+ public void setStreamOperator(StreamOperator<?> operator) {
if (operator != null) {
config.setClass(USER_FUNCTION, operator.getClass());
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index 271c05c..1ad1be0 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -41,13 +41,15 @@ import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.StreamOperator;
-import org.apache.flink.streaming.api.operators.co.CoStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamSource;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer;
-import org.apache.flink.streaming.runtime.tasks.CoStreamTask;
+import org.apache.flink.streaming.runtime.tasks.TwoInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.SourceStreamTask;
import org.apache.flink.streaming.runtime.tasks.StreamIterationHead;
import org.apache.flink.streaming.runtime.tasks.StreamIterationTail;
-import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.sling.commons.json.JSONException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -130,16 +132,20 @@ public class StreamGraph extends StreamingPlan {
return !streamLoops.isEmpty();
}
- public <IN, OUT> void addSource(Integer vertexID, StreamOperator<IN, OUT> operatorObject,
+ public <IN, OUT> void addSource(Integer vertexID, StreamOperator<OUT> operatorObject,
TypeInformation<IN> inTypeInfo, TypeInformation<OUT> outTypeInfo, String operatorName) {
addOperator(vertexID, operatorObject, inTypeInfo, outTypeInfo, operatorName);
sources.add(vertexID);
}
- public <IN, OUT> void addOperator(Integer vertexID, StreamOperator<IN, OUT> operatorObject,
+ public <IN, OUT> void addOperator(Integer vertexID, StreamOperator<OUT> operatorObject,
TypeInformation<IN> inTypeInfo, TypeInformation<OUT> outTypeInfo, String operatorName) {
- addNode(vertexID, StreamTask.class, operatorObject, operatorName);
+ if (operatorObject instanceof StreamSource) {
+ addNode(vertexID, SourceStreamTask.class, operatorObject, operatorName);
+ } else {
+ addNode(vertexID, OneInputStreamTask.class, operatorObject, operatorName);
+ }
StreamRecordSerializer<IN> inSerializer = inTypeInfo != null ? new StreamRecordSerializer<IN>(
inTypeInfo, executionConfig) : null;
@@ -156,10 +162,10 @@ public class StreamGraph extends StreamingPlan {
}
public <IN1, IN2, OUT> void addCoOperator(Integer vertexID,
- CoStreamOperator<IN1, IN2, OUT> taskoperatorObject, TypeInformation<IN1> in1TypeInfo,
+ TwoInputStreamOperator<IN1, IN2, OUT> taskoperatorObject, TypeInformation<IN1> in1TypeInfo,
TypeInformation<IN2> in2TypeInfo, TypeInformation<OUT> outTypeInfo, String operatorName) {
- addNode(vertexID, CoStreamTask.class, taskoperatorObject, operatorName);
+ addNode(vertexID, TwoInputStreamTask.class, taskoperatorObject, operatorName);
StreamRecordSerializer<OUT> outSerializer = (outTypeInfo != null)
&& !(outTypeInfo instanceof MissingTypeInfo) ? new StreamRecordSerializer<OUT>(
@@ -228,7 +234,7 @@ public class StreamGraph extends StreamingPlan {
}
protected StreamNode addNode(Integer vertexID, Class<? extends AbstractInvokable> vertexClass,
- StreamOperator<?, ?> operatorObject, String operatorName) {
+ StreamOperator<?> operatorObject, String operatorName) {
StreamNode vertex = new StreamNode(environemnt, vertexID, operatorObject, operatorName,
new ArrayList<OutputSelector<?>>(), vertexClass);
@@ -286,7 +292,7 @@ public class StreamGraph extends StreamingPlan {
getStreamNode(vertexID).setSerializerOut(serializer);
}
- public <IN, OUT> void setOperator(Integer vertexID, StreamOperator<IN, OUT> operatorObject) {
+ public <IN, OUT> void setOperator(Integer vertexID, StreamOperator<OUT> operatorObject) {
getStreamNode(vertexID).setOperator(operatorObject);
}
@@ -337,10 +343,10 @@ public class StreamGraph extends StreamingPlan {
return streamNodes.values();
}
- public Set<Tuple2<Integer, StreamOperator<?, ?>>> getOperators() {
- Set<Tuple2<Integer, StreamOperator<?, ?>>> operatorSet = new HashSet<Tuple2<Integer, StreamOperator<?, ?>>>();
+ public Set<Tuple2<Integer, StreamOperator<?>>> getOperators() {
+ Set<Tuple2<Integer, StreamOperator<?>>> operatorSet = new HashSet<Tuple2<Integer, StreamOperator<?>>>();
for (StreamNode vertex : streamNodes.values()) {
- operatorSet.add(new Tuple2<Integer, StreamOperator<?, ?>>(vertex.getID(), vertex
+ operatorSet.add(new Tuple2<Integer, StreamOperator<?>>(vertex.getID(), vertex
.getOperator()));
}
return operatorSet;
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
index 7137e3e..147ed97 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
@@ -49,7 +49,7 @@ public class StreamNode implements Serializable {
private Integer slotSharingID;
private boolean isolatedSlot = false;
- private transient StreamOperator<?, ?> operator;
+ private transient StreamOperator<?> operator;
private List<OutputSelector<?>> outputSelectors;
private StreamRecordSerializer<?> typeSerializerIn1;
private StreamRecordSerializer<?> typeSerializerIn2;
@@ -62,7 +62,7 @@ public class StreamNode implements Serializable {
private InputFormat<String, ?> inputFormat;
- public StreamNode(StreamExecutionEnvironment env, Integer ID, StreamOperator<?, ?> operator,
+ public StreamNode(StreamExecutionEnvironment env, Integer ID, StreamOperator<?> operator,
String operatorName, List<OutputSelector<?>> outputSelector,
Class<? extends AbstractInvokable> jobVertexClass) {
this.env = env;
@@ -138,11 +138,11 @@ public class StreamNode implements Serializable {
this.bufferTimeout = bufferTimeout;
}
- public StreamOperator<?, ?> getOperator() {
+ public StreamOperator<?> getOperator() {
return operator;
}
- public void setOperator(StreamOperator<?, ?> operator) {
+ public void setOperator(StreamOperator<?> operator) {
this.operator = operator;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index e8acccc..0d314f1 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -308,8 +308,8 @@ public class StreamingJobGraphGenerator {
StreamNode upStreamVertex = edge.getSourceVertex();
StreamNode downStreamVertex = edge.getTargetVertex();
- StreamOperator<?, ?> headOperator = upStreamVertex.getOperator();
- StreamOperator<?, ?> outOperator = downStreamVertex.getOperator();
+ StreamOperator<?> headOperator = upStreamVertex.getOperator();
+ StreamOperator<?> outOperator = downStreamVertex.getOperator();
return downStreamVertex.getInEdges().size() == 1
&& outOperator != null
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/WindowingOptimizer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/WindowingOptimizer.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/WindowingOptimizer.java
index a0dcdf7..f375536 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/WindowingOptimizer.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/WindowingOptimizer.java
@@ -43,10 +43,10 @@ public class WindowingOptimizer {
@SuppressWarnings("rawtypes")
private static void removeMergeBeforeFlatten(StreamGraph streamGraph) {
- Set<Tuple2<Integer, StreamOperator<?, ?>>> operators = streamGraph.getOperators();
+ Set<Tuple2<Integer, StreamOperator<?>>> operators = streamGraph.getOperators();
List<Integer> flatteners = new ArrayList<Integer>();
- for (Tuple2<Integer, StreamOperator<?, ?>> entry : operators) {
+ for (Tuple2<Integer, StreamOperator<?>> entry : operators) {
if (entry.f1 instanceof WindowFlattener) {
flatteners.add(entry.f0);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
new file mode 100644
index 0000000..a8dc8c5
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
@@ -0,0 +1,67 @@
+/*
+ * 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.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+
+/**
+ * Base class for operators that do not contain a user-defined function.
+ *
+ * @param <OUT> The output type of the operator
+ */
+public abstract class AbstractStreamOperator<OUT> implements StreamOperator<OUT> {
+
+ private static final long serialVersionUID = 1L;
+
+ protected RuntimeContext runtimeContext;
+
+ protected ExecutionConfig executionConfig;
+
+ public Output<OUT> output;
+
+ // A sane default for most operators
+ protected ChainingStrategy chainingStrategy = ChainingStrategy.HEAD;
+
+ @Override
+ public final void setup(Output<OUT> output, RuntimeContext runtimeContext) {
+ this.output = output;
+ this.executionConfig = runtimeContext.getExecutionConfig();
+ this.runtimeContext = runtimeContext;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ }
+
+ @Override
+ public void close() throws Exception {
+ }
+
+ @Override
+ public final void setChainingStrategy(ChainingStrategy strategy) {
+ this.chainingStrategy = strategy;
+ }
+
+ @Override
+ public final ChainingStrategy getChainingStrategy() {
+ return chainingStrategy;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
new file mode 100644
index 0000000..dbd93b5
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
@@ -0,0 +1,93 @@
+/*
+ * 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.operators;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.checkpoint.CheckpointCommitter;
+import org.apache.flink.streaming.api.checkpoint.Checkpointed;
+
+import java.io.Serializable;
+
+/**
+ * This is used as the base class for operators that have a user-defined function.
+ *
+ * @param <OUT> The output type of the operator
+ * @param <F> The type of the user function
+ */
+public abstract class AbstractUdfStreamOperator<OUT, F extends Function> extends AbstractStreamOperator<OUT> implements StatefulStreamOperator<OUT> {
+
+ protected final F userFunction;
+
+ public AbstractUdfStreamOperator(F userFunction) {
+ this.userFunction = userFunction;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ FunctionUtils.setFunctionRuntimeContext(userFunction, runtimeContext);
+ FunctionUtils.openFunction(userFunction, parameters);
+ }
+
+ @Override
+ public void close() throws Exception{
+ super.close();
+ FunctionUtils.closeFunction(userFunction);
+ }
+
+ public void restoreInitialState(Serializable state) throws Exception {
+ if (userFunction instanceof Checkpointed) {
+ setStateOnFunction(state, userFunction);
+ }
+ else {
+ throw new IllegalStateException("Trying to restore state of a non-checkpointed function");
+ }
+ }
+
+ public Serializable getStateSnapshotFromFunction(long checkpointId, long timestamp) throws Exception {
+ if (userFunction instanceof Checkpointed) {
+ return ((Checkpointed<?>) userFunction).snapshotState(checkpointId, timestamp);
+ }
+ else {
+ return null;
+ }
+ }
+
+ public void confirmCheckpointCompleted(long checkpointId, long timestamp) throws Exception {
+ if (userFunction instanceof CheckpointCommitter) {
+ try {
+ ((CheckpointCommitter) userFunction).commitCheckpoint(checkpointId);
+ }
+ catch (Exception e) {
+ throw new Exception("Error while confirming checkpoint " + checkpointId + " to the stream function", e);
+ }
+ }
+ }
+
+ private static <T extends Serializable> void setStateOnFunction(Serializable state, Function function) {
+ @SuppressWarnings("unchecked")
+ T typedState = (T) state;
+ @SuppressWarnings("unchecked")
+ Checkpointed<T> typedFunction = (Checkpointed<T>) function;
+
+ typedFunction.restoreState(typedState);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/ChainableStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/ChainableStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/ChainableStreamOperator.java
deleted file mode 100644
index cc0790c..0000000
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/ChainableStreamOperator.java
+++ /dev/null
@@ -1,57 +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.operators;
-
-import org.apache.flink.api.common.functions.Function;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer;
-import org.apache.flink.util.Collector;
-
-public abstract class ChainableStreamOperator<IN, OUT> extends StreamOperator<IN, OUT> implements
- Collector<IN> {
-
- private static final long serialVersionUID = 1L;
- private boolean copyInput = true;
-
- public ChainableStreamOperator(Function userFunction) {
- super(userFunction);
- setChainingStrategy(ChainingStrategy.ALWAYS);
- }
-
- public void setup(Collector<OUT> collector, StreamRecordSerializer<IN> inSerializer) {
- this.collector = collector;
- this.inSerializer = inSerializer;
- this.objectSerializer = inSerializer.getObjectSerializer();
- }
-
- public ChainableStreamOperator<IN, OUT> withoutInputCopy() {
- copyInput = false;
- return this;
- }
-
- protected IN copyInput(IN input) {
- return copyInput ? copy(input) : input;
- }
-
- @Override
- public void collect(IN record) {
- if (isRunning) {
- nextObject = copyInput(record);
- callUserFunctionAndLogException();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/OneInputStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/OneInputStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/OneInputStreamOperator.java
new file mode 100644
index 0000000..d24ffed
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/OneInputStreamOperator.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+/**
+ * Interface for stream operators with one input. Use
+ * {@link org.apache.flink.streaming.api.operators.AbstractStreamOperator} as a base class if
+ * you want to implement a custom operator.
+ *
+ * @param <IN> The input type of the operator
+ * @param <OUT> The output type of the operator
+ */
+public interface OneInputStreamOperator<IN, OUT> extends StreamOperator<OUT> {
+ public void processElement(IN element) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/Output.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/Output.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/Output.java
new file mode 100644
index 0000000..d6f810a
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/Output.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.util.Collector;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperator} is supplied with an object
+ * of this interface that can be used to emit elements and other messages, such as barriers
+ * and low watermarks, from an operator.
+ *
+ * @param <T> The type of the elments that can be emitted.
+ */
+public interface Output<T> extends Collector<T> {
+ // NOTE: This does not yet have methods for barriers/low watermarks, this needs to be
+ // extended when this functionality arrives.
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java
new file mode 100644
index 0000000..e171af8
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StatefulStreamOperator.java
@@ -0,0 +1,35 @@
+/*
+ * 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.operators;
+
+import java.io.Serializable;
+
+/**
+ * Interface for Stream operators that can have state. This interface is used for checkpointing
+ * and restoring that state.
+ *
+ * @param <OUT> The output type of the operator
+ */
+public interface StatefulStreamOperator<OUT> extends StreamOperator<OUT> {
+
+ void restoreInitialState(Serializable state) throws Exception;
+
+ Serializable getStateSnapshotFromFunction(long checkpointId, long timestamp) throws Exception;
+
+ void confirmCheckpointCompleted(long checkpointId, long timestamp) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamCounter.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamCounter.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamCounter.java
index 4c997d5..d9a67dd 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamCounter.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamCounter.java
@@ -17,28 +17,16 @@
package org.apache.flink.streaming.api.operators;
+public class StreamCounter<IN> extends AbstractStreamOperator<Long> implements OneInputStreamOperator<IN, Long> {
-public class StreamCounter<IN> extends ChainableStreamOperator<IN, Long> {
- private static final long serialVersionUID = 1L;
-
- Long count = 0L;
+ private Long count = 0L;
public StreamCounter() {
- super(null);
+ chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- collector.collect(++count);
- }
+ public void processElement(IN element) {
+ output.collect(++count);
}
-
- @Override
- public void collect(IN record) {
- if (isRunning) {
- collector.collect(++count);
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java
index 898f5ef..e5575db 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFilter.java
@@ -19,29 +19,17 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.functions.FilterFunction;
-public class StreamFilter<IN> extends ChainableStreamOperator<IN, IN> {
-
- private static final long serialVersionUID = 1L;
-
- private boolean collect;
+public class StreamFilter<IN> extends AbstractUdfStreamOperator<IN, FilterFunction<IN>> implements OneInputStreamOperator<IN, IN> {
public StreamFilter(FilterFunction<IN> filterFunction) {
super(filterFunction);
+ chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
- }
-
- @Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- collect = ((FilterFunction<IN>) userFunction).filter(nextObject);
- if (collect) {
- collector.collect(nextObject);
+ public void processElement(IN element) throws Exception {
+ if (userFunction.filter(element)) {
+ output.collect(element);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
index 2b8a3a8..1e836b1 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java
@@ -19,24 +19,17 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.functions.FlatMapFunction;
-public class StreamFlatMap<IN, OUT> extends ChainableStreamOperator<IN, OUT> {
- private static final long serialVersionUID = 1L;
+public class StreamFlatMap<IN, OUT>
+ extends AbstractUdfStreamOperator<OUT, FlatMapFunction<IN, OUT>>
+ implements OneInputStreamOperator<IN, OUT> {
public StreamFlatMap(FlatMapFunction<IN, OUT> flatMapper) {
super(flatMapper);
+ chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
+ public void processElement(IN element) throws Exception {
+ userFunction.flatMap(element, output);
}
-
- @Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- ((FlatMapFunction<IN, OUT>) userFunction).flatMap(nextObject, collector);
- }
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFold.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFold.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFold.java
index 542f65c..1b12e89 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFold.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamFold.java
@@ -20,31 +20,32 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.configuration.Configuration;
-public class StreamFold<IN, OUT> extends ChainableStreamOperator<IN, OUT> {
- private static final long serialVersionUID = 1L;
+public class StreamFold<IN, OUT>
+ extends AbstractUdfStreamOperator<OUT, FoldFunction<IN, OUT>>
+ implements OneInputStreamOperator<IN, OUT> {
private OUT accumulator;
protected TypeSerializer<OUT> outTypeSerializer;
+ protected TypeInformation<OUT> outTypeInformation;
- public StreamFold(FoldFunction<IN, OUT> folder, OUT initialValue,
- TypeInformation<OUT> outTypeInformation) {
+ public StreamFold(FoldFunction<IN, OUT> folder, OUT initialValue, TypeInformation<OUT> outTypeInformation) {
super(folder);
this.accumulator = initialValue;
- this.outTypeSerializer = outTypeInformation.createSerializer(executionConfig);
+ this.outTypeInformation = outTypeInformation;
+ this.chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
+ public void processElement(IN element) throws Exception {
+ accumulator = userFunction.fold(outTypeSerializer.copy(accumulator), element);
+ output.collect(accumulator);
}
@Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- accumulator = ((FoldFunction<IN, OUT>) userFunction).fold(outTypeSerializer.copy(accumulator), nextObject);
- collector.collect(accumulator);
+ public void open(Configuration config) throws Exception {
+ super.open(config);
+ this.outTypeSerializer = outTypeInformation.createSerializer(executionConfig);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedFold.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedFold.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedFold.java
index 88c75df..75217be 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedFold.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedFold.java
@@ -25,7 +25,6 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
public class StreamGroupedFold<IN, OUT> extends StreamFold<IN, OUT> {
- private static final long serialVersionUID = 1L;
private KeySelector<IN, ?> keySelector;
private Map<Object, OUT> values;
@@ -40,20 +39,19 @@ public class StreamGroupedFold<IN, OUT> extends StreamFold<IN, OUT> {
}
@Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- Object key = nextRecord.getKey(keySelector);
+ public void processElement(IN element) throws Exception {
+ Object key = keySelector.getKey(element);
OUT accumulator = values.get(key);
FoldFunction<IN, OUT> folder = ((FoldFunction<IN, OUT>) userFunction);
if (accumulator != null) {
- OUT folded = folder.fold(outTypeSerializer.copy(accumulator), nextObject);
+ OUT folded = folder.fold(outTypeSerializer.copy(accumulator), element);
values.put(key, folded);
- collector.collect(folded);
+ output.collect(folded);
} else {
- OUT first = folder.fold(outTypeSerializer.copy(initialValue), nextObject);
+ OUT first = folder.fold(outTypeSerializer.copy(initialValue), element);
values.put(key, first);
- collector.collect(first);
+ output.collect(first);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedReduce.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedReduce.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedReduce.java
index d254fd4..e3980ce 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedReduce.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamGroupedReduce.java
@@ -24,7 +24,6 @@ import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
public class StreamGroupedReduce<IN> extends StreamReduce<IN> {
- private static final long serialVersionUID = 1L;
private KeySelector<IN, ?> keySelector;
private Map<Object, IN> values;
@@ -36,17 +35,17 @@ public class StreamGroupedReduce<IN> extends StreamReduce<IN> {
}
@Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- Object key = keySelector.getKey(nextObject);
+ public void processElement(IN element) throws Exception {
+ Object key = keySelector.getKey(element);
IN currentValue = values.get(key);
if (currentValue != null) {
- IN reduced = ((ReduceFunction<IN>) userFunction).reduce(copy(currentValue), nextObject);
+ // TODO: find a way to let operators copy elements (maybe)
+ IN reduced = userFunction.reduce(currentValue, element);
values.put(key, reduced);
- collector.collect(reduced);
+ output.collect(reduced);
} else {
- values.put(key, nextObject);
- collector.collect(nextObject);
+ values.put(key, element);
+ output.collect(element);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java
index f421f5d..a379c56 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamMap.java
@@ -19,24 +19,17 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.functions.MapFunction;
-public class StreamMap<IN, OUT> extends ChainableStreamOperator<IN, OUT> {
- private static final long serialVersionUID = 1L;
+public class StreamMap<IN, OUT>
+ extends AbstractUdfStreamOperator<OUT, MapFunction<IN, OUT>>
+ implements OneInputStreamOperator<IN, OUT> {
public StreamMap(MapFunction<IN, OUT> mapper) {
super(mapper);
+ chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
+ public void processElement(IN element) throws Exception {
+ output.collect(userFunction.map(element));
}
-
- @Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- collector.collect(((MapFunction<IN, OUT>) userFunction).map(nextObject));
- }
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
index 24a08eb..6fd3a3c 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamOperator.java
@@ -17,183 +17,41 @@
package org.apache.flink.streaming.api.operators;
-import java.io.IOException;
-import java.io.Serializable;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.common.functions.util.FunctionUtils;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.checkpoint.CheckpointCommitter;
-import org.apache.flink.streaming.api.checkpoint.Checkpointed;
-import org.apache.flink.streaming.runtime.io.IndexedReaderIterator;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer;
-import org.apache.flink.streaming.runtime.tasks.StreamTaskContext;
-import org.apache.flink.util.Collector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
/**
- * The StreamOperator represents the base class for all operators in the
- * streaming topology.
+ * Basic interface for stream operators. Implementers would implement one of
+ * {@link org.apache.flink.streaming.api.operators.OneInputStreamOperator} or
+ * {@link org.apache.flink.streaming.api.operators.TwoInputStreamOperator} to create operators
+ * that process elements. You can use
+ * {@link org.apache.flink.streaming.api.operators.AbstractStreamOperator} as a base class for
+ * custom operators.
*
- * @param <OUT>
- * The output type of the operator
+ * @param <OUT> The output type of the operator
*/
-public abstract class StreamOperator<IN, OUT> implements Serializable {
-
- private static final long serialVersionUID = 1L;
- private static final Logger LOG = LoggerFactory.getLogger(StreamOperator.class);
-
- protected StreamTaskContext<OUT> taskContext;
-
- protected ExecutionConfig executionConfig = null;
-
- protected IndexedReaderIterator<StreamRecord<IN>> recordIterator;
- protected StreamRecordSerializer<IN> inSerializer;
- protected TypeSerializer<IN> objectSerializer;
- protected StreamRecord<IN> nextRecord;
- protected IN nextObject;
-
- public Collector<OUT> collector;
- protected Function userFunction;
- protected volatile boolean isRunning;
-
- private ChainingStrategy chainingStrategy = ChainingStrategy.HEAD;
-
- public StreamOperator(Function userFunction) {
- this.userFunction = userFunction;
- }
-
- /**
- * Initializes the {@link StreamOperator} for input and output handling
- *
- * @param taskContext
- * StreamTaskContext representing the vertex
- */
- public void setup(StreamTaskContext<OUT> taskContext) {
- this.collector = taskContext.getOutputCollector();
- this.recordIterator = taskContext.getIndexedInput(0);
- this.inSerializer = taskContext.getInputSerializer(0);
- if (this.inSerializer != null) {
- this.nextRecord = inSerializer.createInstance();
- this.objectSerializer = inSerializer.getObjectSerializer();
- }
- this.taskContext = taskContext;
- this.executionConfig = taskContext.getExecutionConfig();
- }
-
- /**
- * Method that will be called when the operator starts, should encode the
- * processing logic
- */
- public abstract void run() throws Exception;
-
- /*
- * Reads the next record from the reader iterator and stores it in the
- * nextRecord variable
- */
- protected StreamRecord<IN> readNext() throws IOException {
- this.nextRecord = inSerializer.createInstance();
- try {
- nextRecord = recordIterator.next(nextRecord);
- try {
- nextObject = nextRecord.getObject();
- } catch (NullPointerException e) {
- // end of stream
- }
- return nextRecord;
- } catch (IOException e) {
- if (isRunning) {
- throw new RuntimeException("Could not read next record", e);
- } else {
- // Task already cancelled do nothing
- return null;
- }
- } catch (IllegalStateException e) {
- if (isRunning) {
- throw new RuntimeException("Could not read next record", e);
- } else {
- // Task already cancelled do nothing
- return null;
- }
- }
- }
+public interface StreamOperator<OUT> extends Serializable {
/**
- * The call of the user implemented function should be implemented here
+ * Initializes the {@link StreamOperator} for input and output handling.
*/
- protected void callUserFunction() throws Exception {
- }
-
- /**
- * Method for logging exceptions thrown during the user function call
- */
- protected void callUserFunctionAndLogException() {
- try {
- callUserFunction();
- } catch (Exception e) {
- if (LOG.isErrorEnabled()) {
- LOG.error("Calling user function failed", e);
- }
- throw new RuntimeException(e);
- }
- }
+ public void setup(Output<OUT> output, RuntimeContext runtimeContext);
/**
- * Open method to be used if the user defined function extends the
- * RichFunction class
- *
- * @param parameters
- * The configuration parameters for the operator
+ * This method is called before any elements are processed.
*/
- public void open(Configuration parameters) throws Exception {
- isRunning = true;
- FunctionUtils.openFunction(userFunction, parameters);
- }
+ public void open(Configuration config) throws Exception;
/**
- * Close method to be used if the user defined function extends the
- * RichFunction class
- *
+ * This method is called after no more elements for can arrive for processing.
*/
- public void close() {
- isRunning = false;
- collector.close();
- try {
- FunctionUtils.closeFunction(userFunction);
- } catch (Exception e) {
- throw new RuntimeException("Error when closing the function", e);
- }
- }
+ public void close() throws Exception;
- public void cancel() {
- isRunning = false;
- }
+ public void setChainingStrategy(ChainingStrategy strategy);
- public void setRuntimeContext(RuntimeContext t) {
- FunctionUtils.setFunctionRuntimeContext(userFunction, t);
- }
-
- protected IN copy(IN record) {
- return objectSerializer.copy(record);
- }
-
- public void setChainingStrategy(ChainingStrategy strategy) {
- if (strategy == ChainingStrategy.ALWAYS) {
- if (!(this instanceof ChainableStreamOperator)) {
- throw new RuntimeException("Operator needs to extend ChainableOperator to be chained");
- }
- }
- this.chainingStrategy = strategy;
- }
-
- public ChainingStrategy getChainingStrategy() {
- return chainingStrategy;
- }
+ public ChainingStrategy getChainingStrategy();
/**
* Defines the chaining scheme for the operator. By default <b>ALWAYS</b> is used,
@@ -208,53 +66,4 @@ public abstract class StreamOperator<IN, OUT> implements Serializable {
public static enum ChainingStrategy {
ALWAYS, NEVER, HEAD
}
-
- public Function getUserFunction() {
- return userFunction;
- }
-
- // ------------------------------------------------------------------------
- // Checkpoints and Checkpoint Confirmations
- // ------------------------------------------------------------------------
-
- // NOTE - ALL OF THIS CODE WORKS ONLY FOR THE FIRST OPERATOR IN THE CHAIN
- // IT NEEDS TO BE EXTENDED TO SUPPORT CHAINS
-
- public void restoreInitialState(Serializable state) throws Exception {
- if (userFunction instanceof Checkpointed) {
- setStateOnFunction(state, userFunction);
- }
- else {
- throw new IllegalStateException("Trying to restore state of a non-checkpointed function");
- }
- }
-
- public Serializable getStateSnapshotFromFunction(long checkpointId, long timestamp) throws Exception {
- if (userFunction instanceof Checkpointed) {
- return ((Checkpointed<?>) userFunction).snapshotState(checkpointId, timestamp);
- }
- else {
- return null;
- }
- }
-
- public void confirmCheckpointCompleted(long checkpointId, long timestamp) throws Exception {
- if (userFunction instanceof CheckpointCommitter) {
- try {
- ((CheckpointCommitter) userFunction).commitCheckpoint(checkpointId);
- }
- catch (Exception e) {
- throw new Exception("Error while confirming checkpoint " + checkpointId + " to the stream function", e);
- }
- }
- }
-
- private static <T extends Serializable> void setStateOnFunction(Serializable state, Function function) {
- @SuppressWarnings("unchecked")
- T typedState = (T) state;
- @SuppressWarnings("unchecked")
- Checkpointed<T> typedFunction = (Checkpointed<T>) function;
-
- typedFunction.restoreState(typedState);
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java
index 7f8b10d..d039144 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamProject.java
@@ -22,8 +22,9 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
-public class StreamProject<IN, OUT extends Tuple> extends ChainableStreamOperator<IN, OUT> {
- private static final long serialVersionUID = 1L;
+public class StreamProject<IN, OUT extends Tuple>
+ extends AbstractStreamOperator<OUT>
+ implements OneInputStreamOperator<IN, OUT> {
transient OUT outTuple;
TypeSerializer<OUT> outTypeSerializer;
@@ -32,25 +33,20 @@ public class StreamProject<IN, OUT extends Tuple> extends ChainableStreamOperato
int numFields;
public StreamProject(int[] fields, TypeInformation<OUT> outTypeInformation) {
- super(null);
this.fields = fields;
this.numFields = this.fields.length;
this.outTypeInformation = outTypeInformation;
- }
- @Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
+ chainingStrategy = ChainingStrategy.ALWAYS;
}
+
@Override
- protected void callUserFunction() throws Exception {
+ public void processElement(IN element) throws Exception {
for (int i = 0; i < this.numFields; i++) {
- outTuple.setField(((Tuple)nextObject).getField(fields[i]), i);
+ outTuple.setField(((Tuple) element).getField(fields[i]), i);
}
- collector.collect(outTuple);
+ output.collect(outTuple);
}
@Override
@@ -59,5 +55,4 @@ public class StreamProject<IN, OUT extends Tuple> extends ChainableStreamOperato
this.outTypeSerializer = outTypeInformation.createSerializer(executionConfig);
outTuple = outTypeSerializer.createInstance();
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamReduce.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamReduce.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamReduce.java
index fdf1284..8205fe6 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamReduce.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamReduce.java
@@ -19,35 +19,28 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.functions.ReduceFunction;
-public class StreamReduce<IN> extends ChainableStreamOperator<IN, IN> {
- private static final long serialVersionUID = 1L;
+public class StreamReduce<IN> extends AbstractUdfStreamOperator<IN, ReduceFunction<IN>>
+ implements OneInputStreamOperator<IN, IN> {
private IN currentValue;
public StreamReduce(ReduceFunction<IN> reducer) {
super(reducer);
currentValue = null;
- }
- @Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
+ chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
+ public void processElement(IN element) throws Exception {
if (currentValue != null) {
- currentValue = ((ReduceFunction<IN>) userFunction).reduce(copy(currentValue), nextObject);
+ // TODO: give operator a way to specify that elements should be copied
+ currentValue = userFunction.reduce(currentValue, element);
} else {
- currentValue = nextObject;
+ currentValue = element;
}
- collector.collect(currentValue);
-
+ output.collect(currentValue);
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
index 26e37fa..b1a0212 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
@@ -19,23 +19,17 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-public class StreamSink<IN> extends ChainableStreamOperator<IN, IN> {
- private static final long serialVersionUID = 1L;
+public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFunction<IN>>
+ implements OneInputStreamOperator<IN, Object> {
public StreamSink(SinkFunction<IN> sinkFunction) {
super(sinkFunction);
- }
- @Override
- public void run() throws Exception {
- while (isRunning && readNext() != null) {
- callUserFunctionAndLogException();
- }
+ chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- ((SinkFunction<IN>) userFunction).invoke(nextObject);
+ public void processElement(IN element) throws Exception {
+ userFunction.invoke(element);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
index ef253ac..187f4f0 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
@@ -17,33 +17,27 @@
package org.apache.flink.streaming.api.operators;
-import java.io.Serializable;
-
import org.apache.flink.streaming.api.functions.source.SourceFunction;
-public class StreamSource<OUT> extends StreamOperator<OUT, OUT> implements Serializable {
-
- private static final long serialVersionUID = 1L;
+public class StreamSource<OUT> extends AbstractUdfStreamOperator<OUT, SourceFunction<OUT>> implements StreamOperator<OUT> {
public StreamSource(SourceFunction<OUT> sourceFunction) {
super(sourceFunction);
}
- @Override
- public void run() {
- callUserFunctionAndLogException();
- }
+ public void run() throws Exception {
+ while (true) {
- @Override
- @SuppressWarnings("unchecked")
- protected void callUserFunction() throws Exception {
- ((SourceFunction<OUT>) userFunction).run(collector);
- }
+ synchronized (userFunction) {
+ if (userFunction.reachedEnd()) {
+ break;
+ }
+
+ OUT result = userFunction.next();
- @Override
- @SuppressWarnings("unchecked")
- public void cancel() {
- super.cancel();
- ((SourceFunction<OUT>) userFunction).cancel();
+ output.collect(result);
+ }
+ Thread.yield();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/58865ff3/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TwoInputStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TwoInputStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TwoInputStreamOperator.java
new file mode 100644
index 0000000..2b3090b
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/TwoInputStreamOperator.java
@@ -0,0 +1,35 @@
+/*
+ * 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.operators;
+
+/**
+ * Interface for stream operators with two inputs. Use
+ * {@link org.apache.flink.streaming.api.operators.AbstractStreamOperator} as a base class if
+ * you want to implement a custom operator.
+ *
+ * @param <IN1> The input type of the operator
+ * @param <IN2> The input type of the operator
+ * @param <OUT> The output type of the operator
+ */
+public interface TwoInputStreamOperator<IN1, IN2, OUT> extends StreamOperator<OUT> {
+
+ public void processElement1(IN1 element) throws Exception;
+
+ public void processElement2(IN2 element) throws Exception;
+}