natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cherry-pick fix: Save task seatunnel error #14129

This commit is contained in:
Jay Chung 2023-05-19 10:47:16 +08:00 committed by zhuangchong
parent 862b32ec63
commit 2c87e31274
17 changed files with 162 additions and 229 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/docs/en/guide/task/seatunnel.md
View File

@ -16,7 +16,7 @@ Click [here](https://seatunnel.apache.org/) for more information about `Apache S
[//]: # (- Please refer to [DolphinScheduler Task Parameters Appendix](appendix.md#default-task-parameters) `Default Task Parameters` section for default parameters.)
- Please refer to [DolphinScheduler Task Parameters Appendix](appendix.md) `Default Task Parameters` section for default parameters.
- Engine: Supports FLINK, SPARK AND SEATUNNEL_ENGINE
- Startup script: Select script name to start the task, including `seatunnel.sh`, `start-seatunnel-flink-13-connector-v2.sh`, `start-seatunnel-flink-15-connector-v2.sh`, `start-seatunnel-flink-connector-v2.sh`, `start-seatunnel-flink.sh`, `start-seatunnel-spark-2-connector-v2.sh`, `start-seatunnel-spark-3-connector-v2.sh`, `start-seatunnel-spark-connector-v2.sh`, `start-seatunnel-spark.sh`
- FLINK
- Run model: supports `run` and `run-application` modes
- Option parameters: used to add the parameters of the Flink engine, such as `-m yarn-cluster -ynm seatunnel`

2
docs/docs/zh/guide/task/seatunnel.md
View File

@ -16,7 +16,7 @@
[//]: # (- 默认参数说明请参考[DolphinScheduler任务参数附录](appendix.md#默认任务参数)`默认任务参数`一栏。)
- 默认参数说明请参考[DolphinScheduler任务参数附录](appendix.md)`默认任务参数`一栏。
- 引擎：支持 FLINK 、 SPARK 和 SEATUNNEL_ENGINE
- 启动脚本：选择你想要运行任务的启动脚本，包括 `seatunnel.sh`, `start-seatunnel-flink-13-connector-v2.sh`, `start-seatunnel-flink-15-connector-v2.sh`, `start-seatunnel-flink-connector-v2.sh`, `start-seatunnel-flink.sh`, `start-seatunnel-spark-2-connector-v2.sh`, `start-seatunnel-spark-3-connector-v2.sh`, `start-seatunnel-spark-connector-v2.sh`, `start-seatunnel-spark.sh`
- FLINK
- 运行模型：支持 `run` 和 `run-application` 两种模式
- 选项参数：用于添加 Flink 引擎本身参数，例如 `-m yarn-cluster -ynm seatunnel`

4
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/Constants.java
View File

@ -26,6 +26,8 @@ public class Constants {
public static final String CONFIG_OPTIONS = "--config";
public static final String DEPLOY_MODE_OPTIONS = "--deploy-mode";
public static final String MASTER_OPTIONS = "--master";
public static final String QUEUE_OPTIONS = "--queue";
public static final String STARTUP_SCRIPT_SPARK = "spark";
public static final String STARTUP_SCRIPT_FLINK = "flink";
public static final String STARTUP_SCRIPT_SEATUNNEL = "seatunnel";
}

11
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/DeployModeEnum.java
View File

@ -17,18 +17,19 @@
package org.apache.dolphinscheduler.plugin.task.seatunnel;
import lombok.Getter;
@Getter
public enum DeployModeEnum {
// TODO: use upper case, have to change both frontend and backend code, currently frontend use lower case
cluster("cluster"),
client("client"),
local("local");
private String command;
private final String command;
DeployModeEnum(String command) {
this.command = command;
}
public String getCommand() {
return command;
}
}

40
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/EngineEnum.java
View File

@ -1,40 +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.dolphinscheduler.plugin.task.seatunnel;
public enum EngineEnum {
FLINK("${SEATUNNEL_HOME}/bin/start-seatunnel-flink.sh"),
SPARK("${SEATUNNEL_HOME}/bin/start-seatunnel-spark.sh"),
FLINK_V2("${SEATUNNEL_HOME}/bin/start-seatunnel-flink-connector-v2.sh"),
SPARK_V2("${SEATUNNEL_HOME}/bin/start-seatunnel-spark-connector-v2.sh"),
SEATUNNEL_ENGINE("${SEATUNNEL_HOME}/bin/seatunnel.sh");
private String command;
EngineEnum(String command) {
this.command = command;
}
public String getCommand() {
return command;
}
}

51
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelParameters.java
View File

@ -19,8 +19,6 @@ package org.apache.dolphinscheduler.plugin.task.seatunnel;
import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import org.apache.dolphinscheduler.plugin.task.seatunnel.flink.SeatunnelFlinkParameters;
import org.apache.dolphinscheduler.plugin.task.seatunnel.spark.SeatunnelSparkParameters;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.BooleanUtils;
@ -29,17 +27,16 @@ import org.apache.commons.lang3.StringUtils;
import java.util.List;
import java.util.Objects;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "engine")
@JsonSubTypes({
@JsonSubTypes.Type(value = SeatunnelFlinkParameters.class, name = "FLINK"),
@JsonSubTypes.Type(value = SeatunnelSparkParameters.class, name = "SPARK")
})
@Getter
@Setter
@NoArgsConstructor
public class SeatunnelParameters extends AbstractParameters {
private EngineEnum engine;
private String startupScript;
private Boolean useCustom;
@ -50,41 +47,9 @@ public class SeatunnelParameters extends AbstractParameters {
*/
private List<ResourceInfo> resourceList;
public EngineEnum getEngine() {
return engine;
}
public void setEngine(EngineEnum engine) {
this.engine = engine;
}
public Boolean getUseCustom() {
return useCustom;
}
public void setUseCustom(Boolean useCustom) {
this.useCustom = useCustom;
}
public String getRawScript() {
return rawScript;
}
public void setRawScript(String rawScript) {
this.rawScript = rawScript;
}
public List<ResourceInfo> getResourceList() {
return resourceList;
}
public void setResourceList(List<ResourceInfo> resourceList) {
this.resourceList = resourceList;
}
@Override
public boolean checkParameters() {
return Objects.nonNull(engine)
return Objects.nonNull(startupScript)
&& ((BooleanUtils.isTrue(useCustom) && StringUtils.isNotBlank(rawScript))
|| (BooleanUtils.isFalse(useCustom) && CollectionUtils.isNotEmpty(resourceList) && resourceList.size() == 1));
}

4
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTask.java
View File

@ -50,6 +50,8 @@ import java.util.Map;
*/
public class SeatunnelTask extends AbstractRemoteTask {
private static final String SEATUNNEL_BIN_DIR = "${SEATUNNEL_HOME}/bin/";
/**
* seatunnel parameters
*/
@ -138,7 +140,7 @@ public class SeatunnelTask extends AbstractRemoteTask {
private String buildCommand() throws Exception {
List<String> args = new ArrayList<>();
args.add(seatunnelParameters.getEngine().getCommand());
args.add(SEATUNNEL_BIN_DIR + seatunnelParameters.getStartupScript());
args.addAll(buildOptions());
String command = String.join(" ", args);

25
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTaskChannel.java
View File

@ -17,6 +17,10 @@
package org.apache.dolphinscheduler.plugin.task.seatunnel;
import static org.apache.dolphinscheduler.plugin.task.seatunnel.Constants.STARTUP_SCRIPT_FLINK;
import static org.apache.dolphinscheduler.plugin.task.seatunnel.Constants.STARTUP_SCRIPT_SEATUNNEL;
import static org.apache.dolphinscheduler.plugin.task.seatunnel.Constants.STARTUP_SCRIPT_SPARK;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskChannel;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@ -38,17 +42,18 @@ public class SeatunnelTaskChannel implements TaskChannel {
public SeatunnelTask createTask(TaskExecutionContext taskRequest) {
SeatunnelParameters seatunnelParameters =
JSONUtils.parseObject(taskRequest.getTaskParams(), SeatunnelParameters.class);
switch (seatunnelParameters.getEngine()) {
case FLINK:
case FLINK_V2:
return new SeatunnelFlinkTask(taskRequest);
case SPARK:
case SPARK_V2:
return new SeatunnelSparkTask(taskRequest);
case SEATUNNEL_ENGINE:
return new SeatunnelEngineTask(taskRequest);
assert seatunnelParameters != null;
String startupScript = seatunnelParameters.getStartupScript();
if (startupScript.contains(STARTUP_SCRIPT_SPARK)) {
return new SeatunnelSparkTask(taskRequest);
}
throw new IllegalArgumentException("Unsupported engine type:" + seatunnelParameters.getEngine());
if (startupScript.contains(STARTUP_SCRIPT_FLINK)) {
return new SeatunnelFlinkTask(taskRequest);
}
if (startupScript.contains(STARTUP_SCRIPT_SEATUNNEL)) {
return new SeatunnelEngineTask(taskRequest);
}
throw new IllegalArgumentException("Unsupported startup script name:" + seatunnelParameters.getStartupScript());
}
@Override

33
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/flink/SeatunnelFlinkParameters.java
View File

@ -19,40 +19,29 @@ package org.apache.dolphinscheduler.plugin.task.seatunnel.flink;
import org.apache.dolphinscheduler.plugin.task.seatunnel.SeatunnelParameters;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Getter
@Setter
@NoArgsConstructor
public class SeatunnelFlinkParameters extends SeatunnelParameters {
private RunModeEnum runMode;
private String others;
public static enum RunModeEnum {
@Getter
public enum RunModeEnum {
RUN("--run-mode run"),
RUN_APPLICATION("--run-mode run-application");
private String command;
private final String command;
RunModeEnum(String command) {
this.command = command;
}
public String getCommand() {
return command;
}
}
public RunModeEnum getRunMode() {
return runMode;
}
public void setRunMode(RunModeEnum runMode) {
this.runMode = runMode;
}
public String getOthers() {
return others;
}
public void setOthers(String others) {
this.others = others;
}
}

25
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/self/SeatunnelEngineParameters.java
View File

@ -20,25 +20,18 @@ package org.apache.dolphinscheduler.plugin.task.seatunnel.self;
import org.apache.dolphinscheduler.plugin.task.seatunnel.DeployModeEnum;
import org.apache.dolphinscheduler.plugin.task.seatunnel.SeatunnelParameters;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class SeatunnelEngineParameters extends SeatunnelParameters {
private DeployModeEnum deployMode;
private String others;
public DeployModeEnum getDeployMode() {
return deployMode;
}
public void setDeployMode(DeployModeEnum deployMode) {
this.deployMode = deployMode;
}
public String getOthers() {
return others;
}
public void setOthers(String others) {
this.others = others;
}
}

41
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/spark/SeatunnelSparkParameters.java
View File

@ -24,6 +24,13 @@ import org.apache.commons.lang3.StringUtils;
import java.util.Objects;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Getter
@Setter
@NoArgsConstructor
public class SeatunnelSparkParameters extends SeatunnelParameters {
private DeployModeEnum deployMode;
@ -42,44 +49,18 @@ public class SeatunnelSparkParameters extends SeatunnelParameters {
return result;
}
public static enum MasterTypeEnum {
@Getter
public enum MasterTypeEnum {
YARN("yarn"),
LOCAL("local"),
SPARK("spark://"),
MESOS("mesos://");
private String command;
private final String command;
MasterTypeEnum(String command) {
this.command = command;
}
public String getCommand() {
return command;
}
}
public DeployModeEnum getDeployMode() {
return deployMode;
}
public void setDeployMode(DeployModeEnum deployMode) {
this.deployMode = deployMode;
}
public MasterTypeEnum getMaster() {
return master;
}
public void setMaster(MasterTypeEnum master) {
this.master = master;
}
public String getMasterUrl() {
return masterUrl;
}
public void setMasterUrl(String masterUrl) {
this.masterUrl = masterUrl;
}
}

2
dolphinscheduler-ui/src/locales/en_US/project.ts
View File

@ -759,6 +759,8 @@ export default {
custom_config: 'Custom Config',
engine: 'engine',
engine_tips: 'Please select engine',
startup_script: 'Startup script',
startup_script_tips: 'Please select startup script',
run_mode: 'Run Mode',
dinky_address: 'Dinky address',
dinky_address_tips: 'Please enter the url of your dinky',

2
dolphinscheduler-ui/src/locales/zh_CN/project.ts
View File

@ -740,6 +740,8 @@ export default {
custom_config: '自定义配置',
engine: '引擎',
engine_tips: '请选择引擎',
startup_script: '启动脚本',
startup_script_tips: '请选择启动脚本',
run_mode: '运行模式',
dinky_address: 'dinky 地址',
dinky_address_tips: '请输入 Dinky 地址',

78
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sea-tunnel.ts
View File

@ -24,34 +24,44 @@ export function useSeaTunnel(model: { [field: string]: any }): IJsonItem[] {
const configEditorSpan = computed(() => (model.useCustom ? 24 : 0))
const resourceEditorSpan = computed(() => (model.useCustom ? 0 : 24))
const flinkSpan = computed(() => (model.engine === 'FLINK' || model.engine === 'FLINK_V2' ? 24 : 0))
const deployModeSpan = computed(() => (model.engine === 'SPARK' || model.engine === 'SPARK_V2' || model.engine === "SEATUNNEL_ENGINE" ? 24 : 0))
const masterSpan = computed(() =>
(model.engine === 'SPARK' || model.engine === 'SPARK_V2') && model.deployMode !== 'local' ? 12 : 0
)
const flinkSpan = computed(() => (model.startupScript.includes("flink") ? 24 : 0))
const deployModeSpan = computed(() => (model.startupScript.includes("spark") || model.startupScript === "seatunnel.sh" ? 24 : 0))
const masterSpan = computed(() => (model.startupScript.includes("spark")) && model.deployMode !== 'local' ? 12 : 0)
const masterUrlSpan = computed(() =>
(model.engine === 'SPARK' || model.engine === 'SPARK_V2') &&
(model.startupScript.includes("spark")) &&
model.deployMode !== 'local' &&
(model.master === 'SPARK' || model.master === 'MESOS')
? 12
: 0
)
const showClient = computed(() => model.engine === 'SPARK' || model.engine === 'SPARK_V2')
const showLocal = computed(() => model.engine === 'SEATUNNEL_ENGINE')
const othersSpan = computed(() => (model.engine === 'FLINK' || model.engine === 'FLINK_V2' || model.engine === 'SEATUNNEL_ENGINE' ? 24 : 0))
const showClient = computed(() => model.startupScript.includes("spark"))
const showLocal = computed(() => model.startupScript === 'seatunnel.sh')
const othersSpan = computed(() => (model.startupScript.includes("flink") || model.startupScript === 'seatunnel.sh' ? 24 : 0))
return [
{
type: 'select',
field: 'engine',
span: 12,
name: t('project.node.engine'),
options: ENGINE,
field: 'startupScript',
span: 15,
name: t('project.node.startup_script'),
options: STARTUP_SCRIPT,
validate: {
trigger: ['input', 'blur'],
required: true,
message: t('project.node.engine_tips')
}
message: t('project.node.startup_script_tips')
},
props: {
'on-update:value': (value: boolean) => {
if (value) {
if (model.startupScript === 'seatunnel.sh') {
model.deployMode = 'local'
}
if (model.startupScript.includes("spark")) {
model.deployMode = 'client'
}
}
}
},
},
// SeaTunnel flink parameter
@ -130,26 +140,42 @@ export function useSeaTunnel(model: { [field: string]: any }): IJsonItem[] {
]
}
export const ENGINE = [
export const STARTUP_SCRIPT = [
{
label: 'SPARK',
value: 'SPARK'
label: 'seatunnel.sh',
value: 'seatunnel.sh'
},
{
label: 'FLINK',
value: 'FLINK'
label: 'start-seatunnel-flink-13-connector-v2.sh',
value: 'start-seatunnel-flink-13-connector-v2.sh'
},
{
label: 'SPARK_V2',
value: 'SPARK_V2'
label: 'start-seatunnel-flink-15-connector-v2.sh',
value: 'start-seatunnel-flink-15-connector-v2.sh'
},
{
label: 'FLINK_V2',
value: 'FLINK_V2'
label: 'start-seatunnel-flink-connector-v2.sh',
value: 'start-seatunnel-flink-connector-v2.sh'
},
{
label: 'SEATUNNEL_ENGINE',
value: 'SEATUNNEL_ENGINE'
label: 'start-seatunnel-flink.sh',
value: 'start-seatunnel-flink.sh'
},
{
label: 'start-seatunnel-spark-2-connector-v2.sh',
value: 'start-seatunnel-spark-2-connector-v2.sh'
},
{
label: 'start-seatunnel-spark-3-connector-v2.sh',
value: 'start-seatunnel-spark-3-connector-v2.sh'
},
{
label: 'start-seatunnel-spark-connector-v2.sh',
value: 'start-seatunnel-spark-connector-v2.sh'
},
{
label: 'start-seatunnel-spark.sh',
value: 'start-seatunnel-spark.sh'
}
]

26
dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
View File

@ -200,21 +200,21 @@ export function formatParams(data: INodeData): {
}
if (data.taskType === 'SEATUNNEL') {
taskParams.engine = data.engine
taskParams.startupScript = data.startupScript
taskParams.useCustom = data.useCustom
taskParams.rawScript = data.rawScript
switch (data.engine) {
case 'FLINK':
taskParams.runMode = data.runMode
taskParams.others = data.others
break
case 'SPARK':
taskParams.deployMode = data.deployMode
taskParams.master = data.master
taskParams.masterUrl = data.masterUrl
break
default:
break
if (data.startupScript?.includes("flink")) {
taskParams.runMode = data.runMode
taskParams.others = data.others
}
if (data.startupScript?.includes("spark")) {
taskParams.deployMode = data.deployMode
taskParams.master = data.master
taskParams.masterUrl = data.masterUrl
}
if (data.startupScript === "seatunnel.sh") {
taskParams.deployMode = data.deployMode
taskParams.others = data.others
}
}

44
dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-sea-tunnel.ts
View File

@ -46,7 +46,7 @@ export function useSeaTunnel({
memoryMax: -1,
delayTime: 0,
timeout: 30,
engine: 'FLINK',
startupScript: 'seatunnel.sh',
runMode: 'RUN',
useCustom: true,
deployMode: 'client',
@ -56,25 +56,29 @@ export function useSeaTunnel({
timeoutNotifyStrategy: ['WARN'],
rawScript:
'env {\n' +
' execution.parallelism = 1\n' +
'}\n' +
'\n' +
'source {\n' +
' FakeSourceStream {\n' +
' result_table_name = "fake"\n' +
' field_name = "name,age"\n' +
' }\n' +
'}\n' +
'\n' +
'transform {\n' +
' sql {\n' +
' sql = "select name,age from fake"\n' +
' }\n' +
'}\n' +
'\n' +
'sink {\n' +
' ConsoleSink {}\n' +
'}'
' execution.parallelism = 2\n' +
' job.mode = "BATCH"\n' +
' checkpoint.interval = 10000\n' +
'}\n' +
'\n' +
'source {\n' +
' FakeSource {\n' +
' parallelism = 2\n' +
' result_table_name = "fake"\n' +
' row.num = 16\n' +
' schema = {\n' +
' fields {\n' +
' name = "string"\n' +
' age = "int"\n' +
' }\n' +
' }\n' +
' }\n' +
'}\n' +
'\n' +
'sink {\n' +
' Console {\n' +
' }\n' +
'}'
} as INodeData)
return {

1
dolphinscheduler-ui/src/views/projects/task/components/node/types.ts
View File

@ -329,6 +329,7 @@ interface ITaskParams {
parameters?: string
kernel?: string
engine?: string
startupScript?: string
executionTimeout?: string
startTimeout?: string
processDefinitionCode?: number