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/incubator-dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 24fda3f [Feature][Flink] Support parallelism input (#4937)
24fda3f is described below
commit 24fda3fbb1cdf4a58b519feb85e7613b15804dd2
Author: Shiwen Cheng <[email protected]>
AuthorDate: Wed Mar 3 21:15:20 2021 +0800
[Feature][Flink] Support parallelism input (#4937)
* [Feature][Flink] Support parallelism input
* [Feature][Flink] Fix check style
* [Feature][Flink] Remove FLINK_DETACH
---
.../apache/dolphinscheduler/common/Constants.java | 7 +-
.../common/task/flink/FlinkParameters.java | 386 +++++++++++----------
.../server/utils/FlinkArgsUtils.java | 19 +-
.../server/utils/FlinkArgsUtilsTest.java | 32 +-
.../pages/dag/_source/formModel/tasks/flink.vue | 26 +-
.../src/js/module/i18n/locale/en_US.js | 2 +
.../src/js/module/i18n/locale/zh_CN.js | 2 +
7 files changed, 264 insertions(+), 210 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 0297ae6..4c43c0c 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
@@ -842,7 +842,9 @@ public final class Constants {
*/
public static final String HIVE_CONF = "hiveconf:";
- //flink ??
+ /**
+ * flink
+ */
public static final String FLINK_YARN_CLUSTER = "yarn-cluster";
public static final String FLINK_RUN_MODE = "-m";
public static final String FLINK_YARN_SLOT = "-ys";
@@ -852,8 +854,9 @@ public final class Constants {
public static final String FLINK_JOB_MANAGE_MEM = "-yjm";
public static final String FLINK_TASK_MANAGE_MEM = "-ytm";
- public static final String FLINK_DETACH = "-d";
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 int[] NOT_TERMINATED_STATES = new int[] {
diff --git
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/flink/FlinkParameters.java
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/flink/FlinkParameters.java
index 1b1f0a6..76d68c0 100644
---
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/flink/FlinkParameters.java
+++
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/flink/FlinkParameters.java
@@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.common.task.flink;
import org.apache.dolphinscheduler.common.enums.ProgramType;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.AbstractParameters;
-import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
@@ -29,201 +29,213 @@ import java.util.List;
*/
public class FlinkParameters extends AbstractParameters {
- /**
- * major jar
- */
- private ResourceInfo mainJar;
-
- /**
- * major class
- */
- private String mainClass;
-
- /**
- * deploy mode yarn-cluster yarn-client yarn-local
- */
- private String deployMode;
-
- /**
- * arguments
- */
- private String mainArgs;
-
- /**
- * slot count
- */
- private int slot;
-
- /**
- *Yarn application name
- */
-
- private String appName;
-
- /**
- * taskManager count
- */
- private int taskManager;
-
- /**
- * job manager memory
- */
- private String jobManagerMemory ;
-
- /**
- * task manager memory
- */
- private String taskManagerMemory;
-
- /**
- * resource list
- */
- private List<ResourceInfo> resourceList = new ArrayList<>();
-
- /**
- * The YARN queue to submit to
- */
- private String queue;
-
- /**
- * other arguments
- */
- private String others;
-
- /**
- * flink version
- */
- private String flinkVersion;
-
- /**
- * program type
- * 0 JAVA,1 SCALA,2 PYTHON
- */
- private ProgramType programType;
-
- public ResourceInfo getMainJar() {
- return mainJar;
- }
-
- public void setMainJar(ResourceInfo mainJar) {
- this.mainJar = mainJar;
- }
-
- public String getMainClass() {
- return mainClass;
- }
-
- public void setMainClass(String mainClass) {
- this.mainClass = mainClass;
- }
-
- public String getDeployMode() {
- return deployMode;
- }
-
- public void setDeployMode(String deployMode) {
- this.deployMode = deployMode;
- }
-
- public String getMainArgs() {
- return mainArgs;
- }
-
- public void setMainArgs(String mainArgs) {
- this.mainArgs = mainArgs;
- }
-
- public int getSlot() {
- return slot;
- }
-
- public void setSlot(int slot) {
- this.slot = slot;
- }
-
- public String getAppName() {
- return appName;
- }
-
- public void setAppName(String appName) {
- this.appName = appName;
- }
-
- public int getTaskManager() {
- return taskManager;
- }
-
- public void setTaskManager(int taskManager) {
- this.taskManager = taskManager;
- }
-
- public String getJobManagerMemory() {
- return jobManagerMemory;
- }
-
- public void setJobManagerMemory(String jobManagerMemory) {
- this.jobManagerMemory = jobManagerMemory;
- }
-
- public String getTaskManagerMemory() {
- return taskManagerMemory;
- }
-
- public void setTaskManagerMemory(String taskManagerMemory) {
- this.taskManagerMemory = taskManagerMemory;
- }
-
- public String getQueue() {
- return queue;
- }
-
- public void setQueue(String queue) {
- this.queue = queue;
- }
-
- public List<ResourceInfo> getResourceList() {
- return resourceList;
- }
+ /**
+ * major jar
+ */
+ private ResourceInfo mainJar;
+
+ /**
+ * major class
+ */
+ private String mainClass;
+
+ /**
+ * deploy mode yarn-cluster yarn-local
+ */
+ private String deployMode;
+
+ /**
+ * arguments
+ */
+ private String mainArgs;
+
+ /**
+ * slot count
+ */
+ private int slot;
+
+ /**
+ * parallelism
+ */
+ private int parallelism;
+
+ /**
+ * yarn application name
+ */
+ private String appName;
+
+ /**
+ * taskManager count
+ */
+ private int taskManager;
+
+ /**
+ * job manager memory
+ */
+ private String jobManagerMemory;
+
+ /**
+ * task manager memory
+ */
+ private String taskManagerMemory;
+
+ /**
+ * resource list
+ */
+ private List<ResourceInfo> resourceList = new ArrayList<>();
+
+ /**
+ * The YARN queue to submit to
+ */
+ private String queue;
+
+ /**
+ * other arguments
+ */
+ private String others;
+
+ /**
+ * flink version
+ */
+ private String flinkVersion;
+
+ /**
+ * program type
+ * 0 JAVA,1 SCALA,2 PYTHON
+ */
+ private ProgramType programType;
+
+ public ResourceInfo getMainJar() {
+ return mainJar;
+ }
+
+ public void setMainJar(ResourceInfo mainJar) {
+ this.mainJar = mainJar;
+ }
+
+ public String getMainClass() {
+ return mainClass;
+ }
+
+ public void setMainClass(String mainClass) {
+ this.mainClass = mainClass;
+ }
+
+ public String getDeployMode() {
+ return deployMode;
+ }
+
+ public void setDeployMode(String deployMode) {
+ this.deployMode = deployMode;
+ }
+
+ public String getMainArgs() {
+ return mainArgs;
+ }
+
+ public void setMainArgs(String mainArgs) {
+ this.mainArgs = mainArgs;
+ }
+
+ public int getSlot() {
+ return slot;
+ }
+
+ public void setSlot(int slot) {
+ this.slot = slot;
+ }
+
+ public int getParallelism() {
+ return parallelism;
+ }
+
+ public void setParallelism(int parallelism) {
+ this.parallelism = parallelism;
+ }
+
+ public String getAppName() {
+ return appName;
+ }
+
+ public void setAppName(String appName) {
+ this.appName = appName;
+ }
+
+ public int getTaskManager() {
+ return taskManager;
+ }
- public void setResourceList(List<ResourceInfo> resourceList) {
- this.resourceList = resourceList;
- }
+ public void setTaskManager(int taskManager) {
+ this.taskManager = taskManager;
+ }
- public String getOthers() {
- return others;
- }
+ public String getJobManagerMemory() {
+ return jobManagerMemory;
+ }
+
+ public void setJobManagerMemory(String jobManagerMemory) {
+ this.jobManagerMemory = jobManagerMemory;
+ }
- public void setOthers(String others) {
- this.others = others;
- }
+ public String getTaskManagerMemory() {
+ return taskManagerMemory;
+ }
- public ProgramType getProgramType() {
- return programType;
- }
+ public void setTaskManagerMemory(String taskManagerMemory) {
+ this.taskManagerMemory = taskManagerMemory;
+ }
- public void setProgramType(ProgramType programType) {
- this.programType = programType;
- }
+ public String getQueue() {
+ return queue;
+ }
- public String getFlinkVersion() {
- return flinkVersion;
- }
+ public void setQueue(String queue) {
+ this.queue = queue;
+ }
- public void setFlinkVersion(String flinkVersion) {
- this.flinkVersion = flinkVersion;
- }
+ public List<ResourceInfo> getResourceList() {
+ return resourceList;
+ }
- @Override
- public boolean checkParameters() {
- return mainJar != null && programType != null;
- }
+ public void setResourceList(List<ResourceInfo> resourceList) {
+ this.resourceList = resourceList;
+ }
+ public String getOthers() {
+ return others;
+ }
- @Override
- public List<ResourceInfo> getResourceFilesList() {
- if (mainJar != null && !resourceList.contains(mainJar)) {
- resourceList.add(mainJar);
+ public void setOthers(String others) {
+ this.others = others;
}
- return resourceList;
- }
+
+ public ProgramType getProgramType() {
+ return programType;
+ }
+
+ public void setProgramType(ProgramType programType) {
+ this.programType = programType;
+ }
+
+ public String getFlinkVersion() {
+ return flinkVersion;
+ }
+
+ public void setFlinkVersion(String flinkVersion) {
+ this.flinkVersion = flinkVersion;
+ }
+
+ @Override
+ public boolean checkParameters() {
+ return mainJar != null && programType != null;
+ }
+
+ @Override
+ public List<ResourceInfo> getResourceFilesList() {
+ if (mainJar != null && !resourceList.contains(mainJar)) {
+ resourceList.add(mainJar);
+ }
+ return resourceList;
+ }
+
}
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 519ddf2..2d5198c 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
@@ -17,11 +17,11 @@
package org.apache.dolphinscheduler.server.utils;
-import org.apache.commons.lang.StringUtils;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ProgramType;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.flink.FlinkParameters;
+import org.apache.dolphinscheduler.common.utils.StringUtils;
import java.util.ArrayList;
import java.util.List;
@@ -53,7 +53,7 @@ public class FlinkArgsUtils {
args.add(Constants.FLINK_YARN_CLUSTER); //yarn-cluster
int slot = param.getSlot();
- if (slot != 0) {
+ if (slot > 0) {
args.add(Constants.FLINK_YARN_SLOT);
args.add(String.format("%d", slot)); //-ys
}
@@ -68,7 +68,7 @@ public class FlinkArgsUtils {
String flinkVersion = param.getFlinkVersion();
if (flinkVersion == null ||
FLINK_VERSION_BEFORE_1_10.equals(flinkVersion)) {
int taskManager = param.getTaskManager();
- if (taskManager != 0) { //-yn
+ if (taskManager > 0) { //-yn
args.add(Constants.FLINK_TASK_MANAGE);
args.add(String.format("%d", taskManager));
}
@@ -92,12 +92,19 @@ public class FlinkArgsUtils {
args.add(queue);
}
}
+ }
- args.add(Constants.FLINK_DETACH); //-d
-
+ int parallelism = param.getParallelism();
+ if (parallelism > 0) {
+ args.add(Constants.FLINK_PARALLELISM);
+ args.add(String.format("%d", parallelism)); // -p
}
- // -p -s -yqu -yat -sae -yD -D
+ // If the job is submitted in attached mode, perform a best-effort
cluster shutdown when the CLI is terminated abruptly
+ // The task status will be synchronized with the cluster job status
+ args.add(Constants.FLINK_SHUTDOWN_ON_ATTACHED_EXIT); // -sae
+
+ // -s -yqu -yat -yD -D
if (StringUtils.isNotEmpty(others)) {
args.add(others);
}
diff --git
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java
index bea6775..f030628 100644
---
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java
+++
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java
@@ -17,19 +17,20 @@
package org.apache.dolphinscheduler.server.utils;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
import org.apache.dolphinscheduler.common.enums.ProgramType;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.flink.FlinkParameters;
+
+import java.util.List;
+
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.List;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertSame;
-
/**
* Test FlinkArgsUtils
*/
@@ -39,6 +40,7 @@ public class FlinkArgsUtilsTest {
public String mode = "cluster";
public int slot = 2;
+ public int parallelism = 3;
public String appName = "testFlink";
public int taskManager = 4;
public String taskManagerMemory = "2G";
@@ -48,7 +50,7 @@ public class FlinkArgsUtilsTest {
public ResourceInfo mainJar = null;
public String mainArgs = "testArgs --input file:///home";
public String queue = "queue1";
- public String others = "-p 4";
+ public String others = "-s hdfs:///flink/savepoint-1537";
public String flinkVersion = "<1.10";
@@ -72,6 +74,7 @@ public class FlinkArgsUtilsTest {
param.setMainClass(mainClass);
param.setAppName(appName);
param.setSlot(slot);
+ param.setParallelism(parallelism);
param.setTaskManager(taskManager);
param.setJobManagerMemory(jobManagerMemory);
param.setTaskManagerMemory(taskManagerMemory);
@@ -89,7 +92,7 @@ public class FlinkArgsUtilsTest {
}
//Expected values and order
- assertEquals(20, result.size());
+ assertEquals(22, result.size());
assertEquals("-m", result.get(0));
assertEquals("yarn-cluster", result.get(1));
@@ -112,15 +115,18 @@ public class FlinkArgsUtilsTest {
assertEquals("-yqu", result.get(12));
assertEquals(result.get(13),queue);
- assertEquals("-d", result.get(14));
+ assertEquals("-p", result.get(14));
+ assertSame(Integer.valueOf(result.get(15)),parallelism);
+
+ assertEquals("-sae", result.get(16));
- assertEquals(result.get(15),others);
+ assertEquals(result.get(17),others);
- assertEquals("-c", result.get(16));
- assertEquals(result.get(17),mainClass);
+ assertEquals("-c", result.get(18));
+ assertEquals(result.get(19),mainClass);
- assertEquals(result.get(18),mainJar.getRes());
- assertEquals(result.get(19),mainArgs);
+ assertEquals(result.get(20),mainJar.getRes());
+ assertEquals(result.get(21),mainArgs);
//Others param without -yqu
FlinkParameters param1 = new FlinkParameters();
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/flink.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/flink.vue
index d0dfdf6..d328c5f 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/flink.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/flink.vue
@@ -80,7 +80,7 @@
</el-select>
</div>
</m-list-box>
- <m-list-4-box v-if="deployMode === 'cluster'">
+ <m-list-box v-if="deployMode === 'cluster'">
<div slot="text">{{$t('App Name')}}</div>
<div slot="content">
<el-input
@@ -91,7 +91,7 @@
:placeholder="$t('Please enter app name(optional)')">
</el-input>
</div>
- </m-list-4-box>
+ </m-list-box>
<m-list-4-box v-if="deployMode === 'cluster'">
<div slot="text">{{$t('JobManager Memory')}}</div>
<div slot="content">
@@ -136,6 +136,18 @@
</el-input>
</div>
</m-list-4-box>
+ <m-list-4-box>
+ <div slot="text">{{$t('Parallelism')}}</div>
+ <div slot="content">
+ <el-input
+ :disabled="isDetails"
+ type="input"
+ size="small"
+ v-model="parallelism"
+ :placeholder="$t('Please enter Parallelism')">
+ </el-input>
+ </div>
+ </m-list-4-box>
<m-list-box>
<div slot="text">{{$t('Main Arguments')}}</div>
<div slot="content">
@@ -215,6 +227,8 @@
localParams: [],
// Slot number
slot: 1,
+ // Parallelism
+ parallelism: 1,
// TaskManager mumber
taskManager: '2',
// JobManager memory
@@ -320,6 +334,11 @@
return false
}
+ if (!Number.isInteger(parseInt(this.parallelism))) {
+ this.$message.warning(`${i18n.$t('Please enter Parallelism')}`)
+ return false
+ }
+
if (this.flinkVersion === '<1.10' &&
!Number.isInteger(parseInt(this.taskManager))) {
this.$message.warning(`${i18n.$t('Please enter TaskManager
number')}`)
return false
@@ -349,6 +368,7 @@
localParams: this.localParams,
flinkVersion: this.flinkVersion,
slot: this.slot,
+ parallelism: this.parallelism,
taskManager: this.taskManager,
jobManagerMemory: this.jobManagerMemory,
taskManagerMemory: this.taskManagerMemory,
@@ -485,6 +505,7 @@
resourceList: this.resourceIdArr,
localParams: this.localParams,
slot: this.slot,
+ parallelism: this.parallelism,
taskManager: this.taskManager,
jobManagerMemory: this.jobManagerMemory,
taskManagerMemory: this.taskManagerMemory,
@@ -516,6 +537,7 @@
this.deployMode = o.params.deployMode || ''
this.flinkVersion = o.params.flinkVersion || '<1.10'
this.slot = o.params.slot || 1
+ this.parallelism = o.params.parallelism || 1
this.taskManager = o.params.taskManager || '2'
this.jobManagerMemory = o.params.jobManagerMemory || '1G'
this.taskManagerMemory = o.params.taskManagerMemory || '2G'
diff --git a/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
b/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
index cf90cad..3c16553 100755
--- a/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
+++ b/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
@@ -123,6 +123,8 @@ export default {
'Please enter TaskManager memory': 'Please enter TaskManager memory',
'Slot Number': 'Slot Number',
'Please enter Slot number': 'Please enter Slot number',
+ Parallelism: 'Parallelism',
+ 'Please enter Parallelism': 'Please enter Parallelism',
'TaskManager Number': 'TaskManager Number',
'Please enter TaskManager number': 'Please enter TaskManager number',
'App Name': 'App Name',
diff --git a/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
b/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
index c7e87de..22aef01 100755
--- a/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
+++ b/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
@@ -123,6 +123,8 @@ export default {
'Please enter TaskManager memory': '请输入TaskManager内存数',
'Slot Number': 'Slot数量',
'Please enter Slot number': '请输入Slot数量',
+ Parallelism: '并行度',
+ 'Please enter Parallelism': '请输入并行度',
'TaskManager Number': 'TaskManager数量',
'Please enter TaskManager number': '请输入TaskManager数量',
'App Name': '任务名称',
