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

[Improvement][parameter] Improvement startup parameters/global parameters data type

This commit is contained in:
wuchao 2024-05-08 16:22:49 +08:00
parent c017ac8b5e
commit 8d547d2566
12 changed files with 141 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ExecutorController.java
View File

@ -44,6 +44,7 @@ import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecuteDto;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.commons.lang3.StringUtils;
@ -163,9 +164,9 @@ public class ExecutorController extends BaseController {
if (timeout == null) {
timeout = Constants.MAX_TASK_TIMEOUT;
}
Map<String, String> startParamMap = null;
List<Property> startParamList = null;
if (startParams != null) {
startParamMap = JSONUtils.toMap(startParams);
startParamList = JSONUtils.toList(startParams, Property.class);
}
if (complementDependentMode == null) {
@ -175,7 +176,7 @@ public class ExecutorController extends BaseController {
Map<String, Object> result = execService.execProcessInstance(loginUser, projectCode, processDefinitionCode,
scheduleTime, execType, failureStrategy,
startNodeList, taskDependType, warningType, warningGroupId, runMode, processInstancePriority,
workerGroup, tenantCode, environmentCode, timeout, startParamMap, expectedParallelismNumber, dryRun,
workerGroup, tenantCode, environmentCode, timeout, startParamList, expectedParallelismNumber, dryRun,
testFlag,
complementDependentMode, version, allLevelDependent, executionOrder);
return returnDataList(result);
@ -262,9 +263,9 @@ public class ExecutorController extends BaseController {
timeout = Constants.MAX_TASK_TIMEOUT;
}
Map<String, String> startParamMap = null;
List<Property> startParamList = null;
if (startParams != null) {
startParamMap = JSONUtils.toMap(startParams);
startParamList = JSONUtils.toList(startParams, Property.class);
}
if (complementDependentMode == null) {
@ -283,7 +284,8 @@ public class ExecutorController extends BaseController {
result = execService.execProcessInstance(loginUser, projectCode, processDefinitionCode, scheduleTime,
execType, failureStrategy,
startNodeList, taskDependType, warningType, warningGroupId, runMode, processInstancePriority,
workerGroup, tenantCode, environmentCode, timeout, startParamMap, expectedParallelismNumber, dryRun,
workerGroup, tenantCode, environmentCode, timeout, startParamList, expectedParallelismNumber,
dryRun,
testFlag,
complementDependentMode, null, allLevelDependent, executionOrder);

6
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java
View File

@ -30,7 +30,9 @@ import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecuteDto;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import java.util.List;
import java.util.Map;
/**
@ -57,7 +59,7 @@ public interface ExecutorService {
* @param environmentCode environment code
* @param runMode run mode
* @param timeout timeout
* @param startParams the global param values which pass to new process instance
* @param startParamList the global param values which pass to new process instance
* @param expectedParallelismNumber the expected parallelism number when execute complement in parallel mode
* @param executionOrder the execution order when complementing data
* @return execute process instance code
@ -71,7 +73,7 @@ public interface ExecutorService {
Priority processInstancePriority, String workerGroup, String tenantCode,
Long environmentCode,
Integer timeout,
Map<String, String> startParams, Integer expectedParallelismNumber,
List<Property> startParamList, Integer expectedParallelismNumber,
int dryRun, int testFlag,
ComplementDependentMode complementDependentMode, Integer version,
boolean allLevelDependent, ExecutionOrder executionOrder);

13
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
View File

@ -90,6 +90,7 @@ import org.apache.dolphinscheduler.extract.master.transportor.StreamingTaskTrigg
import org.apache.dolphinscheduler.extract.master.transportor.StreamingTaskTriggerResponse;
import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStateChangeEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.service.command.CommandService;
import org.apache.dolphinscheduler.service.cron.CronUtils;
import org.apache.dolphinscheduler.service.exceptions.CronParseException;
@ -203,7 +204,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
* @param environmentCode environment code
* @param runMode run mode
* @param timeout timeout
* @param startParams the global param values which pass to new process instance
* @param startParamList the global param values which pass to new process instance
* @param expectedParallelismNumber the expected parallelism number when execute complement in parallel mode
* @param testFlag testFlag
* @param executionOrder the execution order when complementing data
@ -219,7 +220,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
Priority processInstancePriority, String workerGroup,
String tenantCode,
Long environmentCode, Integer timeout,
Map<String, String> startParams, Integer expectedParallelismNumber,
List<Property> startParamList, Integer expectedParallelismNumber,
int dryRun, int testFlag,
ComplementDependentMode complementDependentMode, Integer version,
boolean allLevelDependent, ExecutionOrder executionOrder) {
@ -269,7 +270,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
startNodeList,
cronTime, warningType, loginUser.getId(), warningGroupId, runMode, processInstancePriority,
workerGroup, tenantCode,
environmentCode, startParams, expectedParallelismNumber, dryRun, testFlag,
environmentCode, startParamList, expectedParallelismNumber, dryRun, testFlag,
complementDependentMode, allLevelDependent, executionOrder);
if (create > 0) {
@ -731,7 +732,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
WarningType warningType, int executorId, Integer warningGroupId, RunMode runMode,
Priority processInstancePriority, String workerGroup, String tenantCode,
Long environmentCode,
Map<String, String> startParams, Integer expectedParallelismNumber, int dryRun,
List<Property> startParamList, Integer expectedParallelismNumber, int dryRun,
int testFlag, ComplementDependentMode complementDependentMode,
boolean allLevelDependent, ExecutionOrder executionOrder) {
@ -760,8 +761,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
if (warningType != null) {
command.setWarningType(warningType);
}
if (startParams != null && startParams.size() > 0) {
cmdParam.put(CMD_PARAM_START_PARAMS, JSONUtils.toJsonString(startParams));
if (startParamList != null && startParamList.size() > 0) {
cmdParam.put(CMD_PARAM_START_PARAMS, JSONUtils.toJsonString(startParamList));
}
command.setCommandParam(JSONUtils.toJsonString(cmdParam));
command.setExecutorId(executorId);

8
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/ExecuteFunctionControllerTest.java
View File

@ -38,8 +38,13 @@ import org.apache.dolphinscheduler.common.enums.RunMode;
import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;
@ -75,7 +80,8 @@ public class ExecuteFunctionControllerTest extends AbstractControllerTest {
final String tenantCode = "root";
final Long environmentCode = 4L;
final Integer timeout = 5;
final ImmutableMap<String, String> startParams = ImmutableMap.of("start", "params");
final List<Property> startParams =
Collections.singletonList(new Property("start", Direct.IN, DataType.VARCHAR, "params"));
final Integer expectedParallelismNumber = 6;
final int dryRun = 7;
final int testFlag = 0;

26
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceImpl.java
View File

@ -54,6 +54,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
@ -63,6 +64,9 @@ import lombok.NonNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
@Component
public class CuringParamsServiceImpl implements CuringParamsService {
@ -151,8 +155,16 @@ public class CuringParamsServiceImpl implements CuringParamsService {
}
String startParamJson = cmdParam.get(CommandKeyConstants.CMD_PARAM_START_PARAMS);
Map<String, String> startParamMap = JSONUtils.toMap(startParamJson);
return startParamMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
entry -> new Property(entry.getKey(), Direct.IN, DataType.VARCHAR, entry.getValue())));
JsonElement jsonElement = JsonParser.parseString(startParamJson);
// check whether it is json
boolean isJson = jsonElement.isJsonObject();
if (isJson) {
return startParamMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
entry -> new Property(entry.getKey(), Direct.IN, DataType.VARCHAR, entry.getValue())));
} else {
List<Property> propList = JSONUtils.toList(startParamJson, Property.class);
return propList.stream().collect(Collectors.toMap(Property::getProp, Function.identity()));
}
}
@Override
@ -181,8 +193,7 @@ public class CuringParamsServiceImpl implements CuringParamsService {
Map<String, Property> prepareParamsMap = new HashMap<>();
// assign value to definedParams here
Map<String, String> globalParamsMap = setGlobalParamsMap(processInstance);
Map<String, Property> globalParams = ParameterUtils.getUserDefParamsMap(globalParamsMap);
Map<String, Property> globalParams = setGlobalParamsMap(processInstance);
// combining local and global parameters
Map<String, Property> localParams = parameters.getInputLocalParametersMap();
@ -287,15 +298,16 @@ public class CuringParamsServiceImpl implements CuringParamsService {
Long.toString(taskInstance.getProcessInstance().getProcessDefinition().getProjectCode()));
return params;
}
private Map<String, String> setGlobalParamsMap(ProcessInstance processInstance) {
Map<String, String> globalParamsMap = new HashMap<>(16);
private Map<String, Property> setGlobalParamsMap(ProcessInstance processInstance) {
Map<String, Property> globalParamsMap = new HashMap<>(16);
// global params string
String globalParamsStr = processInstance.getGlobalParams();
if (globalParamsStr != null) {
List<Property> globalParamsList = JSONUtils.toList(globalParamsStr, Property.class);
globalParamsMap
.putAll(globalParamsList.stream().collect(Collectors.toMap(Property::getProp, Property::getValue)));
.putAll(globalParamsList.stream()
.collect(Collectors.toMap(Property::getProp, Function.identity())));
}
return globalParamsMap;
}

30
dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag-save-modal.tsx
View File

@ -154,7 +154,8 @@ export default defineComponent({
formValue.value.globalParams = process.globalParamList.map((param) => ({
key: param.prop,
value: param.value,
direct: param.direct
direct: param.direct,
type: param.type
}))
}
}
@ -239,6 +240,7 @@ export default defineComponent({
return {
key: '',
direct: 'IN',
type: 'VARCHAR',
value: ''
}
}}
@ -246,16 +248,16 @@ export default defineComponent({
>
{{
default: (param: {
value: { key: string; direct: string; value: string }
value: { key: string; direct: string; type: string; value: string }
}) => (
<NGrid xGap={12} cols={24}>
<NGridItem span={9}>
<NGridItem span={6}>
<NInput
v-model:value={param.value.key}
placeholder={t('project.dag.key')}
/>
</NGridItem>
<NGridItem span={6}>
<NGridItem span={5}>
<NSelect
options={[
{ value: 'IN', label: 'IN' },
@ -265,7 +267,25 @@ export default defineComponent({
defaultValue={'IN'}
/>
</NGridItem>
<NGridItem span={9}>
<NGridItem span={7}>
<NSelect
options={[
{ value: 'VARCHAR', label: 'VARCHAR' },
{ value: 'INTEGER', label: 'INTEGER' },
{ value: 'LONG', label: 'LONG' },
{ value: 'FLOAT', label: 'FLOAT' },
{ value: 'DOUBLE', label: 'DOUBLE' },
{ value: 'DATE', label: 'DATE' },
{ value: 'TIME', label: 'TIME' },
{ value: 'BOOLEAN', label: 'BOOLEAN' },
{ value: 'LIST', label: 'LIST' },
{ value: 'FILE', label: 'FILE' }
]}
v-model:value={param.value.type}
defaultValue={'VARCHAR'}
/>
</NGridItem>
<NGridItem span={6}>
<NInput
v-model:value={param.value.value}
placeholder={t('project.dag.value')}

1
dolphinscheduler-ui/src/views/projects/workflow/components/dag/types.ts
View File

@ -138,6 +138,7 @@ export interface Coordinate {
export interface GlobalParam {
key: string
direct: string
type: string
value: string
}

110
dolphinscheduler-ui/src/views/projects/workflow/definition/components/start-modal.tsx
View File

@ -45,7 +45,10 @@ import {
NCheckbox,
NDatePicker,
NRadioButton,
NInputNumber
NInputNumber,
NDynamicInput,
NGrid,
NGridItem
} from 'naive-ui'
import {
ArrowDownOutlined,
@ -580,50 +583,67 @@ export default defineComponent({
label={t('project.workflow.startup_parameter')}
path='startup_parameter'
>
{this.startParamsList.length === 0 ? (
<NButton text type='primary' onClick={this.addStartParams}>
<NIcon>
<PlusCircleOutlined />
</NIcon>
</NButton>
) : (
<NSpace vertical>
{this.startParamsList.map((item, index) => (
<NSpace class={styles.startup} key={Date.now() + index}>
<NInput
allowInput={this.trim}
pair
separator=':'
placeholder={['prop', 'value']}
defaultValue={[item.prop, item.value]}
onUpdateValue={(param) =>
this.updateParamsList(index, param)
}
/>
<NButton
text
type='error'
onClick={() => this.removeStartParams(index)}
class='btn-delete-custom-parameter'
>
<NIcon>
<DeleteOutlined />
</NIcon>
</NButton>
<NButton
text
type='primary'
onClick={this.addStartParams}
class='btn-create-custom-parameter'
>
<NIcon>
<PlusCircleOutlined />
</NIcon>
</NButton>
</NSpace>
))}
</NSpace>
)}
<NDynamicInput
v-model:value={this.startParamsList}
onCreate={() => {
return {
key: '',
direct: 'IN',
type: 'VARCHAR',
value: ''
}
}}
class='input-startup-params'
>
{{
default: (param: {
value: { prop: string; direct: string; type: string; value: string }
}) => (
<NGrid xGap={12} cols={24}>
<NGridItem span={6}>
<NInput
v-model:value={param.value.prop}
placeholder={t('project.dag.key')}
/>
</NGridItem>
<NGridItem span={5}>
<NSelect
options={[
{ value: 'IN', label: 'IN' },
{ value: 'OUT', label: 'OUT' }
]}
v-model:value={param.value.direct}
defaultValue={'IN'}
/>
</NGridItem>
<NGridItem span={7}>
<NSelect
options={[
{ value: 'VARCHAR', label: 'VARCHAR' },
{ value: 'INTEGER', label: 'INTEGER' },
{ value: 'LONG', label: 'LONG' },
{ value: 'FLOAT', label: 'FLOAT' },
{ value: 'DOUBLE', label: 'DOUBLE' },
{ value: 'DATE', label: 'DATE' },
{ value: 'TIME', label: 'TIME' },
{ value: 'BOOLEAN', label: 'BOOLEAN' },
{ value: 'LIST', label: 'LIST' },
{ value: 'FILE', label: 'FILE' }
]}
v-model:value={param.value.type}
defaultValue={'VARCHAR'}
/>
</NGridItem>
<NGridItem span={6}>
<NInput
v-model:value={param.value.value}
placeholder={t('project.dag.value')}
/>
</NGridItem>
</NGrid>
)
}}
</NDynamicInput>
</NFormItem>
<NFormItem
label={t('project.workflow.whether_dry_run')}

10
dolphinscheduler-ui/src/views/projects/workflow/definition/components/use-modal.ts
View File

@ -119,14 +119,8 @@ export function useModal(
})
}
const startParams = {} as any
for (const item of variables.startParamsList) {
if (item.value !== '') {
startParams[item.prop] = item.value
}
}
params.startParams = !_.isEmpty(startParams)
? JSON.stringify(startParams)
params.startParams = !_.isEmpty(variables.startParamsList)
? JSON.stringify(variables.startParamsList)
: ''
await startProcessInstance(params, variables.projectCode)
window.$message.success(t('project.workflow.success'))

2
dolphinscheduler-ui/src/views/projects/workflow/definition/create/index.tsx
View File

@ -59,7 +59,7 @@ export default defineComponent({
prop: p.key,
value: p.value,
direct: p.direct,
type: 'VARCHAR'
type: p.type
}
})

2
dolphinscheduler-ui/src/views/projects/workflow/definition/detail/index.tsx
View File

@ -85,7 +85,7 @@ export default defineComponent({
prop: p.key,
value: p.value,
direct: p.direct,
type: 'VARCHAR'
type: p.type
}
})

2
dolphinscheduler-ui/src/views/projects/workflow/instance/detail/index.tsx
View File

@ -82,7 +82,7 @@ export default defineComponent({
prop: p.key,
value: p.value,
direct: p.direct,
type: 'VARCHAR'
type: p.type
}
})