You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/02/17 16:42:46 UTC
[8/8] flink git commit: [FLINK-5886][py] Various refactorings to
Streaming API
[FLINK-5886][py] Various refactorings to Streaming API
This closes #5333.
Changelog:
General:
- rebase branch to current master
- incremented version to 1.5-SNAPSHOT
- removed kafka code
- applied checkstyle
- disabled method/parameter name rules for API classes
- assigned flink-python-streaming to 'libraries' travis profile
- copy streaming-python jar to /opt
- change the name of the final jar to flink-streaming-python (previously flink-python)
- replace maven-jar-plugin with maven-shade-plugin
API:
- PDS#map()/flat_map() now return PythonSingleOutputStreamOperator
- renamed PDS#print() to PDS#output()
- print is a keyword in python and thus not usable in native python APIs
- added PythonSingleOutputStreamOperator#name()
- removed env#execute methods that accepted local execution argument as they are redundant due to environment factory methods
- narrow visibility of *DataStream constructors
Moved/Renamed:
- made SerializerMap top-level class and renamed it to AdapterMap
- Moved UtilityFunctions#adapt to AdapterMap class
- renamed UtilityFunctions to InterpreterUtils
- moved PythonobjectInputStream2 to SerializationUtils
- renamed PythonObjectInputStream2 to SerialVersionOverridingPythonObjectInputStream
Jython:
- renamed InterpreterUtils#smartFunctionDeserialization to deserializeFunction
- added generic return type to #deserializeFunction
- #deserializeFunction uses static initialization flag to detect whether it has to load jython instead of waiting for exception to happen
- removed file cleanup in #initAndExecPythonScript as it is the binders' responsibility
Connectors:
- replaced usage of deprecated serialiation schema interfaces
- P(S/D)Schema#(de)serialize now fails with RuntimeException if schema deserialization fails
Functions:
- Introduced AbstractPythonUDF class for sharing RichRunction#open()/close() implementations
- PythonOutputSelector now throws FlinkRuntimeException when failing during initialization
- added generic return type to Serializationutils#deserializeObject
- added new serializers for PyBoolean/-Float/-Integer/-Long/-String
- PyObjectSerializer not properly fails when an exceptioin occurs
- improved error printing
- PythonCollector now typed to Object and properly converts non-PyObjects
- jython functions that use a collector now have Object has output type
- otherwise you would get ClassCastException if jython returns something that isn't a PyObject
PythonStreamBinder
- adjusted to follow PythonPlanBinder structure
- client-like main() exception handling
- replaced Random usage with UUID.randomUIID()
- now loads GlobalConfiguration
- local/distributed tmp dir now configurable
- introduced PythonOptions
- no longer generate plan.py but instead import it directly via the PythonInterpreter
Environment:
- Reworked static environment factory methods from PythonStreamExecutionEnvironment into a PythonEnvironmentFactory
- program main() method now accepts a PythonEnvironmentFactory
- directories are now passed properly to the environment instead of using static fields
- removed PythonEnvironmentConfig
. #registerJythonSerializers now static
Examples:
- move examples to flink-streaming-python
- change examples location in dist to examples/python/streaming
- replace ParameterTool usage with argparse
- pass arguments via run instead of constructor
- remove 'if __name__ == '__main__':' block
- remove exception wrapping around source/sink creation
- add WordCount example
Tests:
- removed 'if __name__ == '__main__':' blocks from tests since the condition is never fulfilled
- removed python TestBase class
- removed print statements from tests
- standardized test job names
- cleaned up PythonStreamBinderTest / made it more consistent with PythonPlanBinderTest
- run_all_tests improvements
- stop after first failure
- print stacktrace on failure
- no longer relies on dirname() to get cwd but uses the module file location instead
- added log4j properties file
- added end-to-end test
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9b7e4292
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9b7e4292
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9b7e4292
Branch: refs/heads/master
Commit: 9b7e429246d149d02a899071b71e7ee9d321b9f4
Parents: 00284fb
Author: zentol <ch...@apache.org>
Authored: Mon Jan 15 13:03:42 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Sat Feb 17 16:07:20 2018 +0100
----------------------------------------------------------------------
docs/dev/stream/python.md | 37 +--
flink-dist/src/main/assemblies/bin.xml | 18 +-
flink-dist/src/main/assemblies/opt.xml | 8 +
.../src/main/flink-bin/bin/pyflink-stream.sh | 2 +-
.../src/main/python/fibonacci.py | 134 ---------
flink-libraries/flink-streaming-python/pom.xml | 62 ++--
.../flink/streaming/python/PythonOptions.java | 49 ++++
.../python/api/PythonStreamBinder.java | 210 ++++++-------
.../python/api/datastream/PythonDataStream.java | 95 +++---
.../api/datastream/PythonIterativeStream.java | 16 +-
.../api/datastream/PythonKeyedStream.java | 18 +-
.../datastream/PythonObjectInputStream2.java | 68 -----
.../PythonSingleOutputStreamOperator.java | 17 +-
.../api/datastream/PythonSplitStream.java | 12 +-
.../api/datastream/PythonWindowedStream.java | 6 +-
.../python/api/environment/PythonConstants.java | 26 ++
.../environment/PythonEnvironmentConfig.java | 39 ---
.../environment/PythonEnvironmentFactory.java | 141 +++++++++
.../PythonStreamExecutionEnvironment.java | 294 ++++++-------------
.../python/api/functions/AbstractPythonUDF.java | 88 ++++++
.../streaming/python/api/functions/PyKey.java | 3 +-
.../api/functions/PythonApplyFunction.java | 38 +--
.../api/functions/PythonFilterFunction.java | 40 +--
.../api/functions/PythonFlatMapFunction.java | 38 +--
.../api/functions/PythonGeneratorFunction.java | 35 +--
.../api/functions/PythonIteratorFunction.java | 42 ++-
.../python/api/functions/PythonKeySelector.java | 17 +-
.../python/api/functions/PythonMapFunction.java | 39 +--
.../api/functions/PythonOutputSelector.java | 23 +-
.../api/functions/PythonReduceFunction.java | 20 +-
.../api/functions/PythonSinkFunction.java | 41 +--
.../python/api/functions/UtilityFunctions.java | 143 ---------
.../connectors/PythonFlinkKafkaConsumer09.java | 38 ---
.../connectors/PythonFlinkKafkaProducer09.java | 44 ---
.../flink/streaming/python/util/AdapterMap.java | 53 ++++
.../streaming/python/util/InterpreterUtils.java | 136 +++++++++
.../streaming/python/util/PythonCollector.java | 9 +-
.../streaming/python/util/PythonIterator.java | 1 +
.../util/serialization/PyBooleanSerializer.java | 40 +++
.../util/serialization/PyFloatSerializer.java | 40 +++
.../util/serialization/PyIntegerSerializer.java | 40 +++
.../util/serialization/PyLongSerializer.java | 45 +++
.../util/serialization/PyObjectSerializer.java | 23 +-
.../util/serialization/PyStringSerializer.java | 40 +++
.../PythonDeserializationSchema.java | 62 ----
.../PythonSerializationSchema.java | 18 +-
.../util/serialization/SerializationUtils.java | 54 +++-
.../python/api/PythonStreamBinderTest.java | 105 +++----
.../streaming/python/api/examples/fibonacci.py | 122 ++++++++
.../streaming/python/api/examples/word_count.py | 122 ++++++++
.../flink/streaming/python/api/run_all_tests.py | 49 +---
.../flink/streaming/python/api/test_filter.py | 22 +-
.../streaming/python/api/test_flatmap_int.py | 26 +-
.../python/api/test_flatmap_list_int.py | 24 +-
.../streaming/python/api/test_flatmap_string.py | 24 +-
.../python/api/test_from_collection.py | 23 +-
.../streaming/python/api/test_from_elements.py | 23 +-
.../streaming/python/api/test_from_iterator.py | 23 +-
.../python/api/test_generate_sequence.py | 23 +-
.../streaming/python/api/test_iterations.py | 23 +-
.../flink/streaming/python/api/test_kafka09.py | 157 ----------
.../python/api/test_keyed_stream_reduce.py | 25 +-
.../flink/streaming/python/api/test_map.py | 25 +-
.../flink/streaming/python/api/test_map_int.py | 24 +-
.../streaming/python/api/test_read_text_file.py | 24 +-
.../python/api/test_socket_text_stream.py | 27 +-
.../streaming/python/api/test_split_select.py | 25 +-
.../flink/streaming/python/api/test_union.py | 23 +-
.../streaming/python/api/test_user_type.py | 24 +-
.../streaming/python/api/test_window_apply.py | 24 +-
.../streaming/python/api/test_word_count.py | 25 +-
.../streaming/python/api/test_write_as_text.py | 21 +-
.../python/api/test_write_to_socket.py | 22 +-
.../python/api/utils/python_test_base.py | 35 ---
.../src/test/resources/log4j-test.properties | 27 ++
.../test_streaming_python_wordcount.sh | 25 ++
tools/maven/suppressions.xml | 8 +
tools/travis_mvn_watchdog.sh | 11 +-
78 files changed, 1758 insertions(+), 1865 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/docs/dev/stream/python.md
----------------------------------------------------------------------
diff --git a/docs/dev/stream/python.md b/docs/dev/stream/python.md
index 02edbfa..a533819 100644
--- a/docs/dev/stream/python.md
+++ b/docs/dev/stream/python.md
@@ -26,7 +26,7 @@ under the License.
Analysis streaming programs in Flink are regular programs that implement transformations on
streaming data sets (e.g., filtering, mapping, joining, grouping). The streaming data sets are initially
-created from certain sources (e.g., by reading from Apache Kafka, or reading files, or from collections).
+created from certain sources (e.g., by reading files, or from collections).
Results are returned via sinks, which may for example write the data to (distributed) files, or to
standard output (for example the command line terminal). Flink streaming programs run in a variety
of contexts, standalone, or embedded in other programs. The execution can happen in a local JVM, or
@@ -62,7 +62,6 @@ in a stream of sentences, on a window size of 50 milliseconds and prints the res
from org.apache.flink.streaming.api.functions.source import SourceFunction
from org.apache.flink.api.common.functions import FlatMapFunction, ReduceFunction
from org.apache.flink.api.java.functions import KeySelector
-from org.apache.flink.python.api.jython import PythonStreamExecutionEnvironment
from org.apache.flink.streaming.api.windowing.time.Time import milliseconds
@@ -98,8 +97,8 @@ class Sum(ReduceFunction):
count2, word2 = input2
return (count1 + count2, word1)
-def main():
- env = PythonStreamExecutionEnvironment.get_execution_environment()
+def main(factory):
+ env = factory.get_execution_environment()
env.create_python_source(Generator(num_iters=1000)) \
.flat_map(Tokenizer()) \
.key_by(Selector()) \
@@ -108,15 +107,10 @@ def main():
.print()
env.execute()
-
-if __name__ == '__main__':
- main()
{% endhighlight %}
**Notes:**
-- If execution is done on a local cluster, you may replace the last line in the `main()` function
- with **`env.execute(True)`**
- Execution on a multi-node cluster requires a shared medium storage, which needs to be configured (.e.g HDFS)
upfront.
- The output from of the given script is directed to the standard output. Consequently, the output
@@ -130,8 +124,8 @@ Program Skeleton
As we already saw in the example, Flink streaming programs look like regular Python programs.
Each program consists of the same basic parts:
-1. A `main()` function definition, without arguments - the program entry point,
-2. Obtain an `Environment`,
+1. A `main(factory)` function definition, with an environment factory argument - the program entry point,
+2. Obtain an `Environment` from the factory,
3. Load/create the initial data,
4. Specify transformations on this data,
5. Specify where to put the results of your computations, and
@@ -140,21 +134,21 @@ Each program consists of the same basic parts:
We will now give an overview of each of those steps but please refer to the respective sections for
more details.
-The `main()` function is a must and it is used by Flink execution layer to run the
+The `main(factory)` function is a must and it is used by Flink execution layer to run the
given Python streaming program.
The `Environment` is the basis for all Flink programs. You can
-obtain one using these static methods on class `PythonStreamExecutionEnvironment`:
+obtain one using the factory methods provided by the factory:
{% highlight python %}
-get_execution_environment()
+factory.get_execution_environment()
{% endhighlight %}
For specifying data sources the streaming execution environment has several methods.
To just read a text file as a sequence of lines, you can use:
{% highlight python %}
-env = get_execution_environment()
+env = factory.get_execution_environment()
text = env.read_text_file("file:///path/to/file")
{% endhighlight %}
@@ -186,7 +180,7 @@ of these methods on DataStream:
{% highlight python %}
data.write_as_text("<file-path>")
data.write_as_text("<file-path>", mode=WriteMode.OVERWRITE)
-data.print()
+data.output()
{% endhighlight %}
The last method is only useful for developing/debugging on a local machine,
@@ -198,8 +192,7 @@ Please refer to [Data Sinks](#data-sinks) for more information on writing to fil
Once you specified the complete program you need to call `execute` on
the `Environment`. This will either execute on your local machine or submit your program
-for execution on a cluster, depending on how Flink was started. You can force
-a local execution by using `execute(True)`.
+for execution on a cluster, depending on how Flink was started.
{% top %}
@@ -209,7 +202,7 @@ Project setup
Apart from setting up Flink, no additional work is required. Using Jython to execute the Python
script, means that no external packages are needed and the program is executed as if it was a jar file.
-The Python API was tested on Linux/OSX systems.
+The Python API was tested on Windows/Linux/OSX systems.
{% top %}
@@ -540,7 +533,7 @@ Collection-based:
**Examples**
{% highlight python %}
-env = PythonStreamExecutionEnvironment.get_execution_environment()
+env = factory.get_execution_environment()
\# read text file from local files system
localLiens = env.read_text("file:///path/to/my/textfile")
@@ -564,7 +557,7 @@ Data sinks consume DataStreams and are used to store or return them:
- `write_as_text()` - Writes elements line-wise as Strings. The Strings are
obtained by calling the *str()* method of each element.
-- `print()` - Prints the *str()* value of each element on the
+- `output()` - Prints the *str()* value of each element on the
standard out.
- `write_to_socket()` - Writes the DataStream to a socket [host:port] as a byte array.
@@ -615,7 +608,7 @@ The default parallelism of an execution environment can be specified by calling
execution environment as follows:
{% highlight python %}
-env = PythonStreamExecutionEnvironment.get_execution_environment()
+env = factory.get_execution_environment()
env.set_parallelism(3)
text.flat_map(Tokenizer()) \
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-dist/src/main/assemblies/bin.xml
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/assemblies/bin.xml b/flink-dist/src/main/assemblies/bin.xml
index 059d7be..89228b3 100644
--- a/flink-dist/src/main/assemblies/bin.xml
+++ b/flink-dist/src/main/assemblies/bin.xml
@@ -28,7 +28,6 @@ under the License.
<includeBaseDirectory>true</includeBaseDirectory>
<baseDirectory>flink-${project.version}</baseDirectory>
- <!-- Include flink-python.jar & flink-streaming-python.jar in lib/ -->
<dependencySets>
<dependencySet>
<outputDirectory>lib</outputDirectory>
@@ -208,26 +207,17 @@ under the License.
</includes>
</fileSet>
- <!-- copy python jar -->
- <fileSet>
- <directory>../flink-libraries/flink-streaming-python/target</directory>
- <outputDirectory>lib</outputDirectory>
- <fileMode>0644</fileMode>
- <includes>
- <include>flink-python_${scala.binary.version}-${project.version}.jar</include>
- </includes>
- </fileSet>
-
<!-- copy python example to examples of dist -->
<fileSet>
<directory>../flink-libraries/flink-python/src/main/python/org/apache/flink/python/api/flink/example</directory>
- <outputDirectory>examples/python</outputDirectory>
+ <outputDirectory>examples/python/batch</outputDirectory>
<fileMode>0755</fileMode>
</fileSet>
+
<!-- copy python streaming example to examples of dist -->
<fileSet>
- <directory>../flink-examples/flink-examples-streaming/src/main/python</directory>
- <outputDirectory>examples/streaming/python</outputDirectory>
+ <directory>../flink-libraries/flink-streaming-python/src/test/python/org/apache/flink/streaming/python/api/examples</directory>
+ <outputDirectory>examples/python/streaming</outputDirectory>
<fileMode>0755</fileMode>
</fileSet>
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-dist/src/main/assemblies/opt.xml
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/assemblies/opt.xml b/flink-dist/src/main/assemblies/opt.xml
index e5075ef..450f20a 100644
--- a/flink-dist/src/main/assemblies/opt.xml
+++ b/flink-dist/src/main/assemblies/opt.xml
@@ -176,5 +176,13 @@
<destName>flink-json-${project.version}.jar</destName>
<fileMode>0644</fileMode>
</file>
+
+ <!-- Streaming Python API -->
+ <file>
+ <source>../flink-libraries/flink-streaming-python/target/flink-streaming-python_${scala.binary.version}-${project.version}.jar</source>
+ <outputDirectory>opt</outputDirectory>
+ <destName>flink-streaming-python_${scala.binary.version}-${project.version}.jar</destName>
+ <fileMode>0644</fileMode>
+ </file>
</files>
</assembly>
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh b/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh
index c31f702..3e44a10 100644
--- a/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh
+++ b/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh
@@ -22,4 +22,4 @@ bin=`cd "$bin"; pwd`
. "$bin"/config.sh
-"$FLINK_BIN_DIR"/flink run --class org.apache.flink.streaming.python.api.PythonStreamBinder -v "$FLINK_ROOT_DIR"/lib/flink-streaming-python*.jar "$@"
+"$FLINK_BIN_DIR"/flink run --class org.apache.flink.streaming.python.api.PythonStreamBinder -v "$FLINK_ROOT_DIR"/opt/flink-streaming-python*.jar "$@"
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-examples/flink-examples-streaming/src/main/python/fibonacci.py
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/python/fibonacci.py b/flink-examples/flink-examples-streaming/src/main/python/fibonacci.py
deleted file mode 100644
index 96f90c8..0000000
--- a/flink-examples/flink-examples-streaming/src/main/python/fibonacci.py
+++ /dev/null
@@ -1,134 +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.
-################################################################################
-
-#
-# An example that illustrates iterations in Flink streaming. The program sums up random numbers and counts
-# additions. The operation is done until it reaches a specific threshold in an iterative streaming fashion.
-#
-
-import random
-import sys
-from org.apache.flink.api.common.functions import MapFunction
-from org.apache.flink.streaming.api.collector.selector import OutputSelector
-from org.apache.flink.streaming.api.functions.source import SourceFunction
-from org.apache.flink.api.java.utils import ParameterTool
-from org.apache.flink.streaming.python.api.environment import PythonStreamExecutionEnvironment
-
-TARGET_VAL = 100
-MAX_INT_START = 50
-
-
-class Generator(SourceFunction):
- def __init__(self, num_iters=1000):
- self._running = True
- self._num_iters = num_iters
-
- def run(self, ctx):
- counter = 0
- while self._running and counter < self._num_iters:
- self.do(ctx)
- counter += 1
-
- def do(self, ctx):
- two_numbers = "{}, {}".format(random.randrange(1, MAX_INT_START), random.randrange(1, MAX_INT_START))
- ctx.collect(two_numbers)
-
- def cancel(self):
- self._running = False
-
-
-class Fib(MapFunction):
- def map(self, value):
- return (value[0], value[1], value[3], value[2] + value[3], value[4] + 1)
-
-
-class InPut(MapFunction):
- def map(self, value):
- num1, num2 = value.split(",")
-
- num1 = int(num1.strip())
- num2 = int(num2.strip())
-
- return (num1, num2, num1, num2, 0)
-
-
-class OutPut(MapFunction):
- def map(self, value):
- return ((value[0], value[1]), value[4])
-
-
-class StreamSelector(OutputSelector):
- def select(self, value):
- return ["iterate"] if value[3] < TARGET_VAL else ["output"]
-
-
-class Main:
- def __init__(self, args):
- self._args = args
-
- def run(self):
- _params = ParameterTool.fromArgs(self._args)
-
- env = PythonStreamExecutionEnvironment.get_execution_environment()
-
- # create input stream of integer pairs
- if "--input" in self._args:
- try:
- file_path = _params.get("input")
- input_stream = env.read_text_file(file_path)
- except Exception as e:
- print(e)
- print ("Error in reading input file. Exiting...")
- sys.exit(5)
- else:
- input_stream = env.create_python_source(Generator(num_iters=50))
-
- # create an iterative data stream from the input with 5 second timeout
- it = input_stream.map(InPut()).iterate(5000)
-
- # apply the step function to get the next Fibonacci number
- # increment the counter and split the output with the output selector
- step = it.map(Fib()).split(StreamSelector())
- it.close_with(step.select("iterate"))
-
- # to produce the final output select the tuples directed to the
- # 'output' channel then get the input pairs that have the greatest iteration counter
- # on a 1 second sliding window
- output = step.select("output")
- parsed_output = output.map(OutPut())
- if "--output" in self._args:
- try:
- file_path = _params.get("output")
- parsed_output.write_as_text(file_path)
- except Execption as e:
- print (e)
- print ("Error in writing to output file. will print instead.")
- parsed_output.print()
- else:
- parsed_output.print()
- result = env.execute("Fibonacci Example (py)", True if _params.has("local") else False)
- print("Fibonacci job completed, job_id={}".format(result.jobID))
-
-
-def main():
- argv = sys.argv[1:] if len(sys.argv) > 1 else []
- Main(argv).run()
-
-
-if __name__ == '__main__':
- main()
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/pom.xml
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/pom.xml b/flink-libraries/flink-streaming-python/pom.xml
index 68e205a..a3ae8b7 100644
--- a/flink-libraries/flink-streaming-python/pom.xml
+++ b/flink-libraries/flink-streaming-python/pom.xml
@@ -23,7 +23,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-libraries</artifactId>
- <version>1.4-SNAPSHOT</version>
+ <version>1.5-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
@@ -31,30 +31,8 @@ under the License.
<name>flink-streaming-python</name>
<packaging>jar</packaging>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <descriptorRefs>
- <descriptorRef>jar-with-dependencies</descriptorRef>
- </descriptorRefs>
- <archive>
- <manifest>
- <addClasspath>true</addClasspath>
- <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
- </manifest>
- </archive>
- </configuration>
- </plugin>
- </plugins>
- </build>
-
<dependencies>
-
<!-- core dependencies -->
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
@@ -79,20 +57,14 @@ under the License.
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+
<dependency>
<groupId>org.python</groupId>
<artifactId>jython-standalone</artifactId>
<version>2.7.0</version>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
- <version>1.2.0</version>
- <scope>provided</scope>
- </dependency>
<!-- test dependencies -->
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils_${scala.binary.version}</artifactId>
@@ -101,4 +73,34 @@ under the License.
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <includes>
+ <include>org.python:jython-standalone</include>
+ </includes>
+ </artifactSet>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.flink.streaming.python.api.PythonStreamBinder</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/PythonOptions.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/PythonOptions.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/PythonOptions.java
new file mode 100644
index 0000000..e3b2f9a
--- /dev/null
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/PythonOptions.java
@@ -0,0 +1,49 @@
+/*
+ * 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.python;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.io.File;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * Configuration options for the Python API.
+ */
+public class PythonOptions {
+
+ /**
+ * The config parameter defining where temporary plan-related files are stored on the client.
+ */
+ public static final ConfigOption<String> PLAN_TMP_DIR =
+ key("python.plan.tmp.dir")
+ .noDefaultValue();
+
+ /**
+ * The config parameter defining where the flink python library and user supplied files will be uploaded to before
+ * registering them with the Distributed Cache. This directory must be accessible from all worker nodes.
+ */
+ public static final ConfigOption<String> DC_TMP_DIR =
+ key("python.dc.tmp.dir")
+ .defaultValue(System.getProperty("java.io.tmpdir") + File.separator + "flink_dc");
+
+ private PythonOptions() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/PythonStreamBinder.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/PythonStreamBinder.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/PythonStreamBinder.java
index ad5f432..97b1030 100644
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/PythonStreamBinder.java
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/PythonStreamBinder.java
@@ -15,182 +15,138 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.python.api;
-import org.apache.commons.io.FilenameUtils;
+
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.filecache.FileCache;
-import org.apache.flink.streaming.python.api.environment.PythonEnvironmentConfig;
-import org.apache.flink.streaming.python.api.functions.UtilityFunctions;
+import org.apache.flink.streaming.python.PythonOptions;
+import org.apache.flink.streaming.python.api.environment.PythonEnvironmentFactory;
+import org.apache.flink.streaming.python.util.InterpreterUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.FileWriter;
-import java.io.PrintWriter;
import java.io.IOException;
-
-import java.net.URISyntaxException;
-import java.nio.file.Files;
import java.nio.file.Paths;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
-import java.util.Random;
+import java.util.UUID;
/**
- * Allows the execution of Flink stream plan that is written in Python
+ * Allows the execution of Flink stream plan that is written in Python.
*/
public class PythonStreamBinder {
- private static final Random r = new Random(System.currentTimeMillis());
- public static final String FLINK_PYTHON_FILE_PATH = System.getProperty("java.io.tmpdir") + File.separator + "flink_streaming_plan_";
+ private static final Logger LOG = LoggerFactory.getLogger(PythonStreamBinder.class);
+ private final String localTmpPath;
+ private Path tmpDistributedDir;
- private PythonStreamBinder() {
+ PythonStreamBinder(Configuration globalConfig) {
+ String configuredLocalTmpPath = globalConfig.getString(PythonOptions.PLAN_TMP_DIR);
+ this.localTmpPath = configuredLocalTmpPath != null
+ ? configuredLocalTmpPath
+ : System.getProperty("java.io.tmpdir") + File.separator + "flink_streaming_plan_" + UUID.randomUUID();
+
+ this.tmpDistributedDir = new Path(globalConfig.getString(PythonOptions.DC_TMP_DIR));
}
/**
* Entry point for the execution of a python streaming task.
*
- * @param args <pathToScript> [<pathToPackage1> .. [<pathToPackageX]] - [parameter1]..[parameterX]
+ * @param args pathToScript [pathToPackage1 .. [pathToPackageX]] - [parameter1]..[parameterX]
* @throws Exception
*/
public static void main(String[] args) throws Exception {
- PythonStreamBinder binder = new PythonStreamBinder();
- binder.runPlan(args);
+ Configuration globalConfig = GlobalConfiguration.loadConfiguration();
+ PythonStreamBinder binder = new PythonStreamBinder(globalConfig);
+ try {
+ binder.runPlan(args);
+ } catch (Exception e) {
+ System.out.println("Failed to run plan: " + e.getMessage());
+ e.printStackTrace();
+ LOG.error("Failed to run plan.", e);
+ }
}
- private void runPlan(String[] args) throws Exception {
- File script;
+ void runPlan(String[] args) throws Exception {
if (args.length < 1) {
- System.out.println("Usage: prog <pathToScript> [<pathToPackage1> .. [<pathToPackageX]] - [parameter1]..[parameterX]");
+ System.out.println("Usage: prog <pathToScript> [parameter1]..[parameterX] - [<pathToPackage1> .. [<pathToPackageX]]");
return;
}
- else {
- script = new File(args[0]);
- if ((!script.exists()) || (!script.isFile()))
- {
- throw new FileNotFoundException("Could not find file: " + args[0]);
- }
- }
int split = 0;
for (int x = 0; x < args.length; x++) {
- if (args[x].compareTo("-") == 0) {
+ if (args[x].equals("-")) {
split = x;
+ break;
}
}
- GlobalConfiguration.loadConfiguration();
-
- String tmpPath = FLINK_PYTHON_FILE_PATH + r.nextLong();
- prepareFiles(tmpPath, Arrays.copyOfRange(args, 0, split == 0 ? args.length : split));
-
- if (split != 0) {
- String[] a = new String[args.length - split];
- a[0] = args[0];
- System.arraycopy(args, split + 1, a, 1, args.length - (split +1));
- args = a;
- } else if (args.length > 1) {
- args = new String[]{args[0]};
- }
-
- UtilityFunctions.initAndExecPythonScript(new File(
- tmpPath + File.separator + PythonEnvironmentConfig.FLINK_PYTHON_PLAN_NAME), args);
- }
-
- /**
- * Prepares a single package from the given python script. This involves in generating a unique main
- * python script, which loads the user's provided script for execution. The reason for this is to create
- * a different namespace for different script contents. In addition, it copies all relevant files
- * to a temporary folder for execution and distribution.
- *
- * @param tempFilePath The path to copy all the files to
- * @param filePaths The user's script and dependent packages/files
- * @throws IOException
- * @throws URISyntaxException
- */
- private void prepareFiles(String tempFilePath, String... filePaths) throws IOException, URISyntaxException,
- NoSuchAlgorithmException {
- PythonEnvironmentConfig.pythonTmpCachePath = tempFilePath;
-
- Files.createDirectories(Paths.get(tempFilePath));
+ try {
+ String planFile = args[0];
+ String[] filesToCopy = Arrays.copyOfRange(args, 1, split == 0 ? args.length : split);
+ String[] planArgumentsArray = Arrays.copyOfRange(args, split == 0 ? args.length : split + 1, args.length);
- String dstMainScriptFullPath = tempFilePath + File.separator + FilenameUtils.getBaseName(filePaths[0]) +
- calcPythonMd5(filePaths[0]) + ".py";
+ // verify existence of files
+ Path planPath = new Path(planFile);
+ if (!FileSystem.getUnguardedFileSystem(planPath.toUri()).exists(planPath)) {
+ throw new FileNotFoundException("Plan file " + planFile + " does not exist.");
+ }
+ for (String file : filesToCopy) {
+ Path filePath = new Path(file);
+ if (!FileSystem.getUnguardedFileSystem(filePath.toUri()).exists(filePath)) {
+ throw new FileNotFoundException("Additional file " + file + " does not exist.");
+ }
+ }
- generateAndCopyPlanFile(tempFilePath, dstMainScriptFullPath);
+ // setup temporary local directory for flink python library and user files
+ Path targetDir = new Path(localTmpPath);
+ deleteIfExists(targetDir);
+ targetDir.getFileSystem().mkdirs(targetDir);
+
+ // copy user files to temporary location
+ Path tmpPlanFilesPath = new Path(localTmpPath);
+ copyFile(planPath, tmpPlanFilesPath, planPath.getName());
+ for (String file : filesToCopy) {
+ Path source = new Path(file);
+ copyFile(source, tmpPlanFilesPath, source.getName());
+ }
- //main script
- copyFile(filePaths[0], tempFilePath, FilenameUtils.getName(dstMainScriptFullPath));
+ String planNameWithExtension = planPath.getName();
+ String planName = planNameWithExtension.substring(0, planNameWithExtension.indexOf(".py"));
- String parentDir = new File(filePaths[0]).getParent();
+ InterpreterUtils.initAndExecPythonScript(new PythonEnvironmentFactory(localTmpPath, tmpDistributedDir, planName), Paths.get(localTmpPath), planName, planArgumentsArray);
+ } finally {
+ try {
+ // clean up created files
+ FileSystem distributedFS = tmpDistributedDir.getFileSystem();
+ distributedFS.delete(tmpDistributedDir, true);
- //additional files/folders(modules)
- for (int x = 1; x < filePaths.length; x++) {
- String currentParent = (new File(filePaths[x])).getParent();
- if (currentParent.startsWith(".")) {
- filePaths[x] = parentDir + File.separator + filePaths[x];
+ FileSystem local = FileSystem.getLocalFileSystem();
+ local.delete(new Path(localTmpPath), true);
+ } catch (IOException ioe) {
+ LOG.error("PythonAPI file cleanup failed. {}", ioe.getMessage());
}
- copyFile(filePaths[x], tempFilePath, null);
}
}
- private void generateAndCopyPlanFile(String dstPath, String mainScriptFullPath) throws IOException {
- String moduleName = FilenameUtils.getBaseName(mainScriptFullPath);
-
- FileWriter fileWriter = new FileWriter(dstPath + File.separator + PythonEnvironmentConfig.FLINK_PYTHON_PLAN_NAME);
- PrintWriter printWriter = new PrintWriter(fileWriter);
- printWriter.printf("import %s\n\n", moduleName);
- printWriter.printf("%s.main()\n", moduleName);
- printWriter.close();
- }
+ //=====File utils===================================================================================================
- private void copyFile(String path, String target, String name) throws IOException, URISyntaxException {
- if (path.endsWith(File.separator)) {
- path = path.substring(0, path.length() - 1);
+ private static void deleteIfExists(Path path) throws IOException {
+ FileSystem fs = path.getFileSystem();
+ if (fs.exists(path)) {
+ fs.delete(path, true);
}
- String identifier = name == null ? path.substring(path.lastIndexOf(File.separator)) : name;
- String tmpFilePath = target + File.separator + identifier;
- Path p = new Path(path);
- FileCache.copy(p.makeQualified(FileSystem.get(p.toUri())), new Path(tmpFilePath), true);
}
-
- /**
- * Naive MD5 calculation from the python content of the given script. Spaces, blank lines and comments
- * are ignored.
- *
- * @param filePath the full path of the given python script
- * @return the md5 value as a string
- * @throws NoSuchAlgorithmException
- * @throws IOException
- */
- private String calcPythonMd5(String filePath) throws NoSuchAlgorithmException, IOException {
- FileReader fileReader = new FileReader(filePath);
- BufferedReader bufferedReader = new BufferedReader(fileReader);
-
- String line;
- MessageDigest md = MessageDigest.getInstance("MD5");
- while ((line = bufferedReader.readLine()) != null) {
- line = line.trim();
- if (line.isEmpty() || line.startsWith("#")) {
- continue;
- }
- byte[] bytes = line.getBytes();
- md.update(bytes, 0, bytes.length);
- }
- fileReader.close();
-
- byte[] mdBytes = md.digest();
-
- //convert the byte to hex format method 1
- StringBuffer sb = new StringBuffer();
- for (int i = 0; i < mdBytes.length; i++) {
- sb.append(Integer.toString((mdBytes[i] & 0xff) + 0x100, 16).substring(1));
- }
- return sb.toString();
+ private static void copyFile(Path source, Path targetDirectory, String name) throws IOException {
+ Path targetFilePath = new Path(targetDirectory, name);
+ deleteIfExists(targetFilePath);
+ FileCache.copy(source, targetFilePath, true);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java
index 96feb0b..94c844e 100644
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonDataStream.java
@@ -15,12 +15,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.python.api.datastream;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
@@ -35,13 +37,12 @@ import org.apache.flink.streaming.python.api.functions.PythonMapFunction;
import org.apache.flink.streaming.python.api.functions.PythonOutputSelector;
import org.apache.flink.streaming.python.api.functions.PythonSinkFunction;
import org.apache.flink.streaming.python.util.serialization.PythonSerializationSchema;
-import org.apache.flink.streaming.util.serialization.SerializationSchema;
+
import org.python.core.PyObject;
import java.io.IOException;
import java.util.ArrayList;
-
/**
* A {@code PythonDataStream} is a thin wrapper layer over {@link DataStream}, which represents a
* stream of elements of the same type. A {@code PythonDataStream} can be transformed into
@@ -67,8 +68,7 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
/**
* A thin wrapper layer over {@link DataStream#union(DataStream[])}.
*
- * @param streams
- * The Python DataStreams to union output with.
+ * @param streams The Python DataStreams to union output with.
* @return The {@link PythonDataStream}.
*/
@SafeVarargs
@@ -85,8 +85,7 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
/**
* A thin wrapper layer over {@link DataStream#split(OutputSelector)}.
*
- * @param output_selector
- * The user defined {@link OutputSelector} for directing the tuples.
+ * @param output_selector The user defined {@link OutputSelector} for directing the tuples.
* @return The {@link PythonSplitStream}
*/
public PythonSplitStream split(OutputSelector<PyObject> output_selector) throws IOException {
@@ -96,8 +95,7 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
/**
* A thin wrapper layer over {@link DataStream#filter(FilterFunction)}.
*
- * @param filter
- * The FilterFunction that is called for each element of the DataStream.
+ * @param filter The FilterFunction that is called for each element of the DataStream.
* @return The filtered {@link PythonDataStream}.
*/
public PythonSingleOutputStreamOperator filter(FilterFunction<PyObject> filter) throws IOException {
@@ -107,35 +105,31 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
/**
* A thin wrapper layer over {@link DataStream#map(MapFunction)}.
*
- * @param mapper
- * The MapFunction that is called for each element of the
- * DataStream.
+ * @param mapper The MapFunction that is called for each element of the
+ * DataStream.
* @return The transformed {@link PythonDataStream}.
*/
public PythonDataStream<SingleOutputStreamOperator<PyObject>> map(
MapFunction<PyObject, PyObject> mapper) throws IOException {
- return new PythonDataStream<>(stream.map(new PythonMapFunction(mapper)));
+ return new PythonSingleOutputStreamOperator(stream.map(new PythonMapFunction(mapper)));
}
/**
* A thin wrapper layer over {@link DataStream#flatMap(FlatMapFunction)}.
*
- * @param flat_mapper
- * The FlatMapFunction that is called for each element of the
- * DataStream
- *
+ * @param flat_mapper The FlatMapFunction that is called for each element of the
+ * DataStream
* @return The transformed {@link PythonDataStream}.
*/
public PythonDataStream<SingleOutputStreamOperator<PyObject>> flat_map(
- FlatMapFunction<PyObject, PyObject> flat_mapper) throws IOException {
- return new PythonDataStream<>(stream.flatMap(new PythonFlatMapFunction(flat_mapper)));
+ FlatMapFunction<PyObject, Object> flat_mapper) throws IOException {
+ return new PythonSingleOutputStreamOperator(stream.flatMap(new PythonFlatMapFunction(flat_mapper)));
}
/**
* A thin wrapper layer over {@link DataStream#keyBy(KeySelector)}.
*
- * @param selector
- * The KeySelector to be used for extracting the key for partitioning
+ * @param selector The KeySelector to be used for extracting the key for partitioning
* @return The {@link PythonDataStream} with partitioned state (i.e. {@link PythonKeyedStream})
*/
public PythonKeyedStream key_by(KeySelector<PyObject, PyKey> selector) throws IOException {
@@ -146,15 +140,14 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
* A thin wrapper layer over {@link DataStream#print()}.
*/
@PublicEvolving
- public void print() {
+ public void output() {
stream.print();
}
/**
* A thin wrapper layer over {@link DataStream#writeAsText(java.lang.String)}.
*
- * @param path
- * The path pointing to the location the text file is written to.
+ * @param path The path pointing to the location the text file is written to.
*/
@PublicEvolving
public void write_as_text(String path) {
@@ -164,11 +157,9 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
/**
* A thin wrapper layer over {@link DataStream#writeAsText(java.lang.String, WriteMode)}.
*
- * @param path
- * The path pointing to the location the text file is written to
- * @param mode
- * Controls the behavior for existing files. Options are
- * NO_OVERWRITE and OVERWRITE.
+ * @param path The path pointing to the location the text file is written to
+ * @param mode Controls the behavior for existing files. Options are
+ * NO_OVERWRITE and OVERWRITE.
*/
@PublicEvolving
public void write_as_text(String path, WriteMode mode) {
@@ -176,14 +167,11 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
}
/**
- * A thin wrapper layer over {@link DataStream#writeToSocket(java.lang.String, int, SerializationSchema)}
+ * A thin wrapper layer over {@link DataStream#writeToSocket(String, int, org.apache.flink.api.common.serialization.SerializationSchema)}.
*
- * @param host
- * host of the socket
- * @param port
- * port of the socket
- * @param schema
- * schema for serialization
+ * @param host host of the socket
+ * @param port port of the socket
+ * @param schema schema for serialization
*/
@PublicEvolving
public void write_to_socket(String host, Integer port, SerializationSchema<PyObject> schema) throws IOException {
@@ -191,10 +179,9 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
}
/**
- * A thin wrapper layer over {@link DataStream#addSink(SinkFunction)}
+ * A thin wrapper layer over {@link DataStream#addSink(SinkFunction)}.
*
- * @param sink_func
- * The object containing the sink's invoke function.
+ * @param sink_func The object containing the sink's invoke function.
*/
@PublicEvolving
public void add_sink(SinkFunction<PyObject> sink_func) throws IOException {
@@ -211,16 +198,16 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
* given to the {@link PythonIterativeStream#close_with(PythonDataStream)} method is
* the data stream that will be fed back and used as the input for the
* iteration head. </p>
- * <p>
- * A common usage pattern for streaming iterations is to use output
+ *
+ * <p>A common usage pattern for streaming iterations is to use output
* splitting to send a part of the closing data stream to the head. Refer to
* {@link #split(OutputSelector)} for more information.
- * <p>
- * The iteration edge will be partitioned the same way as the first input of
+ *
+ * <p>The iteration edge will be partitioned the same way as the first input of
* the iteration head unless it is changed in the
* {@link PythonIterativeStream#close_with(PythonDataStream)} call.
- * <p>
- * By default a PythonDataStream with iteration will never terminate, but the user
+ *
+ * <p>By default a PythonDataStream with iteration will never terminate, but the user
* can use the maxWaitTime parameter to set a max waiting time for the
* iteration head. If no data received in the set time, the stream
* terminates.
@@ -235,31 +222,29 @@ public class PythonDataStream<D extends DataStream<PyObject>> {
/**
* A thin wrapper layer over {@link DataStream#iterate(long)}.
*
- * <p></p>Initiates an iterative part of the program that feeds back data streams.
+ * <p>Initiates an iterative part of the program that feeds back data streams.
* The iterative part needs to be closed by calling
* {@link PythonIterativeStream#close_with(PythonDataStream)}. The transformation of
* this IterativeStream will be the iteration head. The data stream
* given to the {@link PythonIterativeStream#close_with(PythonDataStream)} method is
* the data stream that will be fed back and used as the input for the
* iteration head.</p>
- * <p>
- * A common usage pattern for streaming iterations is to use output
+ *
+ * <p>A common usage pattern for streaming iterations is to use output
* splitting to send a part of the closing data stream to the head. Refer to
* {@link #split(OutputSelector)} for more information.
- * <p>
- * The iteration edge will be partitioned the same way as the first input of
+ *
+ * <p>The iteration edge will be partitioned the same way as the first input of
* the iteration head unless it is changed in the
* {@link PythonIterativeStream#close_with(PythonDataStream)} call.
- * <p>
- * By default a PythonDataStream with iteration will never terminate, but the user
+ *
+ * <p>By default a PythonDataStream with iteration will never terminate, but the user
* can use the maxWaitTime parameter to set a max waiting time for the
* iteration head. If no data received in the set time, the stream
* terminates.
*
- * @param max_wait_time_ms
- * Number of milliseconds to wait between inputs before shutting
- * down
- *
+ * @param max_wait_time_ms Number of milliseconds to wait between inputs before shutting
+ * down
* @return The iterative data stream created.
*/
@PublicEvolving
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonIterativeStream.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonIterativeStream.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonIterativeStream.java
index ced1c24..235b607 100644
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonIterativeStream.java
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonIterativeStream.java
@@ -15,11 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.python.api.datastream;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
+
import org.python.core.PyObject;
/**
@@ -31,27 +33,23 @@ import org.python.core.PyObject;
@PublicEvolving
public class PythonIterativeStream extends PythonSingleOutputStreamOperator {
- public PythonIterativeStream(IterativeStream<PyObject> iterativeStream) {
+ PythonIterativeStream(IterativeStream<PyObject> iterativeStream) {
super(iterativeStream);
}
/**
* A thin wrapper layer over {@link IterativeStream#closeWith(org.apache.flink.streaming.api.datastream.DataStream)}
*
- * Please note that this function works with {@link PythonDataStream} and thus wherever a DataStream is mentioned in
+ * <p>Please note that this function works with {@link PythonDataStream} and thus wherever a DataStream is mentioned in
* the above {@link IterativeStream#closeWith(org.apache.flink.streaming.api.datastream.DataStream)} description,
* the user may regard it as {@link PythonDataStream} .
*
- *
- * @param feedback_stream
- * {@link PythonDataStream} that will be used as input to the iteration
- * head.
- *
+ * @param feedback_stream {@link PythonDataStream} that will be used as input to the iteration
+ * head.
* @return The feedback stream.
- *
*/
public PythonDataStream close_with(PythonDataStream<? extends DataStream<PyObject>> feedback_stream) {
- ((IterativeStream<PyObject>)this.stream).closeWith(feedback_stream.stream);
+ ((IterativeStream<PyObject>) this.stream).closeWith(feedback_stream.stream);
return feedback_stream;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonKeyedStream.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonKeyedStream.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonKeyedStream.java
index f259374..c78f5dd 100644
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonKeyedStream.java
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonKeyedStream.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.python.api.datastream;
import org.apache.flink.annotation.Public;
@@ -25,21 +26,21 @@ import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.python.api.functions.PyKey;
import org.apache.flink.streaming.python.api.functions.PythonReduceFunction;
+
import org.python.core.PyObject;
import java.io.IOException;
-
/**
* A thin wrapper layer over {@link KeyedStream}.
*
* <p>A {@code PythonKeyedStream} represents a {@link PythonDataStream} on which operator state is
- * partitioned by key using a provided {@link org.apache.flink.api.java.functions.KeySelector;}</p>
+ * partitioned by key using a provided {@link org.apache.flink.api.java.functions.KeySelector}.
*/
@Public
public class PythonKeyedStream extends PythonDataStream<KeyedStream<PyObject, PyKey>> {
- public PythonKeyedStream(KeyedStream<PyObject, PyKey> stream) {
+ PythonKeyedStream(KeyedStream<PyObject, PyKey> stream) {
super(stream);
}
@@ -55,7 +56,7 @@ public class PythonKeyedStream extends PythonDataStream<KeyedStream<PyObject, Py
}
/**
- * A thin wrapper layer over {@link KeyedStream#timeWindow(Time)}
+ * A thin wrapper layer over {@link KeyedStream#timeWindow(Time)}.
*
* @param size The size of the window.
* @return The python windowed stream {@link PythonWindowedStream}
@@ -65,7 +66,7 @@ public class PythonKeyedStream extends PythonDataStream<KeyedStream<PyObject, Py
}
/**
- * A thin wrapper layer over {@link KeyedStream#timeWindow(Time, Time)}
+ * A thin wrapper layer over {@link KeyedStream#timeWindow(Time, Time)}.
*
* @param size The size of the window.
* @return The python wrapper {@link PythonWindowedStream}
@@ -75,11 +76,10 @@ public class PythonKeyedStream extends PythonDataStream<KeyedStream<PyObject, Py
}
/**
- * A thin wrapper layer over {@link KeyedStream#reduce(ReduceFunction)}
+ * A thin wrapper layer over {@link KeyedStream#reduce(ReduceFunction)}.
*
- * @param reducer
- * The {@link ReduceFunction} that will be called for every
- * element of the input values with the same key.
+ * @param reducer The {@link ReduceFunction} that will be called for every
+ * element of the input values with the same key.
* @return The transformed data stream @{link PythonSingleOutputStreamOperator}.
*/
public PythonSingleOutputStreamOperator reduce(ReduceFunction<PyObject> reducer) throws IOException {
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java
deleted file mode 100644
index f701d98..0000000
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonObjectInputStream2.java
+++ /dev/null
@@ -1,68 +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.python.api.datastream;
-
-import org.python.util.PythonObjectInputStream;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.ObjectStreamClass;
-
-/**
- * A helper class to overcome the inability to set the serialVersionUID in a python user-defined
- * function (UDF).
- * <p>The fact the this field is not set, results in a dynamic calculation of this serialVersionUID,
- * using SHA, to make sure it is a unique number. This unique number is a 64-bit hash of the
- * class name, interface class names, methods, and fields. If a Python class inherits from a
- * Java class, as in the case of Python UDFs, then a proxy wrapper class is created. Its name is
- * constructed using the following pattern:
- * <b>{@code org.python.proxies.<module-name>$<UDF-class-name>$<number>}</b>. The {@code <number>}
- * part is increased by one in runtime, for every job submission. It results in different serial
- * version UID for each run for the same Python class. Therefore, it is required to silently
- * suppress the serial version UID mismatch check.</p>
- */
-public class PythonObjectInputStream2 extends PythonObjectInputStream {
-
- public PythonObjectInputStream2(InputStream in) throws IOException {
- super(in);
- }
-
- protected ObjectStreamClass readClassDescriptor() throws IOException, ClassNotFoundException {
- ObjectStreamClass resultClassDescriptor = super.readClassDescriptor(); // initially streams descriptor
-
- Class<?> localClass;
- try {
- localClass = resolveClass(resultClassDescriptor);
- } catch (ClassNotFoundException e) {
- System.out.println("No local class for " + resultClassDescriptor.getName());
- return resultClassDescriptor;
- }
-
- ObjectStreamClass localClassDescriptor = ObjectStreamClass.lookup(localClass);
- if (localClassDescriptor != null) { // only if class implements serializable
- final long localSUID = localClassDescriptor.getSerialVersionUID();
- final long streamSUID = resultClassDescriptor.getSerialVersionUID();
- if (streamSUID != localSUID) { // check for serialVersionUID mismatch.
- // Overriding serialized class version mismatch
- resultClassDescriptor = localClassDescriptor; // Use local class descriptor for deserialization
- }
- }
- return resultClassDescriptor;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSingleOutputStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSingleOutputStreamOperator.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSingleOutputStreamOperator.java
index add81f7..e7b1697 100644
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSingleOutputStreamOperator.java
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSingleOutputStreamOperator.java
@@ -15,12 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.python.api.datastream;
import org.apache.flink.annotation.Public;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
-import org.python.core.PyObject;
+import org.python.core.PyObject;
/**
* A thin wrapper layer over {@link SingleOutputStreamOperator}
@@ -30,8 +31,20 @@ import org.python.core.PyObject;
*/
@Public
public class PythonSingleOutputStreamOperator extends PythonDataStream<SingleOutputStreamOperator<PyObject>> {
- public PythonSingleOutputStreamOperator(SingleOutputStreamOperator<PyObject> stream) {
+
+ PythonSingleOutputStreamOperator(SingleOutputStreamOperator<PyObject> stream) {
super(stream);
}
+
+ /**
+ * A thin wrapper layer over {@link SingleOutputStreamOperator#name(String)} .
+ *
+ * @param name operator name
+ * @return The named operator.
+ */
+ public PythonSingleOutputStreamOperator name(String name) {
+ this.stream.name(name);
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSplitStream.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSplitStream.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSplitStream.java
index e7621ca..81829b8 100644
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSplitStream.java
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonSplitStream.java
@@ -15,12 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.python.api.datastream;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.streaming.api.datastream.SplitStream;
-import org.python.core.PyObject;
+import org.python.core.PyObject;
/**
* A thin wrapper layer over {@link SplitStream}.
@@ -33,16 +34,15 @@ import org.python.core.PyObject;
@PublicEvolving
public class PythonSplitStream extends PythonDataStream<SplitStream<PyObject>> {
- public PythonSplitStream(SplitStream<PyObject> splitStream) {
+ PythonSplitStream(SplitStream<PyObject> splitStream) {
super(splitStream);
}
/**
- * A thin wrapper layer over {@link SplitStream#select(java.lang.String...)}
+ * A thin wrapper layer over {@link SplitStream#select(java.lang.String...)}.
*
- * @param output_names
- * The output names for which the operator will receive the
- * input.
+ * @param output_names The output names for which the operator will receive the
+ * input.
* @return Returns the selected {@link PythonDataStream}
*/
public PythonDataStream select(String... output_names) {
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonWindowedStream.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonWindowedStream.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonWindowedStream.java
index 563661e..3e25a4e 100644
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonWindowedStream.java
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/datastream/PythonWindowedStream.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.streaming.python.api.datastream;
import org.apache.flink.annotation.Public;
@@ -25,6 +26,7 @@ import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.python.api.functions.PyKey;
import org.apache.flink.streaming.python.api.functions.PythonApplyFunction;
import org.apache.flink.streaming.python.api.functions.PythonReduceFunction;
+
import org.python.core.PyObject;
import java.io.IOException;
@@ -41,7 +43,7 @@ import java.io.IOException;
public class PythonWindowedStream<W extends Window> {
private final WindowedStream<PyObject, PyKey, W> stream;
- public PythonWindowedStream(WindowedStream<PyObject, PyKey, W> stream) {
+ PythonWindowedStream(WindowedStream<PyObject, PyKey, W> stream) {
this.stream = stream;
}
@@ -63,7 +65,7 @@ public class PythonWindowedStream<W extends Window> {
* @return The data stream that is the result of applying the window function to the window.
*/
public PythonSingleOutputStreamOperator apply(
- WindowFunction<PyObject, PyObject, Object, W> fun) throws IOException {
+ WindowFunction<PyObject, Object, Object, W> fun) throws IOException {
return new PythonSingleOutputStreamOperator(stream.apply(new PythonApplyFunction<>(fun)));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonConstants.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonConstants.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonConstants.java
new file mode 100644
index 0000000..46dbc62
--- /dev/null
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonConstants.java
@@ -0,0 +1,26 @@
+/*
+ * 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.python.api.environment;
+
+/**
+ * Collection of constants for the Streaming Python API.
+ */
+public class PythonConstants {
+ public static final String FLINK_PYTHON_DC_ID = "flink";
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentConfig.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentConfig.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentConfig.java
deleted file mode 100644
index 2031af1..0000000
--- a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentConfig.java
+++ /dev/null
@@ -1,39 +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.python.api.environment;
-
-
-public class PythonEnvironmentConfig {
- public static final String FLINK_PYTHON_DC_ID = "flink";
-
- public static final String FLINK_PYTHON_PLAN_NAME = "plan.py";
-
- /**
- * Holds the path for the local python files cache. Is is set only on the client side by
- * the python streaming plan binder.
- */
- public static String pythonTmpCachePath;
-
- /**
- * Holds the path in the shared storage at which the python script(s) reside. It is set on the client side
- * within the execution process.
- */
- public static String FLINK_HDFS_PATH = "hdfs:///tmp/flink"; // "file:/tmp/flink"
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/9b7e4292/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentFactory.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentFactory.java b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentFactory.java
new file mode 100644
index 0000000..0f83169
--- /dev/null
+++ b/flink-libraries/flink-streaming-python/src/main/java/org/apache/flink/streaming/python/api/environment/PythonEnvironmentFactory.java
@@ -0,0 +1,141 @@
+/*
+ * 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.python.api.environment;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/**
+ * A factory for {@link PythonStreamExecutionEnvironment}s.
+ *
+ * <p>This class is a replacement for static factory methods defined in {@link StreamExecutionEnvironment} and allows
+ * us to pass state from the {@link org.apache.flink.streaming.python.api.PythonStreamBinder} instance
+ * to the created execution environment without having to rely on static fields.
+ */
+public class PythonEnvironmentFactory {
+ private final String localTmpPath;
+ private final Path tmpDistributedDir;
+ private final String scriptName;
+
+ public PythonEnvironmentFactory(String localTmpPath, Path tmpDistributedDir, String scriptName) {
+ this.localTmpPath = localTmpPath;
+ this.tmpDistributedDir = tmpDistributedDir;
+ this.scriptName = scriptName;
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#getExecutionEnvironment()}. In addition it takes
+ * care for required Jython serializers registration.
+ *
+ * @return The python execution environment of the context in which the program is
+ * executed.
+ */
+ public PythonStreamExecutionEnvironment get_execution_environment() {
+ return new PythonStreamExecutionEnvironment(StreamExecutionEnvironment.getExecutionEnvironment(), new Path(localTmpPath), tmpDistributedDir, scriptName);
+ }
+
+ /**
+ * Creates a {@link LocalStreamEnvironment}. The local execution environment
+ * will run the program in a multi-threaded fashion in the same JVM as the
+ * environment was created in. The default parallelism of the local
+ * environment is the number of hardware contexts (CPU cores / threads),
+ * unless it was specified differently by {@link PythonStreamExecutionEnvironment#set_parallelism(int)}.
+ *
+ * @param config Pass a custom configuration into the cluster
+ * @return A local execution environment with the specified parallelism.
+ */
+ public PythonStreamExecutionEnvironment create_local_execution_environment(Configuration config) {
+ return new PythonStreamExecutionEnvironment(new LocalStreamEnvironment(config), new Path(localTmpPath), tmpDistributedDir, scriptName);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createLocalEnvironment(int, Configuration)}.
+ *
+ * @param parallelism The parallelism for the local environment.
+ * @param config Pass a custom configuration into the cluster
+ * @return A local python execution environment with the specified parallelism.
+ */
+ public PythonStreamExecutionEnvironment create_local_execution_environment(int parallelism, Configuration config) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createLocalEnvironment(parallelism, config), new Path(localTmpPath), tmpDistributedDir, scriptName);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(String, int, String...)}.
+ *
+ * @param host The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port The port of the master (JobManager), where the program should
+ * be executed.
+ * @param jar_files The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, jar_files), new Path(localTmpPath), tmpDistributedDir, scriptName);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ *String, int, Configuration, String...)}.
+ *
+ * @param host The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port The port of the master (JobManager), where the program should
+ * be executed.
+ * @param config The configuration used by the client that connects to the remote cluster.
+ * @param jar_files The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, Configuration config, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, config, jar_files), new Path(localTmpPath), tmpDistributedDir, scriptName);
+ }
+
+ /**
+ * A thin wrapper layer over {@link StreamExecutionEnvironment#createRemoteEnvironment(
+ *String, int, int, String...)}.
+ *
+ * @param host The host name or address of the master (JobManager), where the
+ * program should be executed.
+ * @param port The port of the master (JobManager), where the program should
+ * be executed.
+ * @param parallelism The parallelism to use during the execution.
+ * @param jar_files The JAR files with code that needs to be shipped to the
+ * cluster. If the program uses user-defined functions,
+ * user-defined input formats, or any libraries, those must be
+ * provided in the JAR files.
+ * @return A remote environment that executes the program on a cluster.
+ */
+ public PythonStreamExecutionEnvironment create_remote_execution_environment(
+ String host, int port, int parallelism, String... jar_files) {
+ return new PythonStreamExecutionEnvironment(
+ StreamExecutionEnvironment.createRemoteEnvironment(host, port, parallelism, jar_files), new Path(localTmpPath), tmpDistributedDir, scriptName);
+ }
+}