This is an automated email from the ASF dual-hosted git repository.
zihaoxiang 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 728cb78801 [Improvement-17010][datax] Add datax channel count to a
custom parameter. (#17898)
728cb78801 is described below
commit 728cb7880195ce72aa6182911bd7b30f671f7708
Author: 墨水记忆 <[email protected]>
AuthorDate: Tue Mar 17 15:44:17 2026 +0800
[Improvement-17010][datax] Add datax channel count to a custom parameter.
(#17898)
---
.../plugin/task/datax/DataxParameters.java | 48 ++++++++++++++--------
.../plugin/task/datax/DataxTask.java | 5 ++-
.../plugin/task/datax/DataxParametersTest.java | 2 +
dolphinscheduler-ui/src/locales/en_US/project.ts | 1 +
dolphinscheduler-ui/src/locales/zh_CN/project.ts | 1 +
.../task/components/node/fields/use-datax.ts | 30 ++++++++++++++
.../projects/task/components/node/format-data.ts | 1 +
.../views/projects/task/components/node/types.ts | 1 +
8 files changed, 70 insertions(+), 19 deletions(-)
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParameters.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParameters.java
index 70e774ea83..0609bb4bcd 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParameters.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParameters.java
@@ -95,6 +95,11 @@ public class DataxParameters extends AbstractParameters {
*/
private int jobSpeedRecord;
+ /**
+ * datax channel
+ */
+ private int jobChannel;
+
/**
* Xms memory
*/
@@ -206,6 +211,14 @@ public class DataxParameters extends AbstractParameters {
this.jobSpeedRecord = jobSpeedRecord;
}
+ public int getJobChannel() {
+ return jobChannel;
+ }
+
+ public void setJobChannel(int jobChannel) {
+ this.jobChannel = jobChannel;
+ }
+
public int getXms() {
return xms;
}
@@ -249,23 +262,24 @@ public class DataxParameters extends AbstractParameters {
@Override
public String toString() {
- return "DataxParameters{"
- + "customConfig=" + customConfig
- + ", json='" + json + '\''
- + ", dsType='" + dsType + '\''
- + ", dataSource=" + dataSource
- + ", dtType='" + dtType + '\''
- + ", dataTarget=" + dataTarget
- + ", sql='" + sql + '\''
- + ", targetTable='" + targetTable + '\''
- + ", preStatements=" + preStatements
- + ", postStatements=" + postStatements
- + ", jobSpeedByte=" + jobSpeedByte
- + ", jobSpeedRecord=" + jobSpeedRecord
- + ", xms=" + xms
- + ", xmx=" + xmx
- + ", resourceList=" + JSONUtils.toJsonString(resourceList)
- + '}';
+ return "DataxParameters{" +
+ "customConfig=" + customConfig +
+ ", json='" + json + '\'' +
+ ", dsType='" + dsType + '\'' +
+ ", dataSource=" + dataSource +
+ ", dtType='" + dtType + '\'' +
+ ", dataTarget=" + dataTarget +
+ ", sql='" + sql + '\'' +
+ ", targetTable='" + targetTable + '\'' +
+ ", preStatements=" + preStatements +
+ ", postStatements=" + postStatements +
+ ", jobSpeedByte=" + jobSpeedByte +
+ ", jobSpeedRecord=" + jobSpeedRecord +
+ ", jobChannel=" + jobChannel +
+ ", xms=" + xms +
+ ", xmx=" + xmx +
+ ", resourceList=" + JSONUtils.toJsonString(resourceList) +
+ '}';
}
@Override
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxTask.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxTask.java
index fd173eb4ef..d9e0bd27fe 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxTask.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxTask.java
@@ -55,6 +55,7 @@ import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
@@ -309,7 +310,7 @@ public class DataxTask extends AbstractTask {
ObjectNode speed = JSONUtils.createObjectNode();
- speed.put("channel", DATAX_CHANNEL_COUNT);
+ speed.put("channel",
Optional.of(dataXParameters.getJobChannel()).orElse(DATAX_CHANNEL_COUNT));
if (dataXParameters.getJobSpeedByte() > 0) {
speed.put("byte", dataXParameters.getJobSpeedByte());
@@ -333,7 +334,7 @@ public class DataxTask extends AbstractTask {
private ObjectNode buildDataxCoreJson() {
ObjectNode speed = JSONUtils.createObjectNode();
- speed.put("channel", DATAX_CHANNEL_COUNT);
+ speed.put("channel",
Optional.of(dataXParameters.getJobChannel()).orElse(DATAX_CHANNEL_COUNT));
if (dataXParameters.getJobSpeedByte() > 0) {
speed.put("byte", dataXParameters.getJobSpeedByte());
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/test/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParametersTest.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/test/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParametersTest.java
index 821668ae8f..a20b5a1695 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/test/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParametersTest.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/test/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParametersTest.java
@@ -70,6 +70,7 @@ public class DataxParametersTest {
dataxParameters.setDtType("MYSQL");
dataxParameters.setJobSpeedByte(1);
dataxParameters.setJobSpeedRecord(1);
+ dataxParameters.setJobChannel(1);
dataxParameters.setJson("json");
dataxParameters.setResourceList(resourceInfoList);
@@ -87,6 +88,7 @@ public class DataxParametersTest {
+ "postStatements=null, "
+ "jobSpeedByte=1, "
+ "jobSpeedRecord=1, "
+ + "jobChannel=1, "
+ "xms=0, "
+ "xmx=-100, "
+
"resourceList=[{\"id\":null,\"resourceName\":\"/hdfs.keytab\",\"res\":null}]"
diff --git a/dolphinscheduler-ui/src/locales/en_US/project.ts
b/dolphinscheduler-ui/src/locales/en_US/project.ts
index 59edc7dfea..667153f215 100644
--- a/dolphinscheduler-ui/src/locales/en_US/project.ts
+++ b/dolphinscheduler-ui/src/locales/en_US/project.ts
@@ -648,6 +648,7 @@ export default {
datax_job_speed_byte_info: '(0 means unlimited)',
datax_job_speed_record: 'Speed(Record count)',
datax_job_speed_record_info: '(0 means unlimited)',
+ datax_job_channel: 'datax channel',
datax_job_runtime_memory: 'Runtime Memory Limits',
datax_job_runtime_memory_xms: 'Low Limit Value',
datax_job_runtime_memory_xmx: 'High Limit Value',
diff --git a/dolphinscheduler-ui/src/locales/zh_CN/project.ts
b/dolphinscheduler-ui/src/locales/zh_CN/project.ts
index e6b08093c5..ef9634095d 100644
--- a/dolphinscheduler-ui/src/locales/zh_CN/project.ts
+++ b/dolphinscheduler-ui/src/locales/zh_CN/project.ts
@@ -629,6 +629,7 @@ export default {
datax_job_speed_byte_info: '(KB，0代表不限制)',
datax_job_speed_record: '限流(记录数)',
datax_job_speed_record_info: '(0代表不限制)',
+ datax_job_channel: '数据管道数',
datax_job_runtime_memory: '运行内存',
datax_job_runtime_memory_xms: '最小内存',
datax_job_runtime_memory_xmx: '最大内存',
diff --git
a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-datax.ts
b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-datax.ts
index b390d8891a..cc38bca9b4 100644
---
a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-datax.ts
+++
b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-datax.ts
@@ -78,6 +78,28 @@ export function useDataX(model: { [field: string]: any }):
IJsonItem[] {
value: 3000
}
]
+ const jobChannelOptions: any[] = [
+ {
+ label: '1',
+ value: 1
+ },
+ {
+ label: '3',
+ value: 3
+ },
+ {
+ label: '5',
+ value: 5
+ },
+ {
+ label: '10',
+ value: 10
+ },
+ {
+ label: '15',
+ value: 15
+ }
+ ]
const memoryLimitOptions = [
{
label: '1G',
@@ -264,6 +286,14 @@ export function useDataX(model: { [field: string]: any }):
IJsonItem[] {
options: memoryLimitOptions,
value: 1
},
+ {
+ type: 'input',
+ field: 'jobChannel',
+ name: t('project.node.datax_job_channel'),
+ span: jobSpeedSpan,
+ options: jobChannelOptions,
+ value: 1
+ },
...useCustomParams({ model, field: 'localParams', isSimple: true })
]
}
diff --git
a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
index 8fc6790677..974da0b44c 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
@@ -274,6 +274,7 @@ export function formatParams(data: INodeData): {
taskParams.targetTable = data.targetTable
taskParams.jobSpeedByte = data.jobSpeedByte
taskParams.jobSpeedRecord = data.jobSpeedRecord
+ taskParams.jobChannel = data.jobChannel
taskParams.preStatements = data.preStatements
taskParams.postStatements = data.postStatements
} else {
diff --git
a/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts
b/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts
index 76dcde23ee..b576ca8f1a 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts
@@ -340,6 +340,7 @@ interface ITaskParams {
targetTable?: string
jobSpeedByte?: number
jobSpeedRecord?: number
+ jobChannel?: number
xms?: number
xmx?: number
sparkParameters?: ISparkParameters
