diff --git a/dolphinscheduler-api/src/main/resources/task-type-config.yaml b/dolphinscheduler-api/src/main/resources/task-type-config.yaml index e95d4190fc60..9475e93420cf 100644 --- a/dolphinscheduler-api/src/main/resources/task-type-config.yaml +++ b/dolphinscheduler-api/src/main/resources/task-type-config.yaml @@ -23,6 +23,7 @@ task: - 'PROCEDURE' - 'SQL' - 'SPARK' + - 'FLINK_SGW' - 'FLINK' - 'GRPC' - 'HTTP' diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-all/pom.xml b/dolphinscheduler-task-plugin/dolphinscheduler-task-all/pom.xml index f3e6f8855568..b4a97a851b7a 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-all/pom.xml +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-all/pom.xml @@ -222,6 +222,12 @@ dolphinscheduler-task-aliyunserverlessspark ${project.version} + + + org.apache.dolphinscheduler + dolphinscheduler-task-flink-sqlgateway + ${project.version} + diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/pom.xml b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/pom.xml new file mode 100644 index 000000000000..c360b78209a6 --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/pom.xml @@ -0,0 +1,75 @@ + + + + 4.0.0 + + + org.apache.dolphinscheduler + dolphinscheduler-task-plugin + dev-SNAPSHOT + + + dolphinscheduler-task-flink-sqlgateway + jar + + + task.flink.sqlgateway + 2.2.0 + + + + + org.apache.dolphinscheduler + dolphinscheduler-task-api + ${project.version} + provided + + + + org.apache.dolphinscheduler + dolphinscheduler-common + ${project.version} + provided + + + + + org.apache.flink + flink-sql-jdbc-driver + ${flink-sql-jdbc-driver.version} + + + + + + + org.apache.maven.plugins + maven-shade-plugin + + + + shade + + package + + + + + + diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayParameters.java new file mode 100644 index 000000000000..831934de8ead --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayParameters.java @@ -0,0 +1,202 @@ +/* + * 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.flinksqlgateway; + +import org.apache.dolphinscheduler.plugin.task.api.enums.SqlSourceType; +import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo; +import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters; + +import org.apache.commons.collections4.CollectionUtils; +import org.apache.commons.lang3.StringUtils; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +public class FlinkSqlGatewayParameters extends AbstractParameters { + + /** + * example: jdbc:flink://host:port + */ + private String flinkJdbcUrl; + + /** + * optional: username/password or other jdbc properties + */ + private Map jdbcProperties; + + /** + * init script source: SCRIPT = inline sql text, FILE = sql from resource center file + */ + private SqlSourceType initScriptType = SqlSourceType.FILE; + + /** + * optional init sql script (when initScriptType=SCRIPT), or read-only display (when FILE) + */ + private String initScript; + + /** + * resource list for init script file (when initScriptType=FILE), size 1 + */ + private List initScriptResourceList; + + /** + * main script source: SCRIPT = inline sql text, FILE = sql from resource center file + */ + private SqlSourceType rawScriptType = SqlSourceType.FILE; + + /** + * main sql script (when rawScriptType=SCRIPT), or read-only display content (when rawScriptType=FILE) + */ + private String rawScript; + + /** + * resource list for main script file (when rawScriptType=FILE), size 1 + */ + private List resourceList; + + /** + * default: ; + */ + private String statementSeparator = ";"; + + /** + * print query result rows count in log, default 0 means do not print rows + */ + private int maxPrintRows = 0; + + public Properties toJdbcProperties() { + Properties props = new Properties(); + if (jdbcProperties != null) { + jdbcProperties.forEach((k, v) -> { + if (k != null && v != null) { + props.put(k, v); + } + }); + } + return props; + } + + @Override + public boolean checkParameters() { + if (StringUtils.isBlank(flinkJdbcUrl)) { + return false; + } + if (initScriptType == SqlSourceType.FILE + && (CollectionUtils.isEmpty(initScriptResourceList) || initScriptResourceList.size() < 1)) { + return false; + } + if (rawScriptType == SqlSourceType.FILE) { + return CollectionUtils.isNotEmpty(resourceList) && resourceList.size() >= 1; + } + return rawScriptType == SqlSourceType.SCRIPT && StringUtils.isNotBlank(rawScript); + } + + @Override + public List getResourceFilesList() { + List list = new ArrayList<>(); + if (initScriptType == SqlSourceType.FILE && CollectionUtils.isNotEmpty(initScriptResourceList)) { + list.addAll(initScriptResourceList); + } + if (rawScriptType == SqlSourceType.FILE && CollectionUtils.isNotEmpty(resourceList)) { + list.addAll(resourceList); + } + return list; + } + + public String getFlinkJdbcUrl() { + return flinkJdbcUrl; + } + + public void setFlinkJdbcUrl(String flinkJdbcUrl) { + this.flinkJdbcUrl = flinkJdbcUrl; + } + + public Map getJdbcProperties() { + return jdbcProperties; + } + + public void setJdbcProperties(Map jdbcProperties) { + this.jdbcProperties = jdbcProperties; + } + + public SqlSourceType getInitScriptType() { + return initScriptType; + } + + public void setInitScriptType(SqlSourceType initScriptType) { + this.initScriptType = initScriptType; + } + + public String getInitScript() { + return initScript; + } + + public void setInitScript(String initScript) { + this.initScript = initScript; + } + + public List getInitScriptResourceList() { + return initScriptResourceList; + } + + public void setInitScriptResourceList(List initScriptResourceList) { + this.initScriptResourceList = initScriptResourceList; + } + + public SqlSourceType getRawScriptType() { + return rawScriptType; + } + + public void setRawScriptType(SqlSourceType rawScriptType) { + this.rawScriptType = rawScriptType; + } + + public String getRawScript() { + return rawScript; + } + + public void setRawScript(String rawScript) { + this.rawScript = rawScript; + } + + public List getResourceList() { + return resourceList; + } + + public void setResourceList(List resourceList) { + this.resourceList = resourceList; + } + + public String getStatementSeparator() { + return statementSeparator; + } + + public void setStatementSeparator(String statementSeparator) { + this.statementSeparator = statementSeparator; + } + + public int getMaxPrintRows() { + return maxPrintRows; + } + + public void setMaxPrintRows(int maxPrintRows) { + this.maxPrintRows = maxPrintRows; + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTask.java new file mode 100644 index 000000000000..a1c531852cdf --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTask.java @@ -0,0 +1,221 @@ +/* + * 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.flinksqlgateway; + +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.plugin.task.api.AbstractTask; +import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack; +import org.apache.dolphinscheduler.plugin.task.api.TaskConstants; +import org.apache.dolphinscheduler.plugin.task.api.TaskException; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.enums.SqlSourceType; +import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo; +import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters; +import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext; +import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang3.StringUtils; + +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public class FlinkSqlGatewayTask extends AbstractTask { + + private final TaskExecutionContext taskExecutionContext; + private FlinkSqlGatewayParameters parameters; + + private volatile Connection connection; + private volatile Statement statement; + + public FlinkSqlGatewayTask(TaskExecutionContext taskExecutionContext) { + super(taskExecutionContext); + this.taskExecutionContext = taskExecutionContext; + } + + @Override + public void init() { + parameters = + JSONUtils.parseObject(taskExecutionContext.getTaskParams(), FlinkSqlGatewayParameters.class); + + log.info("Initialize flink sqlgateway task params {}", JSONUtils.toPrettyJsonString(parameters)); + + if (parameters == null || !parameters.checkParameters()) { + throw new TaskException("flink sqlgateway task params is not valid"); + } + } + + @Override + public AbstractParameters getParameters() { + return parameters; + } + + @Override + public void handle(TaskCallBack taskCallBack) throws TaskException { + try { + String jdbcUrl = resolveFlinkJdbcUrl(); + Properties props = parameters.toJdbcProperties(); + log.info("Connecting flink sql gateway with jdbcUrl: {}", jdbcUrl); + connection = DriverManager.getConnection(jdbcUrl, props); + statement = connection.createStatement(); + + executeScriptIfPresent(resolveScriptContent(parameters.getInitScriptType(), + parameters.getInitScript(), parameters.getInitScriptResourceList()), "init"); + executeScriptIfPresent(resolveScriptContent(parameters.getRawScriptType(), + parameters.getRawScript(), parameters.getResourceList()), "main"); + + setExitStatusCode(TaskConstants.EXIT_CODE_SUCCESS); + } catch (Exception e) { + if (exitStatusCode == TaskConstants.EXIT_CODE_KILL) { + log.info("flink sqlgateway task has been killed"); + return; + } + setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE); + throw new TaskException("Execute flink sqlgateway task failed", e); + } finally { + closeQuietly(statement); + closeQuietly(connection); + } + } + + @Override + public void cancel() throws TaskException { + exitStatusCode = TaskConstants.EXIT_CODE_KILL; + try { + if (statement != null) { + statement.cancel(); + } + } catch (Exception e) { + throw new TaskException("Cancel flink sqlgateway task failed", e); + } finally { + closeQuietly(statement); + closeQuietly(connection); + } + } + + private String resolvePlaceholderParams(String content) { + Map paramsMap = taskExecutionContext.getPrepareParamsMap(); + Map stringParams = ParameterUtils.convert(paramsMap); + return ParameterUtils.convertParameterPlaceholders(content, stringParams); + } + + private String resolveFlinkJdbcUrl() { + return resolvePlaceholderParams(parameters.getFlinkJdbcUrl()); + } + + private String resolveScriptContent(SqlSourceType scriptType, String inlineScript, + List resourceList) throws Exception { + if (scriptType == SqlSourceType.FILE && resourceList != null && !resourceList.isEmpty()) { + String resourceName = resourceList.get(0).getResourceName(); + ResourceContext resourceContext = taskExecutionContext.getResourceContext(); + String localPath = resourceContext.getResourceItem(resourceName).getResourceAbsolutePathInLocal(); + String content = FileUtils.readFileToString(new File(localPath), StandardCharsets.UTF_8); + log.info("Load {} sql script from resource file: {}, {} chars", scriptType, localPath, content.length()); + return resolvePlaceholderParams(content); + } + if (scriptType == SqlSourceType.SCRIPT && StringUtils.isNotBlank(inlineScript)) { + return resolvePlaceholderParams(inlineScript); + } + return inlineScript; + } + + private void executeScriptIfPresent(String script, String tag) throws Exception { + if (StringUtils.isBlank(script)) { + return; + } + List sqlList = splitSql(script, parameters.getStatementSeparator()); + for (int i = 0; i < sqlList.size(); i++) { + String sql = sqlList.get(i); + String trimmed = stripLeadingCommentsAndBlanks(sql == null ? "" : sql); + if (trimmed.isEmpty()) { + continue; + } + + log.info("[{}] execute #{} sql: {}", tag, i + 1, trimmed); + + boolean hasResultSet = statement.execute(trimmed); + if (hasResultSet) { + try (ResultSet rs = statement.getResultSet()) { + int row = 0; + int maxPrint = Math.max(parameters.getMaxPrintRows(), 0); + while (rs != null && rs.next() && row < maxPrint) { + row++; + // avoid log explosion, print row index only + log.info("[{}] query result row {}", tag, row); + } + if (maxPrint > 0) { + log.info("[{}] query printed rows: {}", tag, row); + } + } + } else { + int updateCount = statement.getUpdateCount(); + log.info("[{}] updateCount={}", tag, updateCount); + } + } + } + + static List splitSql(String script, String separator) { + String sep = StringUtils.isBlank(separator) ? ";" : separator; + return Arrays.stream(script.split(Pattern.quote(sep))) + .collect(Collectors.toList()); + } + + /** + * Strip leading line comments (-- ...) and blank lines so that Flink parser + * receives only the actual statement (e.g. SET pipeline.name='...'). + */ + static String stripLeadingCommentsAndBlanks(String sql) { + if (sql == null) { + return ""; + } + String[] lines = sql.split("\n"); + for (int i = 0; i < lines.length; i++) { + String trimmed = lines[i].trim(); + if (!trimmed.isEmpty() && !trimmed.startsWith("--")) { + return String.join("\n", Arrays.copyOfRange(lines, i, lines.length)).trim(); + } + } + // all lines are comments or blank, nothing to execute + return ""; + } + + private static void closeQuietly(AutoCloseable c) { + if (c == null) { + return; + } + try { + c.close(); + } catch (Exception ignored) { + // ignored + } + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskChannel.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskChannel.java new file mode 100644 index 000000000000..bb22609fe4cc --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskChannel.java @@ -0,0 +1,36 @@ +/* + * 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.flinksqlgateway; + +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.plugin.task.api.TaskChannel; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters; + +public class FlinkSqlGatewayTaskChannel implements TaskChannel { + + @Override + public FlinkSqlGatewayTask createTask(TaskExecutionContext taskRequest) { + return new FlinkSqlGatewayTask(taskRequest); + } + + @Override + public AbstractParameters parseParameters(String taskParams) { + return JSONUtils.parseObject(taskParams, FlinkSqlGatewayParameters.class); + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskChannelFactory.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskChannelFactory.java new file mode 100644 index 000000000000..bf67b9ec93a8 --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/main/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskChannelFactory.java @@ -0,0 +1,37 @@ +/* + * 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.flinksqlgateway; + +import org.apache.dolphinscheduler.plugin.task.api.TaskChannel; +import org.apache.dolphinscheduler.plugin.task.api.TaskChannelFactory; + +import com.google.auto.service.AutoService; + +@AutoService(TaskChannelFactory.class) +public class FlinkSqlGatewayTaskChannelFactory implements TaskChannelFactory { + + @Override + public TaskChannel create() { + return new FlinkSqlGatewayTaskChannel(); + } + + @Override + public String getName() { + return "FLINK_SGW"; + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/test/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayParametersTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/test/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayParametersTest.java new file mode 100644 index 000000000000..078562b8f8be --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/test/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayParametersTest.java @@ -0,0 +1,127 @@ +/* + * 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.flinksqlgateway; + +import org.apache.dolphinscheduler.plugin.task.api.enums.SqlSourceType; +import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo; + +import java.util.Collections; +import java.util.Properties; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class FlinkSqlGatewayParametersTest { + + private FlinkSqlGatewayParameters buildValidFileBasedParameters() { + FlinkSqlGatewayParameters parameters = new FlinkSqlGatewayParameters(); + parameters.setFlinkJdbcUrl("jdbc:flink://localhost:8083"); + parameters.setRawScriptType(SqlSourceType.FILE); + ResourceInfo mainResource = new ResourceInfo(); + mainResource.setResourceName("main.sql"); + parameters.setResourceList(Collections.singletonList(mainResource)); + parameters.setInitScriptType(SqlSourceType.FILE); + ResourceInfo initResource = new ResourceInfo(); + initResource.setResourceName("init.sql"); + parameters.setInitScriptResourceList(Collections.singletonList(initResource)); + return parameters; + } + + @Test + public void checkParametersShouldPassWhenFileBasedScriptsAreValid() { + FlinkSqlGatewayParameters parameters = buildValidFileBasedParameters(); + Assertions.assertTrue(parameters.checkParameters()); + } + + @Test + public void checkParametersShouldFailWhenJdbcUrlIsBlank() { + FlinkSqlGatewayParameters parameters = buildValidFileBasedParameters(); + parameters.setFlinkJdbcUrl(" "); + Assertions.assertFalse(parameters.checkParameters()); + } + + @Test + public void checkParametersShouldFailWhenMainScriptFileIsMissing() { + FlinkSqlGatewayParameters parameters = buildValidFileBasedParameters(); + parameters.setResourceList(null); + Assertions.assertFalse(parameters.checkParameters()); + + parameters.setResourceList(Collections.emptyList()); + Assertions.assertFalse(parameters.checkParameters()); + } + + @Test + public void checkParametersShouldFailWhenInitScriptFileIsMissing() { + FlinkSqlGatewayParameters parameters = buildValidFileBasedParameters(); + parameters.setInitScriptResourceList(null); + Assertions.assertFalse(parameters.checkParameters()); + } + + @Test + public void checkParametersShouldPassWhenInlineScriptsAreValid() { + FlinkSqlGatewayParameters parameters = new FlinkSqlGatewayParameters(); + parameters.setFlinkJdbcUrl("jdbc:flink://localhost:8083"); + parameters.setRawScriptType(SqlSourceType.SCRIPT); + parameters.setRawScript("SELECT 1"); + parameters.setInitScriptType(SqlSourceType.SCRIPT); + Assertions.assertTrue(parameters.checkParameters()); + } + + @Test + public void checkParametersShouldFailWhenInlineMainScriptIsBlank() { + FlinkSqlGatewayParameters parameters = new FlinkSqlGatewayParameters(); + parameters.setFlinkJdbcUrl("jdbc:flink://localhost:8083"); + parameters.setRawScriptType(SqlSourceType.SCRIPT); + parameters.setRawScript(""); + Assertions.assertFalse(parameters.checkParameters()); + } + + @Test + public void getResourceFilesListShouldAggregateInitAndMainResourceFiles() { + FlinkSqlGatewayParameters parameters = buildValidFileBasedParameters(); + Assertions.assertEquals(2, parameters.getResourceFilesList().size()); + } + + @Test + public void getResourceFilesListShouldReturnEmptyWhenScriptsAreInline() { + FlinkSqlGatewayParameters parameters = new FlinkSqlGatewayParameters(); + parameters.setFlinkJdbcUrl("jdbc:flink://localhost:8083"); + parameters.setRawScriptType(SqlSourceType.SCRIPT); + parameters.setInitScriptType(SqlSourceType.SCRIPT); + Assertions.assertEquals(0, parameters.getResourceFilesList().size()); + } + + @Test + public void toJdbcPropertiesShouldFilterNullKeysAndValues() { + FlinkSqlGatewayParameters parameters = new FlinkSqlGatewayParameters(); + parameters.setJdbcProperties(null); + Assertions.assertEquals(0, parameters.toJdbcProperties().size()); + + parameters.setJdbcProperties(new java.util.HashMap() { + + { + put("user", "flink"); + put("badKey", null); + put(null, "badValue"); + } + }); + Properties properties = parameters.toJdbcProperties(); + Assertions.assertEquals(1, properties.size()); + Assertions.assertEquals("flink", properties.getProperty("user")); + } +} diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/test/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/test/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskTest.java new file mode 100644 index 000000000000..3db51d856b27 --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink-sqlgateway/src/test/java/org/apache/dolphinscheduler/plugin/task/flinksqlgateway/FlinkSqlGatewayTaskTest.java @@ -0,0 +1,74 @@ +/* + * 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.flinksqlgateway; + +import java.util.List; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class FlinkSqlGatewayTaskTest { + + @Test + public void splitSqlShouldUseDefaultSemicolonWhenSeparatorIsBlank() { + List sqlList = FlinkSqlGatewayTask.splitSql("SET a=1;SELECT 2", " "); + Assertions.assertEquals(2, sqlList.size()); + Assertions.assertEquals("SET a=1", sqlList.get(0)); + Assertions.assertEquals("SELECT 2", sqlList.get(1)); + } + + @Test + public void splitSqlShouldSupportCustomSeparator() { + List sqlList = FlinkSqlGatewayTask.splitSql("SET a=1@@SELECT 2", "@@"); + Assertions.assertEquals(2, sqlList.size()); + Assertions.assertEquals("SET a=1", sqlList.get(0)); + Assertions.assertEquals("SELECT 2", sqlList.get(1)); + } + + @Test + public void splitSqlShouldEscapeRegexSpecialCharactersInSeparator() { + List sqlList = FlinkSqlGatewayTask.splitSql("SET a=1|SELECT 2", "|"); + Assertions.assertEquals(2, sqlList.size()); + } + + @Test + public void stripLeadingCommentsAndBlanksShouldRemoveLeadingCommentAndBlankLines() { + String sql = FlinkSqlGatewayTask.stripLeadingCommentsAndBlanks( + "-- job comment\n\n-- another comment\nSET pipeline.name='demo'"); + Assertions.assertEquals("SET pipeline.name='demo'", sql); + } + + @Test + public void stripLeadingCommentsAndBlanksShouldKeepStatementWithoutComments() { + String sql = FlinkSqlGatewayTask.stripLeadingCommentsAndBlanks("SELECT 1"); + Assertions.assertEquals("SELECT 1", sql); + } + + @Test + public void stripLeadingCommentsAndBlanksShouldReturnEmptyWhenOnlyComments() { + Assertions.assertEquals("", FlinkSqlGatewayTask.stripLeadingCommentsAndBlanks("-- comment only\n\n")); + Assertions.assertEquals("", FlinkSqlGatewayTask.stripLeadingCommentsAndBlanks(null)); + } + + @Test + public void stripLeadingCommentsAndBlanksShouldKeepInlineCommentInsideStatement() { + String sql = FlinkSqlGatewayTask.stripLeadingCommentsAndBlanks( + "SELECT 1 -- trailing comment"); + Assertions.assertEquals("SELECT 1 -- trailing comment", sql); + } +} diff --git a/dolphinscheduler-task-plugin/pom.xml b/dolphinscheduler-task-plugin/pom.xml index 830650f50f55..5a9d4f3903b4 100644 --- a/dolphinscheduler-task-plugin/pom.xml +++ b/dolphinscheduler-task-plugin/pom.xml @@ -63,6 +63,7 @@ dolphinscheduler-task-remoteshell dolphinscheduler-task-aliyunserverlessspark dolphinscheduler-task-emr-serverless + dolphinscheduler-task-flink-sqlgateway diff --git a/dolphinscheduler-ui/src/locales/en_US/project.ts b/dolphinscheduler-ui/src/locales/en_US/project.ts index 626410631650..7a32201b233c 100644 --- a/dolphinscheduler-ui/src/locales/en_US/project.ts +++ b/dolphinscheduler-ui/src/locales/en_US/project.ts @@ -449,6 +449,7 @@ export default { master_tips: 'Please enter master url(optional)', init_script: 'Initialization script', init_script_tips: 'Please enter initialization script', + init_script_source: 'Init Script Source', resources: 'Resources', resources_tips: 'Please select resources', resources_limit_tips: 'Please select again, resource limit:', diff --git a/dolphinscheduler-ui/src/locales/zh_CN/project.ts b/dolphinscheduler-ui/src/locales/zh_CN/project.ts index 90d526b175b7..0519daf0cf91 100644 --- a/dolphinscheduler-ui/src/locales/zh_CN/project.ts +++ b/dolphinscheduler-ui/src/locales/zh_CN/project.ts @@ -436,6 +436,7 @@ export default { master_tips: '请输入master url（选填）', init_script: '初始化脚本', init_script_tips: '请输入初始化脚本', + init_script_source: 'Init Script 来源', resources: '资源', resources_tips: '请选择资源', resources_limit_tips: '请重新选择，资源个数限制:', diff --git a/dolphinscheduler-ui/src/store/project/task-type.ts b/dolphinscheduler-ui/src/store/project/task-type.ts index 597340750ae3..1bcd0c7dff24 100644 --- a/dolphinscheduler-ui/src/store/project/task-type.ts +++ b/dolphinscheduler-ui/src/store/project/task-type.ts @@ -44,6 +44,9 @@ export const TASK_TYPES_MAP = { FLINK: { alias: 'FLINK' }, + FLINK_SGW: { + alias: 'FLINK_SGW' + }, MR: { alias: 'MapReduce', helperLinkDisable: true diff --git a/dolphinscheduler-ui/src/store/project/types.ts b/dolphinscheduler-ui/src/store/project/types.ts index a58135b3e741..1e67171a2ade 100644 --- a/dolphinscheduler-ui/src/store/project/types.ts +++ b/dolphinscheduler-ui/src/store/project/types.ts @@ -58,6 +58,7 @@ type TaskType = | 'DATA_FACTORY' | 'REMOTESHELL' | 'ALIYUN_SERVERLESS_SPARK' + | 'FLINK_SGW' type ProgramType = 'JAVA' | 'SCALA' | 'PYTHON' diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts index da4cce0763e5..bbf215839414 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts @@ -90,3 +90,4 @@ export { useDataFactory } from './use-data-factory' export { useRemoteShell } from './use-remote-shell' export { useYarnQueue } from './use-queue' export { useAliyunServerlessSpark } from './use-aliyun-serverless-spark' +export { useFlinkSqlGateway } from './use-flink-sqlgateway' diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-flink-sqlgateway.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-flink-sqlgateway.ts new file mode 100644 index 000000000000..07bc4f77626f --- /dev/null +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-flink-sqlgateway.ts @@ -0,0 +1,231 @@ +import { computed, watch, ref } from 'vue' +import { useI18n } from 'vue-i18n' +import { useResources, useCustomParams } from '.' +import { viewResource } from '@/service/modules/resources' +import { useUserStore } from '@/store/user/user' +import type { IJsonItem } from '../types' + +/** Script source: inline input */ +const SCRIPT_SOURCE_SCRIPT = 'SCRIPT' +/** Script source: resource center file */ +const SCRIPT_SOURCE_FILE = 'FILE' + +export function useFlinkSqlGateway(model: { [field: string]: any }): IJsonItem[] { + const { t } = useI18n() + const userStore = useUserStore() + const resourceContentLoading = ref(false) + + if (!model.flinkJdbcUrl) model.flinkJdbcUrl = 'jdbc:flink://host:port' + if (!model.statementSeparator) model.statementSeparator = ';' + if (!model.maxPrintRows) model.maxPrintRows = 0 + if (!model.initScript) model.initScript = '' + if (!model.rawScript) model.rawScript = '' + if (!model.rawScriptType) model.rawScriptType = SCRIPT_SOURCE_FILE + if (!model.resourceList) model.resourceList = [] + if (!model.initScriptType) model.initScriptType = SCRIPT_SOURCE_FILE + if (!model.initScriptResourceList) model.initScriptResourceList = [] + + const initResourceContentLoading = ref(false) + const resourcesRequired = ref(model.rawScriptType === SCRIPT_SOURCE_FILE) + const initResourcesRequired = ref(model.initScriptType === SCRIPT_SOURCE_FILE) + const resourcesLimit = computed(() => + model.rawScriptType === SCRIPT_SOURCE_SCRIPT ? -1 : 1 + ) + const resourcesSpan = computed(() => + model.rawScriptType === SCRIPT_SOURCE_FILE ? 24 : 0 + ) + const initResourcesSpan = computed(() => + model.initScriptType === SCRIPT_SOURCE_FILE ? 24 : 0 + ) + const scriptEditorReadonly = computed(() => model.rawScriptType === SCRIPT_SOURCE_FILE) + const initScriptEditorReadonly = computed(() => model.initScriptType === SCRIPT_SOURCE_FILE) + + const loadResourceContent = (fullName: string) => { + const tenantCode = (userStore.getUserInfo as any)?.tenantCode + if (!tenantCode) { + model.rawScript = '' + return + } + resourceContentLoading.value = true + viewResource({ + fullName, + tenantCode, + skipLineNum: 0, + limit: -1 + }) + .then((res: { content: string }) => { + model.rawScript = res?.content ?? '' + }) + .catch(() => { + model.rawScript = '' + }) + .finally(() => { + resourceContentLoading.value = false + }) + } + + const loadInitResourceContent = (fullName: string) => { + const tenantCode = (userStore.getUserInfo as any)?.tenantCode + if (!tenantCode) { + model.initScript = '' + return + } + initResourceContentLoading.value = true + viewResource({ + fullName, + tenantCode, + skipLineNum: 0, + limit: -1 + }) + .then((res: { content: string }) => { + model.initScript = res?.content ?? '' + }) + .catch(() => { + model.initScript = '' + }) + .finally(() => { + initResourceContentLoading.value = false + }) + } + + watch( + () => model.rawScriptType, + () => { + resourcesRequired.value = model.rawScriptType === SCRIPT_SOURCE_FILE + if (model.rawScriptType === SCRIPT_SOURCE_SCRIPT) { + model.resourceList = [] + } else if (model.resourceList?.length === 1) { + loadResourceContent(model.resourceList[0]) + } else { + model.rawScript = '' + } + } + ) + + watch( + () => model.resourceList, + (list) => { + if (model.rawScriptType === SCRIPT_SOURCE_FILE && list?.length === 1) { + loadResourceContent(list[0]) + } else if (model.rawScriptType === SCRIPT_SOURCE_FILE) { + model.rawScript = '' + } + }, + { deep: true, immediate: true } + ) + + watch( + () => model.initScriptType, + () => { + initResourcesRequired.value = model.initScriptType === SCRIPT_SOURCE_FILE + if (model.initScriptType === SCRIPT_SOURCE_SCRIPT) { + model.initScriptResourceList = [] + } else if (model.initScriptResourceList?.length === 1) { + loadInitResourceContent(model.initScriptResourceList[0]) + } else { + model.initScript = '' + } + } + ) + + watch( + () => model.initScriptResourceList, + (list) => { + if (model.initScriptType === SCRIPT_SOURCE_FILE && list?.length === 1) { + loadInitResourceContent(list[0]) + } else if (model.initScriptType === SCRIPT_SOURCE_FILE) { + model.initScript = '' + } + }, + { deep: true, immediate: true } + ) + + const SCRIPT_SOURCE_OPTIONS = [ + { label: t('project.node.sql_execution_type_from_script'), value: SCRIPT_SOURCE_SCRIPT }, + { label: t('project.node.sql_execution_type_from_file'), value: SCRIPT_SOURCE_FILE } + ] + + return [ + { + type: 'input', + field: 'flinkJdbcUrl', + span: 24, + name: 'JDBC URL', + props: { placeholder: 'jdbc:flink://host:port' }, + value: model.flinkJdbcUrl, + validate: { trigger: ['blur', 'input'], required: true } + }, + { + type: 'input', + field: 'statementSeparator', + span: 24, + name: 'Statement Separator', + props: { placeholder: ';' }, + value: model.statementSeparator + }, + { + type: 'input-number', + field: 'maxPrintRows', + span: 24, + name: 'Max Print Rows', + props: { min: 0 }, + value: model.maxPrintRows + }, + { + type: 'select', + field: 'initScriptType', + span: 12, + name: t('project.node.init_script_source'), + options: SCRIPT_SOURCE_OPTIONS, + validate: { trigger: ['input', 'blur'], required: true } + }, + useResources( + initResourcesSpan, + initResourcesRequired, + computed(() => (model.initScriptType === SCRIPT_SOURCE_FILE ? 1 : -1)), + 'initScriptResourceList' + ), + { + type: 'editor', + field: 'initScript', + span: 24, + name: 'Init Script', + props: { + language: 'sql', + readOnly: initScriptEditorReadonly, + loading: initResourceContentLoading + }, + value: model.initScript + }, + { + type: 'select', + field: 'rawScriptType', + span: 12, + name: t('project.node.sql_execution_type'), + options: SCRIPT_SOURCE_OPTIONS, + validate: { trigger: ['input', 'blur'], required: true } + }, + useResources(resourcesSpan, resourcesRequired, resourcesLimit), + { + type: 'editor', + field: 'rawScript', + span: 24, + name: 'Script', + props: { + language: 'sql', + readOnly: scriptEditorReadonly, + loading: resourceContentLoading + }, + value: model.rawScript, + validate: { + trigger: ['blur', 'input'], + required: model.rawScriptType === SCRIPT_SOURCE_SCRIPT + } + }, + ...useCustomParams({ + model, + field: 'localParams', + isSimple: model.readonly + }) + ] +} diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-resources.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-resources.ts index b77e8f0a764f..1b2a83214419 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-resources.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-resources.ts @@ -28,7 +28,8 @@ import { useClipboard } from '@vueuse/core' export function useResources( span: number | Ref = 24, required: boolean | Ref = false, - limit: number | Ref = -1 + limit: number | Ref = -1, + field: string = 'resourceList' ): IJsonItem { const { t } = useI18n() @@ -141,7 +142,7 @@ export function useResources( return { type: 'tree-select', - field: 'resourceList', + field, class: 'resource-select', name: t('project.node.resources'), span: span, 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 47498c2ceacb..413476832316 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 @@ -90,6 +90,17 @@ export function formatParams(data: INodeData): { taskParams.taskManager = data.taskManager taskParams.parallelism = data.parallelism } + if (data.taskType === 'FLINK_SGW') { + taskParams.flinkJdbcUrl = data.flinkJdbcUrl + taskParams.statementSeparator = data.statementSeparator + taskParams.maxPrintRows = data.maxPrintRows + taskParams.rawScriptType = data.rawScriptType + taskParams.initScriptType = data.initScriptType + taskParams.initScript = + data.initScriptType === 'FILE' ? '' : (data.initScript ?? '') + taskParams.rawScript = + data.rawScriptType === 'FILE' ? '' : (data.rawScript ?? '') + } if (data.taskType === 'GRPC') { taskParams.url = data.url taskParams.grpcCredentialType = data.grpcCredentialType @@ -508,13 +519,22 @@ export function formatParams(data: INodeData): { item.value = item.value || '' return item }), - initScript: data.initScript, - rawScript: data.rawScript, + initScript: + data.initScriptType === 'FILE' ? '' : (data.initScript ?? ''), + rawScript: + data.rawScriptType === 'FILE' ? '' : (data.rawScript ?? ''), resourceList: data.resourceList?.length ? data.resourceList.map((fullName: string) => ({ resourceName: `${fullName}` })) : [], + initScriptType: data.initScriptType, + initScriptResourceList: + (data.initScriptResourceList as string[] | undefined)?.length + ? (data.initScriptResourceList as string[]).map((fullName) => ({ + resourceName: `${fullName}` + })) + : [], ...taskParams }, taskPriority: data.taskPriority, @@ -547,7 +567,12 @@ export function formatModel(data: ITaskData) { 'timeoutNotifyStrategy', 'taskParams' ]), - ...omit(data.taskParams, ['resourceList', 'mainJar', 'localParams']), + ...omit(data.taskParams, [ + 'resourceList', + 'mainJar', + 'localParams', + 'initScriptResourceList' + ]), environmentCode: data.environmentCode === -1 ? null : data.environmentCode, timeoutFlag: data.timeoutFlag === 'OPEN', timeoutNotifyStrategy: data.timeoutNotifyStrategy @@ -564,6 +589,12 @@ export function formatModel(data: ITaskData) { (item: { resourceName: string }) => `${item.resourceName}` ) } + if (data.taskParams?.initScriptResourceList) { + (params as INodeData).initScriptResourceList = + data.taskParams.initScriptResourceList.map( + (item: { resourceName: string }) => `${item.resourceName}` + ) + } if (data.taskParams?.mainJar) { params.mainJar = data.taskParams?.mainJar.resourceName } diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts index 89fa49e17b4a..984ba635f060 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts @@ -52,6 +52,7 @@ import { useLinkis } from './use-linkis' import { useDataFactory } from './use-data-factory' import { useRemoteShell } from './use-remote-shell' import { useAliyunServerlessSpark } from './use-aliyun-serverless-spark' +import { useFlinkSqlGateway } from './use-flink-sqlgateway' export default { SHELL: useShell, @@ -90,5 +91,6 @@ export default { LINKIS: useLinkis, DATA_FACTORY: useDataFactory, REMOTESHELL: useRemoteShell, - ALIYUN_SERVERLESS_SPARK: useAliyunServerlessSpark + ALIYUN_SERVERLESS_SPARK: useAliyunServerlessSpark, + FLINK_SGW: useFlinkSqlGateway } diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-flink-sqlgateway.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-flink-sqlgateway.ts new file mode 100644 index 000000000000..17c4fc892e1c --- /dev/null +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-flink-sqlgateway.ts @@ -0,0 +1,54 @@ +import { reactive } from 'vue' +import * as Fields from '../fields/index' +import type { IJsonItem, INodeData, ITaskData } from '../types' + +export function useFlinkSqlGateway(params: any) { + const { projectCode, from = 0, readonly, data } = params + + const model = reactive({ + taskType: 'FLINK_SGW', + name: '', + flag: 'YES', + description: '', + timeoutFlag: false, + localParams: [], + environmentCode: null, + failRetryInterval: 1, + failRetryTimes: 0, + workerGroup: 'default', + delayTime: 0, + timeout: 30, + timeoutNotifyStrategy: ['WARN'], + + // --- FLINK_SGW custom fields --- + flinkJdbcUrl: 'jdbc:flink://host:port', + rawScriptType: 'FILE', + initScriptType: 'FILE', + initScript: '', + rawScript: '', + resourceList: [], + initScriptResourceList: [], + statementSeparator: ';', + maxPrintRows: 0, + jdbcProperties: {} + }) + + return { + json: [ + Fields.useName(from), + ...Fields.useTaskDefinition({ projectCode, from, readonly, data, model }), + Fields.useRunFlag(), + Fields.useDescription(), + Fields.useTaskPriority(), + Fields.useWorkerGroup(projectCode), + Fields.useEnvironmentName(model, !data?.id), + ...Fields.useTaskGroup(model, projectCode), + ...Fields.useFailed(), + Fields.useDelayTime(model), + ...Fields.useTimeoutAlarm(model), + ...Fields.useFlinkSqlGateway(model), + Fields.usePreTasks() + ] as IJsonItem[], + model + } +} 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 49d88de66920..c031fabb14ff 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts @@ -480,12 +480,20 @@ interface ITaskParams { yarnQueue?: string awsRegion?: string kubeConfig?: string + flinkJdbcUrl?: string + rawScriptType?: 'SCRIPT' | 'FILE' + initScriptType?: 'SCRIPT' | 'FILE' + initScriptResourceList?: ISourceItem[] + statementSeparator?: string + maxPrintRows?: number + jdbcProperties?: Record } interface INodeData extends Omit< ITaskParams, | 'resourceList' + | 'initScriptResourceList' | 'mainJar' | 'targetParams' | 'sourceParams' @@ -522,6 +530,7 @@ interface INodeData preTaskOptions?: [] postTaskOptions?: [] resourceList?: string[] + initScriptResourceList?: string[] mainJar?: string timeoutSetting?: boolean isCustomTask?: boolean diff --git a/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts b/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts index d0eb36b277a8..afd2040284e6 100644 --- a/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts +++ b/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts @@ -52,6 +52,7 @@ export type TaskType = | 'DATA_FACTORY' | 'REMOTESHELL' | 'ALIYUN_SERVERLESS_SPARK' + | 'FLINK_SGW' export type TaskExecuteType = 'STREAM' | 'BATCH' @@ -191,6 +192,11 @@ export const TASK_TYPES_MAP = { ALIYUN_SERVERLESS_SPARK: { alias: 'ALIYUN_SERVERLESS_SPARK', helperLinkDisable: true + }, + FLINK_SGW: { + alias: 'FLINK_SGW', + helperLinkDisable: true, + taskExecuteType: 'BATCH' } } as { [key in TaskType]: { diff --git a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss index 809bc6fae3cf..cc61323ff8b9 100644 --- a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss +++ b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/dag.module.scss @@ -114,6 +114,7 @@ $bgLight: #ffffff; background-image: url('/images/task-icons/sql.png'); } &.icon-flink, + &.icon-flink_sgw, &.icon-flink_stream { background-image: url('/images/task-icons/flink.png'); } @@ -224,6 +225,7 @@ $bgLight: #ffffff; background-image: url('/images/task-icons/sql_hover.png'); } &.icon-flink, + &.icon-flink_sgw, &.icon-flink_stream { background-image: url('/images/task-icons/flink_hover.png'); } diff --git a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-cell-active.ts b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-cell-active.ts index 03a19a563d02..61736064ec69 100644 --- a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-cell-active.ts +++ b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-cell-active.ts @@ -90,8 +90,9 @@ export function useCellActive(options: Options) { let portAttrs = null if (isHover || isSelected) { - img = `${import.meta.env.BASE_URL}images/task-icons/${(node.data - .taskType !== 'FLINK_STREAM' + img = `${import.meta.env.BASE_URL}images/task-icons/${( + (node.data.taskType !== 'FLINK_STREAM' && + node.data.taskType !== 'FLINK_SGW') ? node.data.taskType : 'FLINK' ).toLocaleLowerCase()}_hover.png` @@ -103,8 +104,9 @@ export function useCellActive(options: Options) { portAttrs = _.merge(portDefault, portSelected) } } else { - img = `${import.meta.env.BASE_URL}images/task-icons/${(node.data - .taskType !== 'FLINK_STREAM' + img = `${import.meta.env.BASE_URL}images/task-icons/${( + (node.data.taskType !== 'FLINK_STREAM' && + node.data.taskType !== 'FLINK_SGW' ) ? node.data.taskType : 'FLINK' ).toLocaleLowerCase()}.png` diff --git a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-custom-cell-builder.ts b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-custom-cell-builder.ts index c9ca1a1277a7..4ceceb77f07a 100644 --- a/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-custom-cell-builder.ts +++ b/dolphinscheduler-ui/src/views/projects/workflow/components/dag/use-custom-cell-builder.ts @@ -94,7 +94,9 @@ export function useCustomCellBuilder() { // Use href instead of xlink:href, you may lose the icon when downloadPNG 'xlink:href': `${ import.meta.env.BASE_URL - }images/task-icons/${(type !== ('FLINK_STREAM' as TaskType) + }images/task-icons/${( + (type !== ('FLINK_STREAM' as TaskType) && + type !== ('FLINK_SGW' as TaskType)) ? type : 'FLINK' ).toLocaleLowerCase()}.png`