You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2019/05/28 07:35:06 UTC
[flink] branch master updated (3a5bf89 -> 8510e29)
This is an automated email from the ASF dual-hosted git repository.
jincheng pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 3a5bf89 [FLINK-11283] Accessing the key when processing connected keyed stream
new ec61cf2 [FLINK-12327][python] Adds support to submit Python Table API job in CliFrontend This closes #8472
new 8510e29 [hotfix][python]fix command error for python API doc, and function call bug in table environment.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/ops/cli.md | 66 ++++++
docs/ops/cli.zh.md | 63 +++++-
.../org/apache/flink/client/cli/CliFrontend.java | 56 ++++--
.../apache/flink/client/cli/CliFrontendParser.java | 29 +++
.../apache/flink/client/cli/ProgramOptions.java | 74 ++++++-
.../flink/client/program/PackagedProgram.java | 32 ++-
.../apache/flink/client/python/PythonDriver.java | 168 ++++++++++++++++
.../flink/client/python/PythonGatewayServer.java | 21 +-
.../org/apache/flink/client/python/PythonUtil.java | 223 +++++++++++++++++++++
.../flink/client/python/PythonDriverTest.java | 104 ++++++++++
.../apache/flink/client/python/PythonUtilTest.java | 118 +++++++++++
flink-dist/pom.xml | 3 +
flink-dist/src/main/assemblies/bin.xml | 7 +
flink-python/pyflink/find_flink_home.py | 3 +
flink-python/pyflink/java_gateway.py | 14 +-
.../{util => table/examples/batch}/__init__.py | 0
.../pyflink/table/examples/batch/word_count.py | 79 ++++++++
flink-python/pyflink/table/table_environment.py | 2 +-
18 files changed, 1016 insertions(+), 46 deletions(-)
create mode 100644 flink-clients/src/main/java/org/apache/flink/client/python/PythonDriver.java
create mode 100644 flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java
create mode 100644 flink-clients/src/test/java/org/apache/flink/client/python/PythonDriverTest.java
create mode 100644 flink-clients/src/test/java/org/apache/flink/client/python/PythonUtilTest.java
copy flink-python/pyflink/{util => table/examples/batch}/__init__.py (100%)
create mode 100644 flink-python/pyflink/table/examples/batch/word_count.py
[flink] 02/02: [hotfix][python]fix command error for python API doc,
and function call bug in table environment.
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jincheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8510e294802c096418cdc121b21d64de630c201f
Author: sunjincheng121 <su...@gmail.com>
AuthorDate: Tue May 28 15:26:14 2019 +0800
[hotfix][python]fix command error for python API doc, and function call bug in table environment.
---
docs/ops/cli.md | 2 +-
docs/ops/cli.zh.md | 2 +-
.../src/main/java/org/apache/flink/client/cli/CliFrontend.java | 2 +-
.../src/main/java/org/apache/flink/client/python/PythonUtil.java | 2 +-
flink-python/pyflink/table/table_environment.py | 2 +-
5 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/docs/ops/cli.md b/docs/ops/cli.md
index 505207d..ab1fd16 100644
--- a/docs/ops/cli.md
+++ b/docs/ops/cli.md
@@ -121,7 +121,7 @@ These examples about how to submit a job in CLI.
- Run Python Table program in detached mode:
- ./bin/flink run -d examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+ ./bin/flink run -d -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
- Run Python Table program on a specific JobManager:
diff --git a/docs/ops/cli.zh.md b/docs/ops/cli.zh.md
index 93f16fb..ec6001b 100644
--- a/docs/ops/cli.zh.md
+++ b/docs/ops/cli.zh.md
@@ -121,7 +121,7 @@ available.
- 提交一个运行在detached模式下的Python Table的作业:
- ./bin/flink run -d examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+ ./bin/flink run -d -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
- 提交一个运行在指定JobManager上的Python Table的作业:
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index c591e6e..fe641fe 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -786,7 +786,7 @@ public class CliFrontend {
entryPointClass = PythonDriver.class.getCanonicalName();
} else {
if (jarFilePath == null) {
- throw new IllegalArgumentException("The program JAR file was not specified.");
+ throw new IllegalArgumentException("Java program should be specified a JAR file.");
}
jarFile = getJarFile(jarFilePath);
// Get assembler class
diff --git a/flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java b/flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java
index b9012a3..9fecd49 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java
@@ -96,7 +96,7 @@ public final class PythonUtil {
// 1. setup temporary local directory for the user files
String tmpDir = System.getProperty("java.io.tmpdir") +
- File.separator + "pyflink" + UUID.randomUUID();
+ File.separator + "pyflink" + File.separator + UUID.randomUUID();
Path tmpDirPath = new Path(tmpDir);
try {
diff --git a/flink-python/pyflink/table/table_environment.py b/flink-python/pyflink/table/table_environment.py
index 337a0e4..45e2688 100644
--- a/flink-python/pyflink/table/table_environment.py
+++ b/flink-python/pyflink/table/table_environment.py
@@ -374,7 +374,7 @@ class TableEnvironment(object):
j_execution_env, table_config._j_table_config)
t_env = BatchTableEnvironment(j_tenv)
- if table_config.parallelism is not None:
+ if table_config.parallelism() is not None:
t_env._j_tenv.execEnv().setParallelism(table_config.parallelism())
return t_env
[flink] 01/02: [FLINK-12327][python] Adds support to submit Python
Table API job in CliFrontend This closes #8472
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jincheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ec61cf21276fb6eea599b40ea999613ffeef48ef
Author: Huang Xingbo <hx...@gmail.com>
AuthorDate: Fri May 17 11:17:23 2019 +0800
[FLINK-12327][python] Adds support to submit Python Table API job in CliFrontend
This closes #8472
---
docs/ops/cli.md | 66 ++++++
docs/ops/cli.zh.md | 63 +++++-
.../org/apache/flink/client/cli/CliFrontend.java | 56 ++++--
.../apache/flink/client/cli/CliFrontendParser.java | 29 +++
.../apache/flink/client/cli/ProgramOptions.java | 74 ++++++-
.../flink/client/program/PackagedProgram.java | 32 ++-
.../apache/flink/client/python/PythonDriver.java | 168 ++++++++++++++++
.../flink/client/python/PythonGatewayServer.java | 21 +-
.../org/apache/flink/client/python/PythonUtil.java | 223 +++++++++++++++++++++
.../flink/client/python/PythonDriverTest.java | 104 ++++++++++
.../apache/flink/client/python/PythonUtilTest.java | 118 +++++++++++
flink-dist/pom.xml | 3 +
flink-dist/src/main/assemblies/bin.xml | 7 +
flink-python/pyflink/find_flink_home.py | 3 +
flink-python/pyflink/java_gateway.py | 14 +-
.../pyflink/table/examples/batch/__init__.py | 17 ++
.../pyflink/table/examples/batch/word_count.py | 79 ++++++++
17 files changed, 1032 insertions(+), 45 deletions(-)
diff --git a/docs/ops/cli.md b/docs/ops/cli.md
index b414b30..505207d 100644
--- a/docs/ops/cli.md
+++ b/docs/ops/cli.md
@@ -47,6 +47,12 @@ available.
{:toc}
## Examples
+### Job Submission Examples
+-----------------------------
+
+These examples about how to submit a job in CLI.
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
- Run example program with no arguments:
@@ -88,6 +94,53 @@ available.
./examples/batch/WordCount.jar \
--input hdfs:///user/hamlet.txt --output hdfs:///user/wordcount_out
+</div>
+
+<div data-lang="python" markdown="1">
+
+- Run Python Table program:
+
+ ./bin/flink run -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+
+- Run Python Table program with pyFiles:
+
+ ./bin/flink run -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar> \
+ -pyfs file:///user.txt,hdfs:///$namenode_address/username.txt
+
+- Run Python Table program with pyFiles and pyModule:
+
+ ./bin/flink run -pym batch.word_count -pyfs examples/python/table/batch -j <path/to/flink-table.jar>
+
+- Run Python Table program with parallelism 16:
+
+ ./bin/flink run -p 16 -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+
+- Run Python Table program with flink log output disabled:
+
+ ./bin/flink run -q -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+
+- Run Python Table program in detached mode:
+
+ ./bin/flink run -d examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+
+- Run Python Table program on a specific JobManager:
+
+ ./bin/flink run -m myJMHost:8081 \
+ -py examples/python/table/batch/word_count.py \
+ -j <path/to/flink-table.jar>
+
+- Run Python Table program using a [per-job YARN cluster]({{site.baseurl}}/ops/deployment/yarn_setup.html#run-a-single-flink-job-on-hadoop-yarn) with 2 TaskManagers:
+
+ ./bin/flink run -m yarn-cluster -yn 2 \
+ -py examples/python/table/batch/word_count.py \
+ -j <path/to/flink-table.jar>
+</div>
+
+### Job Management Examples
+-----------------------------
+
+These examples about how to manage a job in CLI.
+
- Display the optimized execution plan for the WordCount example program as JSON:
./bin/flink info ./examples/batch/WordCount.jar \
@@ -251,6 +304,19 @@ Action "run" compiles and runs a program.
program. Optional flag to override the
default value specified in the
configuration.
+ -py,--python <python-file> Python script with the program entry
+ point.The dependent resources can be
+ configured with the `--pyFiles` option.
+ -pyfs,--pyFiles <python-files> Attach custom python files for job.
+ Comma can be used as the separator to
+ specify multiple files. The standard
+ python resource file suffixes such as
+ .py/.egg/.zip are all supported.
+ (eg:--pyFiles file:///tmp/myresource.zip
+ ,hdfs:///$namenode_address/myresource2.zip)
+ -pym,--pyModule <python-module> Python module with the program entry
+ point. This option must be used in
+ conjunction with ` --pyFiles`.
-q,--sysoutLogging If present, suppress logging output to
standard out.
-s,--fromSavepoint <savepointPath> Path to a savepoint to restore the job
diff --git a/docs/ops/cli.zh.md b/docs/ops/cli.zh.md
index 7c02047..93f16fb 100644
--- a/docs/ops/cli.zh.md
+++ b/docs/ops/cli.zh.md
@@ -47,6 +47,12 @@ available.
{:toc}
## Examples
+### 作业提交示例
+-----------------------------
+
+这些示例是关于如何通过脚本提交一个作业
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
- Run example program with no arguments:
@@ -82,11 +88,57 @@ available.
./examples/batch/WordCount.jar \
--input file:///home/user/hamlet.txt --output file:///home/user/wordcount_out
-- Run example program using a [per-job YARN cluster]({{site.baseurl}}/ops/deployment/yarn_setup.html#run-a-single-flink-job-on-hadoop-yarn) with 2 TaskManagers:
+- Run example program using a [per-job YARN cluster]({{site.baseurl}}/zh/ops/deployment/yarn_setup.html#run-a-single-flink-job-on-hadoop-yarn) with 2 TaskManagers:
./bin/flink run -m yarn-cluster -yn 2 \
./examples/batch/WordCount.jar \
--input hdfs:///user/hamlet.txt --output hdfs:///user/wordcount_out
+
+</div>
+
+<div data-lang="python" markdown="1">
+
+- 提交一个Python Table的作业:
+
+ ./bin/flink run -py WordCount.py -j <path/to/flink-table.jar>
+
+- 提交一个有多个依赖的Python Table的作业:
+
+ ./bin/flink run -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar> \
+ -pyfs file:///user.txt,hdfs:///$namenode_address/username.txt
+
+- 提交一个有多个依赖的Python Table的作业，Python作业的主入口通过pym选项指定:
+
+ ./bin/flink run -pym batch.word_count -pyfs examples/python/table/batch -j <path/to/flink-table.jar>
+
+- 提交一个指定并发度为16的Python Table的作业:
+
+ ./bin/flink run -p 16 -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+
+- 提交一个关闭flink日志输出的Python Table的作业:
+
+ ./bin/flink run -q -py examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+
+- 提交一个运行在detached模式下的Python Table的作业:
+
+ ./bin/flink run -d examples/python/table/batch/word_count.py -j <path/to/flink-table.jar>
+
+- 提交一个运行在指定JobManager上的Python Table的作业:
+
+ ./bin/flink run -m myJMHost:8081 \
+ -py examples/python/table/batch/word_count.py \
+ -j <path/to/flink-table.jar>
+
+- 提交一个运行在有两个TaskManager的[per-job YARN cluster]({{site.baseurl}}/ops/deployment/yarn_setup.html#run-a-single-flink-job-on-hadoop-yarn)的Python Table的作业:
+
+ ./bin/flink run -m yarn-cluster -yn 2 \
+ -py examples/python/table/batch/word_count.py \
+ -j <path/to/flink-table.jar>
+
+</div>
+
+### 作业管理示例
+-----------------------------
- Display the optimized execution plan for the WordCount example program as JSON:
@@ -251,6 +303,15 @@ Action "run" compiles and runs a program.
program. Optional flag to override the
default value specified in the
configuration.
+ -py,--python <python-file> 指定Python作业的入口，依赖的资源文件可以通过
+ `--pyFiles`进行指定。
+ -pyfs,--pyFiles <python-files> 指定Python作业依赖的一些自定义的python文件，
+ 如果有多个文件，可以通过逗号(,)进行分隔。支持
+ 常用的python资源文件，例如(.py/.egg/.zip)。
+ (例如:--pyFiles file:///tmp/myresource.zip
+ ,hdfs:///$namenode_address/myresource2.zip)
+ -pym,--pyModule <python-module> 指定python程序的运行的模块入口，这个选项必须配合
+ `--pyFiles`一起使用。
-q,--sysoutLogging If present, suppress logging output to
standard out.
-s,--fromSavepoint <savepointPath> Path to a savepoint to restore the job
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index c6b5c9a..c591e6e 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -32,6 +32,7 @@ import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.client.program.ProgramMissingJobException;
import org.apache.flink.client.program.ProgramParametrizationException;
+import org.apache.flink.client.python.PythonDriver;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
@@ -185,8 +186,11 @@ public class CliFrontend {
return;
}
- if (runOptions.getJarFilePath() == null) {
- throw new CliArgsException("The program JAR file was not specified.");
+ if (!runOptions.isPython()) {
+ // Java program should be specified a JAR file
+ if (runOptions.getJarFilePath() == null) {
+ throw new CliArgsException("Java program should be specified a JAR file.");
+ }
}
final PackagedProgram program;
@@ -771,12 +775,42 @@ public class CliFrontend {
String jarFilePath = options.getJarFilePath();
List<URL> classpaths = options.getClasspaths();
- if (jarFilePath == null) {
- throw new IllegalArgumentException("The program JAR file was not specified.");
+ String entryPointClass;
+ File jarFile = null;
+ if (options.isPython()) {
+ // If the job is specified a jar file
+ if (jarFilePath != null) {
+ jarFile = getJarFile(jarFilePath);
+ }
+ // The entry point class of python job is PythonDriver
+ entryPointClass = PythonDriver.class.getCanonicalName();
+ } else {
+ if (jarFilePath == null) {
+ throw new IllegalArgumentException("The program JAR file was not specified.");
+ }
+ jarFile = getJarFile(jarFilePath);
+ // Get assembler class
+ entryPointClass = options.getEntryPointClassName();
}
- File jarFile = new File(jarFilePath);
+ PackagedProgram program = entryPointClass == null ?
+ new PackagedProgram(jarFile, classpaths, programArgs) :
+ new PackagedProgram(jarFile, classpaths, entryPointClass, programArgs);
+
+ program.setSavepointRestoreSettings(options.getSavepointRestoreSettings());
+
+ return program;
+ }
+ /**
+ * Gets the JAR file from the path.
+ *
+ * @param jarFilePath The path of JAR file
+ * @return The JAR file
+ * @throws FileNotFoundException The JAR file does not exist.
+ */
+ private File getJarFile(String jarFilePath) throws FileNotFoundException {
+ File jarFile = new File(jarFilePath);
// Check if JAR file exists
if (!jarFile.exists()) {
throw new FileNotFoundException("JAR file does not exist: " + jarFile);
@@ -784,17 +818,7 @@ public class CliFrontend {
else if (!jarFile.isFile()) {
throw new FileNotFoundException("JAR file is not a file: " + jarFile);
}
-
- // Get assembler class
- String entryPointClass = options.getEntryPointClassName();
-
- PackagedProgram program = entryPointClass == null ?
- new PackagedProgram(jarFile, classpaths, programArgs) :
- new PackagedProgram(jarFile, classpaths, entryPointClass, programArgs);
-
- program.setSavepointRestoreSettings(options.getSavepointRestoreSettings());
-
- return program;
+ return jarFile;
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
index cea3998..5872a54 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
@@ -118,6 +118,20 @@ public class CliFrontendParser {
public static final Option STOP_AND_DRAIN = new Option("d", "drain", false,
"Send MAX_WATERMARK before taking the savepoint and stopping the pipelne.");
+ static final Option PY_OPTION = new Option("py", "python", true,
+ "Python script with the program entry point. " +
+ "The dependent resources can be configured with the `--pyFiles` option.");
+
+ static final Option PYFILES_OPTION = new Option("pyfs", "pyFiles", true,
+ "Attach custom python files for job. " +
+ "Comma can be used as the separator to specify multiple files. " +
+ "The standard python resource file suffixes such as .py/.egg/.zip are all supported." +
+ "(eg: --pyFiles file:///tmp/myresource.zip,hdfs:///$namenode_address/myresource2.zip)");
+
+ static final Option PYMODULE_OPTION = new Option("pym", "pyModule", true,
+ "Python module with the program entry point. " +
+ "This option must be used in conjunction with `--pyFiles`.");
+
static {
HELP_OPTION.setRequired(false);
@@ -165,6 +179,15 @@ public class CliFrontendParser {
STOP_WITH_SAVEPOINT.setOptionalArg(true);
STOP_AND_DRAIN.setRequired(false);
+
+ PY_OPTION.setRequired(false);
+ PY_OPTION.setArgName("python");
+
+ PYFILES_OPTION.setRequired(false);
+ PYFILES_OPTION.setArgName("pyFiles");
+
+ PYMODULE_OPTION.setRequired(false);
+ PYMODULE_OPTION.setArgName("pyModule");
}
private static final Options RUN_OPTIONS = getRunCommandOptions();
@@ -186,6 +209,9 @@ public class CliFrontendParser {
options.addOption(DETACHED_OPTION);
options.addOption(SHUTDOWN_IF_ATTACHED_OPTION);
options.addOption(YARN_DETACHED_OPTION);
+ options.addOption(PY_OPTION);
+ options.addOption(PYFILES_OPTION);
+ options.addOption(PYMODULE_OPTION);
return options;
}
@@ -196,6 +222,9 @@ public class CliFrontendParser {
options.addOption(LOGGING_OPTION);
options.addOption(DETACHED_OPTION);
options.addOption(SHUTDOWN_IF_ATTACHED_OPTION);
+ options.addOption(PY_OPTION);
+ options.addOption(PYFILES_OPTION);
+ options.addOption(PYMODULE_OPTION);
return options;
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
index da03d64..30b3867 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
@@ -36,6 +36,9 @@ import static org.apache.flink.client.cli.CliFrontendParser.DETACHED_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.JAR_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.LOGGING_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.PARALLELISM_OPTION;
+import static org.apache.flink.client.cli.CliFrontendParser.PYFILES_OPTION;
+import static org.apache.flink.client.cli.CliFrontendParser.PYMODULE_OPTION;
+import static org.apache.flink.client.cli.CliFrontendParser.PY_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.SHUTDOWN_IF_ATTACHED_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.YARN_DETACHED_OPTION;
@@ -62,17 +65,71 @@ public abstract class ProgramOptions extends CommandLineOptions {
private final SavepointRestoreSettings savepointSettings;
+ /**
+ * Flag indicating whether the job is a Python job.
+ */
+ private final boolean isPython;
+
protected ProgramOptions(CommandLine line) throws CliArgsException {
super(line);
String[] args = line.hasOption(ARGS_OPTION.getOpt()) ?
- line.getOptionValues(ARGS_OPTION.getOpt()) :
- line.getArgs();
+ line.getOptionValues(ARGS_OPTION.getOpt()) :
+ line.getArgs();
+
+ isPython = line.hasOption(PY_OPTION.getOpt()) | line.hasOption(PYMODULE_OPTION.getOpt());
+ // If specified the option -py(--python)
+ if (line.hasOption(PY_OPTION.getOpt())) {
+ // Cannot use option -py and -pym simultaneously.
+ if (line.hasOption(PYMODULE_OPTION.getOpt())) {
+ throw new CliArgsException("Cannot use option -py and -pym simultaneously.");
+ }
+ // The cli cmd args which will be transferred to PythonDriver will be transformed as follows:
+ // CLI cmd : -py ${python.py} pyfs [optional] ${py-files} [optional] ${other args}.
+ // PythonDriver args: py ${python.py} [optional] pyfs [optional] ${py-files} [optional] ${other args}.
+ // -------------------------------transformed-------------------------------------------------------
+ // e.g. -py wordcount.py(CLI cmd) -----------> py wordcount.py(PythonDriver args)
+ // e.g. -py wordcount.py -pyfs file:///AAA.py,hdfs:///BBB.py --input in.txt --output out.txt(CLI cmd)
+ // -----> py wordcount.py pyfs file:///AAA.py,hdfs:///BBB.py --input in.txt --output out.txt(PythonDriver args)
+ String[] newArgs;
+ int argIndex;
+ if (line.hasOption(PYFILES_OPTION.getOpt())) {
+ newArgs = new String[args.length + 4];
+ newArgs[2] = PYFILES_OPTION.getOpt();
+ newArgs[3] = line.getOptionValue(PYFILES_OPTION.getOpt());
+ argIndex = 4;
+ } else {
+ newArgs = new String[args.length + 2];
+ argIndex = 2;
+ }
+ newArgs[0] = PY_OPTION.getOpt();
+ newArgs[1] = line.getOptionValue(PY_OPTION.getOpt());
+ System.arraycopy(args, 0, newArgs, argIndex, args.length);
+ args = newArgs;
+ }
+
+ // If specified the option -pym(--pyModule)
+ if (line.hasOption(PYMODULE_OPTION.getOpt())) {
+ // If you specify the option -pym, you should specify the option --pyFiles simultaneously.
+ if (!line.hasOption(PYFILES_OPTION.getOpt())) {
+ throw new CliArgsException("-pym must be used in conjunction with `--pyFiles`");
+ }
+ // The cli cmd args which will be transferred to PythonDriver will be transformed as follows:
+ // CLI cmd : -pym ${py-module} -pyfs ${py-files} [optional] ${other args}.
+ // PythonDriver args: pym ${py-module} pyfs ${py-files} [optional] ${other args}.
+ // e.g. -pym AAA.fun -pyfs AAA.zip(CLI cmd) ----> pym AAA.fun -pyfs AAA.zip(PythonDriver args)
+ String[] newArgs = new String[args.length + 4];
+ newArgs[0] = PYMODULE_OPTION.getOpt();
+ newArgs[1] = line.getOptionValue(PYMODULE_OPTION.getOpt());
+ newArgs[2] = PYFILES_OPTION.getOpt();
+ newArgs[3] = line.getOptionValue(PYFILES_OPTION.getOpt());
+ System.arraycopy(args, 0, newArgs, 4, args.length);
+ args = newArgs;
+ }
if (line.hasOption(JAR_OPTION.getOpt())) {
this.jarFilePath = line.getOptionValue(JAR_OPTION.getOpt());
- }
- else if (args.length > 0) {
+ } else if (!isPython && args.length > 0) {
jarFilePath = args[0];
args = Arrays.copyOfRange(args, 1, args.length);
}
@@ -95,7 +152,7 @@ public abstract class ProgramOptions extends CommandLineOptions {
this.classpaths = classpaths;
this.entryPointClass = line.hasOption(CLASS_OPTION.getOpt()) ?
- line.getOptionValue(CLASS_OPTION.getOpt()) : null;
+ line.getOptionValue(CLASS_OPTION.getOpt()) : null;
if (line.hasOption(PARALLELISM_OPTION.getOpt())) {
String parString = line.getOptionValue(PARALLELISM_OPTION.getOpt());
@@ -156,4 +213,11 @@ public abstract class ProgramOptions extends CommandLineOptions {
public SavepointRestoreSettings getSavepointRestoreSettings() {
return savepointSettings;
}
+
+ /**
+ * Indicates whether the job is a Python job.
+ */
+ public boolean isPython() {
+ return isPython;
+ }
}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
index 8f5ccba..77b5d29 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
@@ -21,6 +21,7 @@ package org.apache.flink.client.program;
import org.apache.flink.api.common.Plan;
import org.apache.flink.api.common.Program;
import org.apache.flink.api.common.ProgramDescription;
+import org.apache.flink.client.python.PythonDriver;
import org.apache.flink.optimizer.Optimizer;
import org.apache.flink.optimizer.dag.DataSinkNode;
import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
@@ -91,6 +92,11 @@ public class PackagedProgram {
private SavepointRestoreSettings savepointSettings = SavepointRestoreSettings.none();
/**
+ * Flag indicating whether the job is a Python job.
+ */
+ private final boolean isPython;
+
+ /**
* Creates an instance that wraps the plan defined in the jar file using the given
* argument.
*
@@ -169,18 +175,21 @@ public class PackagedProgram {
* may be a missing / wrong class or manifest files.
*/
public PackagedProgram(File jarFile, List<URL> classpaths, @Nullable String entryPointClassName, String... args) throws ProgramInvocationException {
- if (jarFile == null) {
- throw new IllegalArgumentException("The jar file must not be null.");
- }
+ // Whether the job is a Python job.
+ isPython = entryPointClassName != null && entryPointClassName.equals(PythonDriver.class.getCanonicalName());
- URL jarFileUrl;
- try {
- jarFileUrl = jarFile.getAbsoluteFile().toURI().toURL();
- } catch (MalformedURLException e1) {
- throw new IllegalArgumentException("The jar file path is invalid.");
- }
+ URL jarFileUrl = null;
+ if (jarFile != null) {
+ try {
+ jarFileUrl = jarFile.getAbsoluteFile().toURI().toURL();
+ } catch (MalformedURLException e1) {
+ throw new IllegalArgumentException("The jar file path is invalid.");
+ }
- checkJarFile(jarFileUrl);
+ checkJarFile(jarFileUrl);
+ } else if (!isPython) {
+ throw new IllegalArgumentException("The jar file must not be null.");
+ }
this.jarFile = jarFileUrl;
this.args = args == null ? new String[0] : args;
@@ -191,7 +200,7 @@ public class PackagedProgram {
}
// now that we have an entry point, we can extract the nested jar files (if any)
- this.extractedTempLibraries = extractContainedLibraries(jarFileUrl);
+ this.extractedTempLibraries = jarFileUrl == null ? Collections.emptyList() : extractContainedLibraries(jarFileUrl);
this.classpaths = classpaths;
this.userCodeClassLoader = JobWithJars.buildUserCodeClassLoader(getAllLibraries(), classpaths, getClass().getClassLoader());
@@ -233,6 +242,7 @@ public class PackagedProgram {
// load the entry point class
this.mainClass = entryPointClass;
+ isPython = entryPointClass == PythonDriver.class;
// if the entry point is a program, instantiate the class and get the plan
if (Program.class.isAssignableFrom(this.mainClass)) {
diff --git a/flink-clients/src/main/java/org/apache/flink/client/python/PythonDriver.java b/flink-clients/src/main/java/org/apache/flink/client/python/PythonDriver.java
new file mode 100644
index 0000000..e43a24e
--- /dev/null
+++ b/flink-clients/src/main/java/org/apache/flink/client/python/PythonDriver.java
@@ -0,0 +1,168 @@
+/*
+ * 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.client.python;
+
+import org.apache.flink.core.fs.Path;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.GatewayServer;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A main class used to launch Python applications. It executes python as a
+ * subprocess and then has it connect back to the JVM to access system properties, etc.
+ */
+public class PythonDriver {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonDriver.class);
+
+ public static void main(String[] args) {
+ // the python job needs at least 2 args.
+ // e.g. py a.py ...
+ // e.g. pym a.b -pyfs a.zip ...
+ if (args.length < 2) {
+ LOG.error("Required at least two arguments, only python file or python module is available.");
+ System.exit(1);
+ }
+ // parse args
+ Map<String, List<String>> parsedArgs = parseOptions(args);
+ // start gateway server
+ GatewayServer gatewayServer = startGatewayServer();
+ // prepare python env
+
+ // map filename to its Path
+ Map<String, Path> filePathMap = new HashMap<>();
+ // commands which will be exec in python progress.
+ List<String> commands = constructPythonCommands(filePathMap, parsedArgs);
+ try {
+ // prepare the exec environment of python progress.
+ PythonUtil.PythonEnvironment pythonEnv = PythonUtil.preparePythonEnvironment(filePathMap);
+ // set env variable PYFLINK_GATEWAY_PORT for connecting of python gateway in python progress.
+ pythonEnv.systemEnv.put("PYFLINK_GATEWAY_PORT", String.valueOf(gatewayServer.getListeningPort()));
+ // start the python process.
+ Process pythonProcess = PythonUtil.startPythonProcess(pythonEnv, commands);
+ int exitCode = pythonProcess.waitFor();
+ if (exitCode != 0) {
+ throw new RuntimeException("Python process exits with code: " + exitCode);
+ }
+ } catch (Throwable e) {
+ LOG.error("Run python process failed", e);
+ } finally {
+ gatewayServer.shutdown();
+ }
+ }
+
+ /**
+ * Creates a GatewayServer run in a daemon thread.
+ *
+ * @return The created GatewayServer
+ */
+ public static GatewayServer startGatewayServer() {
+ InetAddress localhost = InetAddress.getLoopbackAddress();
+ GatewayServer gatewayServer = new GatewayServer.GatewayServerBuilder()
+ .javaPort(0)
+ .javaAddress(localhost)
+ .build();
+ Thread thread = new Thread(gatewayServer::start);
+ thread.setName("py4j-gateway");
+ thread.setDaemon(true);
+ thread.start();
+ try {
+ thread.join();
+ } catch (InterruptedException e) {
+ LOG.error("The gateway server thread join failed.", e);
+ System.exit(1);
+ }
+ return gatewayServer;
+ }
+
+ /**
+ * Constructs the Python commands which will be executed in python process.
+ *
+ * @param filePathMap stores python file name to its path
+ * @param parsedArgs parsed args
+ */
+ public static List<String> constructPythonCommands(Map<String, Path> filePathMap, Map<String, List<String>> parsedArgs) {
+ List<String> commands = new ArrayList<>();
+ if (parsedArgs.containsKey("py")) {
+ String pythonFile = parsedArgs.get("py").get(0);
+ Path pythonFilePath = new Path(pythonFile);
+ filePathMap.put(pythonFilePath.getName(), pythonFilePath);
+ commands.add(pythonFilePath.getName());
+ }
+ if (parsedArgs.containsKey("pym")) {
+ String pyModule = parsedArgs.get("pym").get(0);
+ commands.add("-m");
+ commands.add(pyModule);
+ }
+ if (parsedArgs.containsKey("pyfs")) {
+ List<String> pyFiles = parsedArgs.get("pyfs");
+ for (String pyFile : pyFiles) {
+ Path pyFilePath = new Path(pyFile);
+ filePathMap.put(pyFilePath.getName(), pyFilePath);
+ }
+ }
+ if (parsedArgs.containsKey("args")) {
+ commands.addAll(parsedArgs.get("args"));
+ }
+ return commands;
+ }
+
+ /**
+ * Parses the args to the map format.
+ *
+ * @param args ["py", "xxx.py",
+ * "pyfs", "a.py,b.py,c.py",
+ * "--input", "in.txt"]
+ * @return {"py"->List("xxx.py"),"pyfs"->List("a.py","b.py","c.py"),"args"->List("--input","in.txt")}
+ */
+ public static Map<String, List<String>> parseOptions(String[] args) {
+ Map<String, List<String>> parsedArgs = new HashMap<>();
+ int argIndex = 0;
+ boolean isEntrypointSpecified = false;
+ // valid args should include python or pyModule field and their value.
+ if (args[0].equals("py") || args[0].equals("pym")) {
+ parsedArgs.put(args[0], Collections.singletonList(args[1]));
+ argIndex = 2;
+ isEntrypointSpecified = true;
+ }
+ if (isEntrypointSpecified && args.length > 2 && args[2].equals("pyfs")) {
+ List<String> pyFilesList = new ArrayList<>(Arrays.asList(args[3].split(",")));
+ parsedArgs.put(args[2], pyFilesList);
+ argIndex = 4;
+ }
+ if (!isEntrypointSpecified) {
+ throw new RuntimeException("The Python entrypoint has not been specified. It can be specified with option -py or -pym");
+ }
+ // if arg include other args, the key "args" will map to other args.
+ if (args.length > argIndex) {
+ List<String> otherArgList = new ArrayList<>(args.length - argIndex);
+ otherArgList.addAll(Arrays.asList(args).subList(argIndex, args.length));
+ parsedArgs.put("args", otherArgList);
+ }
+ return parsedArgs;
+ }
+}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/python/PythonGatewayServer.java b/flink-clients/src/main/java/org/apache/flink/client/python/PythonGatewayServer.java
index 6432a67..64f2ef1 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/python/PythonGatewayServer.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/python/PythonGatewayServer.java
@@ -25,6 +25,7 @@ import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetAddress;
+import java.nio.file.Files;
/**
* The Py4j Gateway Server provides RPC service for user's python process.
@@ -56,15 +57,17 @@ public class PythonGatewayServer {
// Tells python side the port of our java rpc server
String handshakeFilePath = System.getenv("_PYFLINK_CONN_INFO_PATH");
File handshakeFile = new File(handshakeFilePath);
- if (handshakeFile.createNewFile()) {
- FileOutputStream fileOutputStream = new FileOutputStream(handshakeFile);
- DataOutputStream stream = new DataOutputStream(fileOutputStream);
- stream.writeInt(boundPort);
- stream.close();
- fileOutputStream.close();
- } else {
- System.out.println("Can't create handshake file: " + handshakeFilePath + ", now exit...");
- return;
+ File tmpPath = Files.createTempFile(handshakeFile.getParentFile().toPath(),
+ "connection", ".info").toFile();
+ FileOutputStream fileOutputStream = new FileOutputStream(tmpPath);
+ DataOutputStream stream = new DataOutputStream(fileOutputStream);
+ stream.writeInt(boundPort);
+ stream.close();
+ fileOutputStream.close();
+
+ if (!tmpPath.renameTo(handshakeFile)) {
+ System.out.println("Unable to write connection information to handshake file: " + handshakeFilePath + ", now exit...");
+ System.exit(1);
}
// Exit on EOF or broken pipe. This ensures that the server dies
diff --git a/flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java b/flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java
new file mode 100644
index 0000000..b9012a3
--- /dev/null
+++ b/flink-clients/src/main/java/org/apache/flink/client/python/PythonUtil.java
@@ -0,0 +1,223 @@
+/*
+ * 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.client.python;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.FileUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * The util class help to prepare Python env and run the python process.
+ */
+public final class PythonUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(PythonUtil.class);
+
+ private static final String FLINK_OPT_DIR = System.getenv("FLINK_OPT_DIR");
+
+ private static final String FLINK_OPT_DIR_PYTHON = FLINK_OPT_DIR + File.separator + "python";
+
+ /**
+ * Wraps Python exec environment.
+ */
+ public static class PythonEnvironment {
+ public String workingDirectory;
+
+ public String pythonExec = "python";
+
+ public String pythonPath;
+
+ Map<String, String> systemEnv = new HashMap<>();
+ }
+
+ /**
+ * The hook thread that delete the tmp working dir of python process after the python process shutdown.
+ */
+ private static class ShutDownPythonHook extends Thread {
+ private Process p;
+ private String pyFileDir;
+
+ public ShutDownPythonHook(Process p, String pyFileDir) {
+ this.p = p;
+ this.pyFileDir = pyFileDir;
+ }
+
+ public void run() {
+
+ p.destroyForcibly();
+
+ if (pyFileDir != null) {
+ File pyDir = new File(pyFileDir);
+ FileUtils.deleteDirectoryQuietly(pyDir);
+ }
+ }
+ }
+
+
+ /**
+ * Prepares PythonEnvironment to start python process.
+ *
+ * @param filePathMap map file name to its file path.
+ * @return PythonEnvironment the Python environment which will be executed in Python process.
+ */
+ public static PythonEnvironment preparePythonEnvironment(Map<String, Path> filePathMap) {
+ PythonEnvironment env = new PythonEnvironment();
+
+ // 1. setup temporary local directory for the user files
+ String tmpDir = System.getProperty("java.io.tmpdir") +
+ File.separator + "pyflink" + UUID.randomUUID();
+
+ Path tmpDirPath = new Path(tmpDir);
+ try {
+ FileSystem fs = tmpDirPath.getFileSystem();
+ if (fs.exists(tmpDirPath)) {
+ fs.delete(tmpDirPath, true);
+ }
+ fs.mkdirs(tmpDirPath);
+ } catch (IOException e) {
+ LOG.error("Prepare tmp directory failed.", e);
+ }
+
+ env.workingDirectory = tmpDirPath.toString();
+
+ StringBuilder pythonPathEnv = new StringBuilder();
+
+ pythonPathEnv.append(env.workingDirectory);
+
+ // 2. create symbolLink in the working directory for the pyflink dependency libs.
+ List<java.nio.file.Path> pythonLibs = getLibFiles(FLINK_OPT_DIR_PYTHON);
+ for (java.nio.file.Path libPath : pythonLibs) {
+ java.nio.file.Path symbolicLinkFilePath = FileSystems.getDefault().getPath(env.workingDirectory,
+ libPath.getFileName().toString());
+ createSymbolicLinkForPyflinkLib(libPath, symbolicLinkFilePath);
+ pythonPathEnv.append(File.pathSeparator);
+ pythonPathEnv.append(symbolicLinkFilePath.toString());
+ }
+
+ // 3. copy relevant python files to tmp dir and set them in PYTHONPATH.
+ filePathMap.forEach((sourceFileName, sourcePath) -> {
+ Path targetPath = new Path(tmpDirPath, sourceFileName);
+ try {
+ FileUtils.copy(sourcePath, targetPath, true);
+ } catch (IOException e) {
+ LOG.error("Copy files to tmp dir failed", e);
+ }
+ String targetFileName = targetPath.toString();
+ pythonPathEnv.append(File.pathSeparator);
+ pythonPathEnv.append(targetFileName);
+
+ });
+
+ env.pythonPath = pythonPathEnv.toString();
+ return env;
+ }
+
+ /**
+ * Gets pyflink dependent libs in specified directory.
+ *
+ * @param libDir The lib directory
+ */
+ public static List<java.nio.file.Path> getLibFiles(String libDir) {
+ final List<java.nio.file.Path> libFiles = new ArrayList<>();
+ SimpleFileVisitor<java.nio.file.Path> finder = new SimpleFileVisitor<java.nio.file.Path>() {
+ @Override
+ public FileVisitResult visitFile(java.nio.file.Path file, BasicFileAttributes attrs) throws IOException {
+ // exclude .txt file
+ if (!file.toString().endsWith(".txt")) {
+ libFiles.add(file);
+ }
+ return FileVisitResult.CONTINUE;
+ }
+ };
+ try {
+ Files.walkFileTree(FileSystems.getDefault().getPath(libDir), finder);
+ } catch (IOException e) {
+ LOG.error("Gets pyflink dependent libs failed.", e);
+ }
+ return libFiles;
+ }
+
+ /**
+ * Creates symbolLink in working directory for pyflink lib.
+ *
+ * @param libPath the pyflink lib file path.
+ * @param symbolicLinkPath the symbolic link to pyflink lib.
+ */
+ public static void createSymbolicLinkForPyflinkLib(java.nio.file.Path libPath, java.nio.file.Path symbolicLinkPath) {
+ try {
+ Files.createSymbolicLink(symbolicLinkPath, libPath);
+ } catch (IOException e) {
+ LOG.error("Create symbol link for pyflink lib failed.", e);
+ LOG.info("Try to copy pyflink lib to working directory");
+ try {
+ Files.copy(libPath, symbolicLinkPath);
+ } catch (IOException ex) {
+ LOG.error("Copy pylink lib to working directory failed", ex);
+ }
+ }
+ }
+
+ /**
+ * Starts python process.
+ *
+ * @param pythonEnv the python Environment which will be in a process.
+ * @param commands the commands that python process will execute.
+ * @return the process represent the python process.
+ * @throws IOException Thrown if an error occurred when python process start.
+ */
+ public static Process startPythonProcess(PythonEnvironment pythonEnv, List<String> commands) throws IOException {
+ ProcessBuilder pythonProcessBuilder = new ProcessBuilder();
+ Map<String, String> env = pythonProcessBuilder.environment();
+ env.put("PYTHONPATH", pythonEnv.pythonPath);
+ pythonEnv.systemEnv.forEach(env::put);
+ commands.add(0, pythonEnv.pythonExec);
+ pythonProcessBuilder.command(commands);
+ // set the working directory.
+ pythonProcessBuilder.directory(new File(pythonEnv.workingDirectory));
+ // redirect the stderr to stdout
+ pythonProcessBuilder.redirectErrorStream(true);
+ // set the child process the output same as the parent process.
+ pythonProcessBuilder.redirectOutput(ProcessBuilder.Redirect.INHERIT);
+ Process process = pythonProcessBuilder.start();
+ if (!process.isAlive()) {
+ throw new RuntimeException("Failed to start Python process. ");
+ }
+
+ // Make sure that the python sub process will be killed when JVM exit
+ ShutDownPythonHook hook = new ShutDownPythonHook(process, pythonEnv.workingDirectory);
+ Runtime.getRuntime().addShutdownHook(hook);
+
+ return process;
+ }
+}
diff --git a/flink-clients/src/test/java/org/apache/flink/client/python/PythonDriverTest.java b/flink-clients/src/test/java/org/apache/flink/client/python/PythonDriverTest.java
new file mode 100644
index 0000000..0b6f570
--- /dev/null
+++ b/flink-clients/src/test/java/org/apache/flink/client/python/PythonDriverTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.client.python;
+
+import org.apache.flink.core.fs.Path;
+
+import org.junit.Assert;
+import org.junit.Test;
+import py4j.GatewayServer;
+
+import java.io.IOException;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Tests for the {@link PythonDriver}.
+ */
+public class PythonDriverTest {
+ @Test
+ public void testStartGatewayServer() {
+ GatewayServer gatewayServer = PythonDriver.startGatewayServer();
+ try {
+ Socket socket = new Socket("localhost", gatewayServer.getListeningPort());
+ assert socket.isConnected();
+ } catch (IOException e) {
+ throw new RuntimeException("Connect Gateway Server failed");
+ } finally {
+ gatewayServer.shutdown();
+ }
+ }
+
+ @Test
+ public void testConstructCommands() {
+ Map<String, Path> filePathMap = new HashMap<>();
+ Map<String, List<String>> parseArgs = new HashMap<>();
+ parseArgs.put("py", Collections.singletonList("xxx.py"));
+ List<String> pyFilesList = new ArrayList<>();
+ pyFilesList.add("a.py");
+ pyFilesList.add("b.py");
+ pyFilesList.add("c.py");
+ parseArgs.put("pyfs", pyFilesList);
+ List<String> otherArgs = new ArrayList<>();
+ otherArgs.add("--input");
+ otherArgs.add("in.txt");
+ parseArgs.put("args", otherArgs);
+ List<String> commands = PythonDriver.constructPythonCommands(filePathMap, parseArgs);
+ Path pythonPath = filePathMap.get("xxx.py");
+ Assert.assertNotNull(pythonPath);
+ Assert.assertEquals(pythonPath.getName(), "xxx.py");
+ Path aPyFilePath = filePathMap.get("a.py");
+ Assert.assertNotNull(aPyFilePath);
+ Assert.assertEquals(aPyFilePath.getName(), "a.py");
+ Path bPyFilePath = filePathMap.get("b.py");
+ Assert.assertNotNull(bPyFilePath);
+ Assert.assertEquals(bPyFilePath.getName(), "b.py");
+ Path cPyFilePath = filePathMap.get("c.py");
+ Assert.assertNotNull(cPyFilePath);
+ Assert.assertEquals(cPyFilePath.getName(), "c.py");
+ Assert.assertEquals(3, commands.size());
+ Assert.assertEquals(commands.get(0), "xxx.py");
+ Assert.assertEquals(commands.get(1), "--input");
+ Assert.assertEquals(commands.get(2), "in.txt");
+ }
+
+ @Test
+ public void testParseOptions() {
+ String[] args = {"py", "xxx.py", "pyfs", "a.py,b.py,c.py", "--input", "in.txt"};
+ Map<String, List<String>> parsedArgs = PythonDriver.parseOptions(args);
+ List<String> pythonMainFile = parsedArgs.get("py");
+ Assert.assertNotNull(pythonMainFile);
+ Assert.assertEquals(1, pythonMainFile.size());
+ Assert.assertEquals(pythonMainFile.get(0), args[1]);
+ List<String> pyFilesList = parsedArgs.get("pyfs");
+ Assert.assertEquals(3, pyFilesList.size());
+ String[] pyFiles = args[3].split(",");
+ for (int i = 0; i < pyFiles.length; i++) {
+ assert pyFilesList.get(i).equals(pyFiles[i]);
+ }
+ List<String> otherArgs = parsedArgs.get("args");
+ for (int i = 4; i < args.length; i++) {
+ Assert.assertEquals(otherArgs.get(i - 4), args[i]);
+ }
+ }
+}
diff --git a/flink-clients/src/test/java/org/apache/flink/client/python/PythonUtilTest.java b/flink-clients/src/test/java/org/apache/flink/client/python/PythonUtilTest.java
new file mode 100644
index 0000000..4b14ced
--- /dev/null
+++ b/flink-clients/src/test/java/org/apache/flink/client/python/PythonUtilTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.client.python;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Tests for the {@link PythonUtil}.
+ */
+public class PythonUtilTest {
+ private Path sourceTmpDirPath;
+ private Path targetTmpDirPath;
+ private FileSystem sourceFs;
+ private FileSystem targetFs;
+
+ @Before
+ public void prepareTestEnvironment() {
+ String sourceTmpDir = System.getProperty("java.io.tmpdir") +
+ File.separator + "source_" + UUID.randomUUID();
+ String targetTmpDir = System.getProperty("java.io.tmpdir") +
+ File.separator + "target_" + UUID.randomUUID();
+
+ sourceTmpDirPath = new Path(sourceTmpDir);
+ targetTmpDirPath = new Path(targetTmpDir);
+ try {
+ sourceFs = sourceTmpDirPath.getFileSystem();
+ if (sourceFs.exists(sourceTmpDirPath)) {
+ sourceFs.delete(sourceTmpDirPath, true);
+ }
+ sourceFs.mkdirs(sourceTmpDirPath);
+ targetFs = targetTmpDirPath.getFileSystem();
+ if (targetFs.exists(targetTmpDirPath)) {
+ targetFs.delete(targetTmpDirPath, true);
+ }
+ targetFs.mkdirs(targetTmpDirPath);
+ } catch (IOException e) {
+ throw new RuntimeException("initial PythonUtil test environment failed");
+ }
+ }
+
+ @Test
+ public void testStartPythonProcess() {
+ PythonUtil.PythonEnvironment pythonEnv = new PythonUtil.PythonEnvironment();
+ pythonEnv.workingDirectory = targetTmpDirPath.toString();
+ pythonEnv.pythonPath = targetTmpDirPath.toString();
+ List<String> commands = new ArrayList<>();
+ Path pyPath = new Path(targetTmpDirPath, "word_count.py");
+ try {
+ targetFs.create(pyPath, FileSystem.WriteMode.OVERWRITE);
+ File pyFile = new File(pyPath.toString());
+ String pyProgram = "#!/usr/bin/python\n" +
+ "# -*- coding: UTF-8 -*-\n" +
+ "import sys\n" +
+ "\n" +
+ "if __name__=='__main__':\n" +
+ "\tfilename = sys.argv[1]\n" +
+ "\tfo = open(filename, \"w\")\n" +
+ "\tfo.write( \"hello world\")\n" +
+ "\tfo.close()";
+ Files.write(pyFile.toPath(), pyProgram.getBytes(), StandardOpenOption.WRITE);
+ Path result = new Path(targetTmpDirPath, "word_count_result.txt");
+ commands.add(pyFile.getName());
+ commands.add(result.getName());
+ Process pythonProcess = PythonUtil.startPythonProcess(pythonEnv, commands);
+ int exitCode = pythonProcess.waitFor();
+ if (exitCode != 0) {
+ throw new RuntimeException("Python process exits with code: " + exitCode);
+ }
+ String cmdResult = new String(Files.readAllBytes(new File(result.toString()).toPath()));
+ Assert.assertEquals(cmdResult, "hello world");
+ pythonProcess.destroyForcibly();
+ targetFs.delete(pyPath, true);
+ targetFs.delete(result, true);
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException("test start Python process failed " + e.getMessage());
+ }
+ }
+
+ @After
+ public void cleanEnvironment() {
+ try {
+ sourceFs.delete(sourceTmpDirPath, true);
+ targetFs.delete(targetTmpDirPath, true);
+ } catch (IOException e) {
+ throw new RuntimeException("delete tmp dir failed " + e.getMessage());
+ }
+ }
+}
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index 52a6466..1350f10 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -572,6 +572,9 @@ under the License.
<relocation>
<pattern>py4j</pattern>
<shadedPattern>org.apache.flink.api.python.py4j</shadedPattern>
+ <includes>
+ <include>py4j.*</include>
+ </includes>
</relocation>
</relocations>
</configuration>
diff --git a/flink-dist/src/main/assemblies/bin.xml b/flink-dist/src/main/assemblies/bin.xml
index 3eb5698..788ec1b 100644
--- a/flink-dist/src/main/assemblies/bin.xml
+++ b/flink-dist/src/main/assemblies/bin.xml
@@ -242,6 +242,13 @@ under the License.
<fileMode>0755</fileMode>
</fileSet>
+ <!-- copy python table example to examples of dist -->
+ <fileSet>
+ <directory>../flink-python/pyflink/table/examples</directory>
+ <outputDirectory>examples/python/table</outputDirectory>
+ <fileMode>0755</fileMode>
+ </fileSet>
+
</fileSets>
</assembly>
diff --git a/flink-python/pyflink/find_flink_home.py b/flink-python/pyflink/find_flink_home.py
index 0491368..9806490 100644
--- a/flink-python/pyflink/find_flink_home.py
+++ b/flink-python/pyflink/find_flink_home.py
@@ -28,6 +28,9 @@ def _find_flink_home():
# If the environment has set FLINK_HOME, trust it.
if 'FLINK_HOME' in os.environ:
return os.environ['FLINK_HOME']
+ elif 'FLINK_ROOT_DIR' in os.environ:
+ os.environ['FLINK_HOME'] = os.environ['FLINK_ROOT_DIR']
+ return os.environ['FLINK_ROOT_DIR']
else:
try:
flink_root_dir = os.path.abspath(os.path.dirname(os.path.abspath(__file__)) + "/../../")
diff --git a/flink-python/pyflink/java_gateway.py b/flink-python/pyflink/java_gateway.py
index e5c8330..b218d23 100644
--- a/flink-python/pyflink/java_gateway.py
+++ b/flink-python/pyflink/java_gateway.py
@@ -28,7 +28,6 @@ from threading import RLock
from py4j.java_gateway import java_import, JavaGateway, GatewayParameters
from pyflink.find_flink_home import _find_flink_home
-
_gateway = None
_lock = RLock()
@@ -46,6 +45,9 @@ def get_gateway():
_gateway = JavaGateway(gateway_parameters=gateway_param)
else:
_gateway = launch_gateway()
+
+ # import the flink view
+ import_flink_view(_gateway)
return _gateway
@@ -97,6 +99,14 @@ def launch_gateway():
gateway = JavaGateway(
gateway_parameters=GatewayParameters(port=gateway_port, auto_convert=True))
+ return gateway
+
+
+def import_flink_view(gateway):
+ """
+ import the classes used by PyFlink.
+ :param gateway:gateway connected to JavaGateWayServer
+ """
# Import the classes used by PyFlink
java_import(gateway.jvm, "org.apache.flink.table.api.*")
java_import(gateway.jvm, "org.apache.flink.table.api.java.*")
@@ -109,5 +119,3 @@ def launch_gateway():
java_import(gateway.jvm, "org.apache.flink.api.java.ExecutionEnvironment")
java_import(gateway.jvm,
"org.apache.flink.streaming.api.environment.StreamExecutionEnvironment")
-
- return gateway
diff --git a/flink-python/pyflink/table/examples/batch/__init__.py b/flink-python/pyflink/table/examples/batch/__init__.py
new file mode 100644
index 0000000..65b48d4
--- /dev/null
+++ b/flink-python/pyflink/table/examples/batch/__init__.py
@@ -0,0 +1,17 @@
+################################################################################
+# 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.
+################################################################################
diff --git a/flink-python/pyflink/table/examples/batch/word_count.py b/flink-python/pyflink/table/examples/batch/word_count.py
new file mode 100644
index 0000000..a324af4
--- /dev/null
+++ b/flink-python/pyflink/table/examples/batch/word_count.py
@@ -0,0 +1,79 @@
+################################################################################
+# 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.
+################################################################################
+import os
+import tempfile
+
+from pyflink.table import TableEnvironment, TableConfig
+from pyflink.table.table_sink import CsvTableSink
+from pyflink.table.table_source import CsvTableSource
+from pyflink.table.types import DataTypes
+
+
+# TODO: the word_count.py is just a test example for CLI.
+# After pyflink have aligned Java Table API Connectors, this example will be improved.
+def word_count():
+ tmp_dir = tempfile.gettempdir()
+ source_path = tmp_dir + '/streaming.csv'
+ if os.path.isfile(source_path):
+ os.remove(source_path)
+ content = "line Licensed to the Apache Software Foundation ASF under one " \
+ "line or more contributor license agreements See the NOTICE file " \
+ "line distributed with this work for additional information " \
+ "line regarding copyright ownership The ASF licenses this file " \
+ "to you under the Apache License Version the " \
+ "License you may not use this file except in compliance " \
+ "with the License"
+
+ with open(source_path, 'w') as f:
+ for word in content.split(" "):
+ f.write(",".join([word, "1"]))
+ f.write("\n")
+ f.flush()
+ f.close()
+
+ t_config = TableConfig.Builder().as_batch_execution().set_parallelism(1).build()
+ t_env = TableEnvironment.create(t_config)
+
+ field_names = ["word", "cout"]
+ field_types = [DataTypes.STRING, DataTypes.LONG]
+
+ # register Orders table in table environment
+ t_env.register_table_source(
+ "Word",
+ CsvTableSource(source_path, field_names, field_types))
+
+ # register Results table in table environment
+ tmp_dir = tempfile.gettempdir()
+ tmp_csv = tmp_dir + '/streaming2.csv'
+ if os.path.isfile(tmp_csv):
+ os.remove(tmp_csv)
+
+ t_env.register_table_sink(
+ "Results",
+ field_names, field_types, CsvTableSink(tmp_csv))
+
+ t_env.scan("Word") \
+ .group_by("word") \
+ .select("word, count(1) as count") \
+ .insert_into("Results")
+
+ t_env.execute()
+
+
+if __name__ == '__main__':
+ word_count()