This class defines all the streaming operations.
*
* @param This stream does not create a physical operation, it only affects how upstream data are
- * connected to downstream data.
+ * connected to downstream data.
*
* @param Python worker can use information in this class to create a function object. If this object is constructed from serialized python function,
- * python worker can deserialize it to create python function directly.
- * If this object is constructed from moduleName and className/functionName,
- * python worker will use `importlib` to load python function. If the python data stream api is invoked from python, `function` will be not null. If the python data stream api is invoked from java, `moduleName` and
* `functionName` will be not null.
*/
public class PythonFunction implements Function {
+
public enum FunctionInterface {
SOURCE_FUNCTION("SourceFunction"),
MAP_FUNCTION("MapFunction"),
@@ -47,8 +45,8 @@ public class PythonFunction implements Function {
// null if this function is constructed from serialized python function.
private final String functionName;
/**
- * FunctionInterface can be used to validate python function,
- * and look up operator class from FunctionInterface.
+ * FunctionInterface can be used to validate python function, and look up operator class from
+ * FunctionInterface.
*/
private String functionInterface;
@@ -69,10 +67,12 @@ public class PythonFunction implements Function {
*
* @param moduleName module name of streaming function.
* @param functionName function name of streaming function. {@code functionName} is the name
- * of a python function, or class name of subclass of `ray.streaming.function.`
+ * of a
+ * python function, or class name of subclass of `ray.streaming.function.`
*/
- public PythonFunction(String moduleName,
- String functionName) {
+ public PythonFunction(
+ String moduleName,
+ String functionName) {
Preconditions.checkArgument(StringUtils.isNotBlank(moduleName));
Preconditions.checkArgument(StringUtils.isNotBlank(functionName));
this.function = null;
diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/PythonOperator.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/PythonOperator.java
index 81afb497c..185812f99 100644
--- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/PythonOperator.java
+++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/python/PythonOperator.java
@@ -17,6 +17,7 @@ import java.util.stream.Collectors;
*/
@SuppressWarnings("unchecked")
public class PythonOperator extends StreamOperator {
+
private final String moduleName;
private final String className;
@@ -80,7 +81,7 @@ public class PythonOperator extends StreamOperator {
StringBuilder builder = new StringBuilder();
builder.append(PythonOperator.class.getSimpleName()).append("[");
if (function != null) {
- builder.append(((PythonFunction)function).toSimpleString());
+ builder.append(((PythonFunction) function).toSimpleString());
} else {
builder.append(moduleName).append(".").append(className);
}
@@ -101,6 +102,7 @@ public class PythonOperator extends StreamOperator {
}
public static class ChainedPythonOperator extends PythonOperator {
+
private final List
- * Python worker can create a partition object using information in this
- * PythonPartition.
+ * Python worker can create a partition object using information in this PythonPartition.
*
- * If this object is constructed from serialized python partition,
- * python worker can deserialize it to create python partition directly.
- * If this object is constructed from moduleName and className/functionName,
- * python worker will use `importlib` to load python partition function.
+ * If this object is constructed from serialized python partition, python worker can deserialize it
+ * to create python partition directly. If this object is constructed from moduleName and
+ * className/functionName, python worker will use `importlib` to load python partition function.
*
*/
public class PythonPartition implements Partition, T>
implements Serializable {
+
private final int id;
private final StreamingContext streamingContext;
private final Stream inputStream;
@@ -34,9 +35,10 @@ public abstract class Stream, T>
this(streamingContext, null, streamOperator, getForwardPartition(streamOperator));
}
- public Stream(StreamingContext streamingContext,
- StreamOperator streamOperator,
- Partition, T>
this(inputStream.getStreamingContext(), inputStream, streamOperator, partition);
}
- protected Stream(StreamingContext streamingContext,
- Stream inputStream,
- StreamOperator streamOperator,
- Partition, T>
}
/**
- * Create a proxy stream of original stream.
- * Changes in new stream will be reflected in original stream and vice versa
+ * Create a proxy stream of original stream. Changes in new stream will be reflected in original
+ * stream and vice versa
*/
protected Stream(Stream originalStream) {
this.originalStream = originalStream;
@@ -183,8 +186,8 @@ public abstract class Stream, T>
}
/**
- * Set the partition function of this {@link Stream} so that output elements are forwarded to
- * next operator locally.
+ * Set the partition function of this {@link Stream} so that output elements are forwarded to next
+ * operator locally.
*/
public S forward() {
return setPartition(getForwardPartition(operator));
diff --git a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/StreamSink.java b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/StreamSink.java
index f03b1baa4..1dd9fcd4f 100644
--- a/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/StreamSink.java
+++ b/streaming/java/streaming-api/src/main/java/io/ray/streaming/api/stream/StreamSink.java
@@ -8,6 +8,7 @@ import io.ray.streaming.operator.StreamOperator;
* @param