[dolphinscheduler] branch dev updated: [Improvement-5452][Task] ds flink task support submit a PyFlink job via the CLI (#5453)

[kirs]

This is an automated email from the ASF dual-hosted git repository.

kirs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git


The following commit(s) were added to refs/heads/dev by this push:
     new 3026f04  [Improvement-5452][Task] ds flink task support submit a 
PyFlink job via the CLI (#5453)
3026f04 is described below

commit 3026f04d8528a63f26d9b62da00a495c8e9f47ab
Author: zhuangchong <37063904+zhuangch...@users.noreply.github.com>
AuthorDate: Thu Jun 17 15:19:25 2021 +0800

    [Improvement-5452][Task] ds flink task support submit a PyFlink job via the 
CLI (#5453)
    
    * flink task support submit a PyFlink job via the CLI.
    
    * optimize attribute name.
    
    * Modify pyflink parameter judgment logic
---
 .../java/org/apache/dolphinscheduler/common/Constants.java   |  1 +
 .../apache/dolphinscheduler/server/utils/FlinkArgsUtils.java | 12 +++++++++---
 2 files changed, 10 insertions(+), 3 deletions(-)

diff --git 
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
 
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
index 59e65c5..d9e44c5 100644
--- 
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
+++ 
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
@@ -863,6 +863,7 @@ public final class Constants {
     public static final String FLINK_MAIN_CLASS = "-c";
     public static final String FLINK_PARALLELISM = "-p";
     public static final String FLINK_SHUTDOWN_ON_ATTACHED_EXIT = "-sae";
+    public static final String FLINK_PYTHON = "-py";
 
 
     public static final int[] NOT_TERMINATED_STATES = new int[] {
diff --git 
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java
 
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java
index dbd92e0..e12ebfc 100644
--- 
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java
+++ 
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java
@@ -117,9 +117,15 @@ public class FlinkArgsUtils {
 
         ProgramType programType = param.getProgramType();
         String mainClass = param.getMainClass();
-        if (programType != null && programType != ProgramType.PYTHON && 
StringUtils.isNotEmpty(mainClass)) {
-            args.add(Constants.FLINK_MAIN_CLASS);    //-c
-            args.add(param.getMainClass());          //main class
+
+        if (ProgramType.PYTHON == programType) {
+            // -py
+            args.add(Constants.FLINK_PYTHON);
+        } else if (programType != null && StringUtils.isNotEmpty(mainClass)) {
+            // -c
+            args.add(Constants.FLINK_MAIN_CLASS);
+            // main class
+            args.add(param.getMainClass());
         }
 
         ResourceInfo mainJar = param.getMainJar();