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

Merge 64cf2b6ca7 into b29965bdce

This commit is contained in:
John Huang 2024-04-25 22:22:19 -07:00 committed by GitHub
parent b29965bdce 64cf2b6ca7
commit c67c90bb75
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 423 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docs/configs/docsdev.js
View File

@ -366,6 +366,10 @@ export default {
{
title: 'doris',
link: '/en-us/docs/dev/user_doc/guide/datasource/doris.html',
},
{
title: 'flink',
link: '/en-us/docs/dev/user_doc/guide/datasource/flink.html',
}
],
},
@ -1086,6 +1090,10 @@ export default {
{
title: 'Doris',
link: '/zh-cn/docs/dev/user_doc/guide/datasource/doris.html',
},
{
title: 'Flink',
link: '/zh-cn/docs/dev/user_doc/guide/datasource/flink.html',
}
],
},

0
docs/docs/en/guide/datasource/flink.md Normal file
View File

0
docs/docs/zh/guide/datasource/flink.md Normal file
View File

7
dolphinscheduler-bom/pom.xml
View File

@ -124,6 +124,7 @@
<zeppelin-client.version>0.10.1</zeppelin-client.version>
<aliyun-voice.version>2.1.4</aliyun-voice.version>
<nashorn-sandbox.version>0.3.2</nashorn-sandbox.version>
<flink-jdbc.version>1.18.1</flink-jdbc.version>
</properties>
<dependencyManagement>
@ -578,6 +579,12 @@
<version>${commons-io.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-jdbc-driver-bundle</artifactId>
<version>${flink-jdbc.version}</version>
</dependency>
<dependency>
<groupId>com.github.oshi</groupId>
<artifactId>oshi-core</artifactId>

3
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/DataSourceConstants.java
View File

@ -39,6 +39,7 @@ public class DataSourceConstants {
public static final String COM_TRINO_JDBC_DRIVER = "io.trino.jdbc.TrinoDriver";
public static final String COM_DAMENG_JDBC_DRIVER = "dm.jdbc.driver.DmDriver";
public static final String ORG_APACHE_KYUUBI_JDBC_DRIVER = "org.apache.kyuubi.jdbc.KyuubiHiveDriver";
public static final String ORG_APACHE_FLINK_JDBC_DRIVER = "org.apache.flink.table.jdbc.FlinkDriver";
public static final String COM_OCEANBASE_JDBC_DRIVER = "com.oceanbase.jdbc.Driver";
public static final String NET_SNOWFLAKE_JDBC_DRIVER = "net.snowflake.client.jdbc.SnowflakeDriver";
public static final String COM_VERTICA_JDBC_DRIVER = "com.vertica.jdbc.Driver";
@ -63,6 +64,7 @@ public class DataSourceConstants {
public static final String SNOWFLAKE_VALIDATION_QUERY = "select 1";
public static final String KYUUBI_VALIDATION_QUERY = "select 1";
public static final String FLINK_VALIDATION_QUERY = "select 1";
public static final String VERTICA_VALIDATION_QUERY = "select 1";
public static final String HANA_VALIDATION_QUERY = "select 1 from DUMMY";
@ -91,6 +93,7 @@ public class DataSourceConstants {
public static final String JDBC_SNOWFLAKE = "jdbc:snowflake://";
public static final String JDBC_VERTICA = "jdbc:vertica://";
public static final String JDBC_HANA = "jdbc:sap://";
public static final String JDBC_FLINK = "jdbc:flink://";
/**
* database type

5
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml
View File

@ -158,5 +158,10 @@
<artifactId>dolphinscheduler-datasource-hana</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-flink</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>

61
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/pom.xml Normal file
View File

@ -0,0 +1,61 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-plugin</artifactId>
<version>dev-SNAPSHOT</version>
</parent>
<artifactId>dolphinscheduler-datasource-flink</artifactId>
<packaging>jar</packaging>
<name>${project.artifactId}</name>
<dependencies>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-spi</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-task-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-hive</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-jdbc-driver-bundle</artifactId>
</dependency>
</dependencies>
</project>

12
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/src/main/java/org/apache/dolphinscheduler/plugin/datasource/flink/FlinkAdHocDataSourceClient.java Normal file
View File

@ -0,0 +1,12 @@
package org.apache.dolphinscheduler.plugin.datasource.flink;
import org.apache.dolphinscheduler.plugin.datasource.api.client.BaseAdHocDataSourceClient;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
import org.apache.dolphinscheduler.spi.enums.DbType;
public class FlinkAdHocDataSourceClient extends BaseAdHocDataSourceClient {
public FlinkAdHocDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
super(baseConnectionParam, dbType);
}
}

20
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/src/main/java/org/apache/dolphinscheduler/plugin/datasource/flink/FlinkDataSourceChannel.java Normal file
View File

@ -0,0 +1,20 @@
package org.apache.dolphinscheduler.plugin.datasource.flink;
import org.apache.dolphinscheduler.spi.datasource.AdHocDataSourceClient;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
import org.apache.dolphinscheduler.spi.datasource.PooledDataSourceClient;
import org.apache.dolphinscheduler.spi.enums.DbType;
public class FlinkDataSourceChannel implements DataSourceChannel {
@Override
public AdHocDataSourceClient createAdHocDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
return new FlinkAdHocDataSourceClient(baseConnectionParam, dbType);
}
@Override
public PooledDataSourceClient createPooledDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
return new FlinkPooledDataSourceClient(baseConnectionParam, dbType);
}
}

19
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/src/main/java/org/apache/dolphinscheduler/plugin/datasource/flink/FlinkDataSourceChannelFactory.java Normal file
View File

@ -0,0 +1,19 @@
package org.apache.dolphinscheduler.plugin.datasource.flink;
import com.google.auto.service.AutoService;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannelFactory;
@AutoService(DataSourceChannelFactory.class)
public class FlinkDataSourceChannelFactory implements DataSourceChannelFactory {
@Override
public String getName() {
return "flink";
}
@Override
public DataSourceChannel create() {
return new FlinkDataSourceChannel();
}
}

28
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/src/main/java/org/apache/dolphinscheduler/plugin/datasource/flink/FlinkPooledDataSourceClient.java Normal file
View File

@ -0,0 +1,28 @@
package org.apache.dolphinscheduler.plugin.datasource.flink;
import lombok.extern.slf4j.Slf4j;
import org.apache.dolphinscheduler.plugin.datasource.api.client.BasePooledDataSourceClient;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
import org.apache.dolphinscheduler.spi.enums.DbType;
import java.sql.Connection;
import java.sql.SQLException;
@Slf4j
public class FlinkPooledDataSourceClient extends BasePooledDataSourceClient {
public FlinkPooledDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
super(baseConnectionParam, dbType);
}
@Override
public Connection getConnection() throws SQLException {
return dataSource.getConnection();
}
@Override
public void close() {
super.close();
log.info("Closed Flink datasource client.");
}
}

21
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/src/main/java/org/apache/dolphinscheduler/plugin/datasource/flink/param/FlinkConnectionParam.java Normal file
View File

@ -0,0 +1,21 @@
package org.apache.dolphinscheduler.plugin.datasource.flink.param;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
public class FlinkConnectionParam extends BaseConnectionParam {
@Override
public String toString() {
return "FlinkConnectionParam{"
+ "user='" + user + '\''
+ ", password='" + password + '\''
+ ", address='" + address + '\''
+ ", database='" + database + '\''
+ ", jdbcUrl='" + jdbcUrl + '\''
+ ", driverLocation='" + driverLocation + '\''
+ ", driverClassName='" + driverClassName + '\''
+ ", validationQuery='" + validationQuery + '\''
+ ", other='" + other + '\''
+ '}';
}
}

24
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/src/main/java/org/apache/dolphinscheduler/plugin/datasource/flink/param/FlinkDataSourceParamDTO.java Normal file
View File

@ -0,0 +1,24 @@
package org.apache.dolphinscheduler.plugin.datasource.flink.param;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
import org.apache.dolphinscheduler.spi.enums.DbType;
public class FlinkDataSourceParamDTO extends BaseDataSourceParamDTO {
@Override
public String toString() {
return "FlinkDataSourceParamDTO{"
+ "host='" + host + '\''
+ ", port=" + port
+ ", database='" + database + '\''
+ ", userName='" + userName + '\''
+ ", password='" + password + '\''
+ ", other='" + other + '\''
+ '}';
}
@Override
public DbType getType() {
return DbType.FLINK;
}
}

152
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-flink/src/main/java/org/apache/dolphinscheduler/plugin/datasource/flink/param/FlinkDataSourceProcessor.java Normal file
View File

@ -0,0 +1,152 @@
/*
* 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.datasource.flink.param;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.constants.DataSourceConstants;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.AbstractDataSourceProcessor;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.PasswordUtils;
import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
import org.apache.dolphinscheduler.spi.enums.DbType;
import org.apache.commons.collections4.MapUtils;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import com.google.auto.service.AutoService;
@AutoService(DataSourceProcessor.class)
public class FlinkDataSourceProcessor extends AbstractDataSourceProcessor {
@Override
public void checkDatasourceParam(BaseDataSourceParamDTO baseDataSourceParamDTO) {
}
@Override
public BaseDataSourceParamDTO castDatasourceParamDTO(String paramJson) {
return JSONUtils.parseObject(paramJson, FlinkDataSourceParamDTO.class);
}
@Override
public BaseDataSourceParamDTO createDatasourceParamDTO(String connectionJson) {
FlinkDataSourceParamDTO flinkDataSourceParamDTO = new FlinkDataSourceParamDTO();
FlinkConnectionParam kyuubiConnectionParam = (FlinkConnectionParam) createConnectionParams(connectionJson);
flinkDataSourceParamDTO.setDatabase(kyuubiConnectionParam.getDatabase());
flinkDataSourceParamDTO.setUserName(kyuubiConnectionParam.getUser());
flinkDataSourceParamDTO.setOther(kyuubiConnectionParam.getOther());
String[] tmpArray = kyuubiConnectionParam.getAddress().split(Constants.DOUBLE_SLASH);
StringBuilder hosts = new StringBuilder();
String[] hostPortArray = tmpArray[tmpArray.length - 1].split(Constants.COMMA);
for (String hostPort : hostPortArray) {
hosts.append(hostPort.split(Constants.COLON)[0]).append(Constants.COMMA);
}
hosts.deleteCharAt(hosts.length() - 1);
flinkDataSourceParamDTO.setHost(hosts.toString());
flinkDataSourceParamDTO.setPort(Integer.parseInt(hostPortArray[0].split(Constants.COLON)[1]));
return flinkDataSourceParamDTO;
}
@Override
public BaseConnectionParam createConnectionParams(BaseDataSourceParamDTO datasourceParam) {
FlinkDataSourceParamDTO flinkParam = (FlinkDataSourceParamDTO) datasourceParam;
StringBuilder address = new StringBuilder();
address.append(DataSourceConstants.JDBC_FLINK);
for (String zkHost : flinkParam.getHost().split(",")) {
address.append(String.format("%s:%s,", zkHost, flinkParam.getPort()));
}
address.deleteCharAt(address.length() - 1);
String jdbcUrl = address + "/" + flinkParam.getDatabase();
FlinkConnectionParam kyuubiConnectionParam = new FlinkConnectionParam();
kyuubiConnectionParam.setDatabase(flinkParam.getDatabase());
kyuubiConnectionParam.setAddress(address.toString());
kyuubiConnectionParam.setJdbcUrl(jdbcUrl);
kyuubiConnectionParam.setUser(flinkParam.getUserName());
kyuubiConnectionParam.setPassword(PasswordUtils.encodePassword(flinkParam.getPassword()));
kyuubiConnectionParam.setDriverClassName(getDatasourceDriver());
kyuubiConnectionParam.setValidationQuery(getValidationQuery());
kyuubiConnectionParam.setOther(flinkParam.getOther());
return kyuubiConnectionParam;
}
@Override
public ConnectionParam createConnectionParams(String connectionJson) {
return JSONUtils.parseObject(connectionJson, FlinkConnectionParam.class);
}
@Override
public String getDatasourceDriver() {
return DataSourceConstants.ORG_APACHE_FLINK_JDBC_DRIVER;
}
@Override
public String getValidationQuery() {
return DataSourceConstants.FLINK_VALIDATION_QUERY;
}
@Override
public String getJdbcUrl(ConnectionParam connectionParam) {
FlinkConnectionParam flinkConnectionParam = (FlinkConnectionParam) connectionParam;
String jdbcUrl = flinkConnectionParam.getJdbcUrl();
if (MapUtils.isNotEmpty(flinkConnectionParam.getOther())) {
return jdbcUrl + "?" + transformOther(flinkConnectionParam.getOther());
}
return jdbcUrl;
}
@Override
public Connection getConnection(ConnectionParam connectionParam) throws ClassNotFoundException, SQLException {
FlinkConnectionParam flinkConnectionParam = (FlinkConnectionParam) connectionParam;
Class.forName(getDatasourceDriver());
// todo:
return DriverManager.getConnection(getJdbcUrl(connectionParam),
flinkConnectionParam.getUser(), PasswordUtils.decodePassword(flinkConnectionParam.getPassword()));
}
@Override
public DbType getDbType() {
return DbType.FLINK;
}
@Override
public DataSourceProcessor create() {
return new FlinkDataSourceProcessor();
}
private String transformOther(Map<String, String> otherMap) {
if (MapUtils.isEmpty(otherMap)) {
return null;
}
List<String> otherList = new ArrayList<>();
otherMap.forEach((key, value) -> otherList.add(String.format("%s=%s", key, value)));
return String.join(";", otherList);
}
}

1
dolphinscheduler-datasource-plugin/pom.xml
View File

@ -56,6 +56,7 @@
<module>dolphinscheduler-datasource-sagemaker</module>
<module>dolphinscheduler-datasource-k8s</module>
<module>dolphinscheduler-datasource-hana</module>
<module>dolphinscheduler-datasource-flink</module>
</modules>
<dependencyManagement>

3
dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/enums/DbType.java
View File

@ -55,7 +55,8 @@ public enum DbType {
ZEPPELIN(24, "zeppelin"),
SAGEMAKER(25, "sagemaker"),
K8S(26, "k8s");
K8S(26, "k8s"),
FLINK(27, "flink");
private static final Map<Integer, DbType> DB_TYPE_MAP =
Arrays.stream(DbType.values()).collect(toMap(DbType::getCode, Functions.identity()));
@EnumValue

38
dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
View File

@ -188,6 +188,22 @@ public class SqlTask extends AbstractTask {
Connection connection =
DataSourceClientProvider.getAdHocConnection(DbType.valueOf(sqlParameters.getType()),
baseConnectionParam)) {
// main execute
String result = null;
if (sqlParameters.getType().equals("FLINK")) {
// pre execute
executeFlinkSQLQuery(connection, preStatementsBinds, "pre");
// query statements need to be convert to JsonArray and inserted into Alert to send
result = executeFlinkSQLQuery(connection, mainStatementsBinds.get(0), "main");
sqlParameters.dealOutParam(result);
// post execute
executeFlinkSQLQuery(connection, postStatementsBinds, "post");
return;
}
// create temp function
if (CollectionUtils.isNotEmpty(createFuncs)) {
@ -197,8 +213,6 @@ public class SqlTask extends AbstractTask {
// pre execute
executeUpdate(connection, preStatementsBinds, "pre");
// main execute
String result = null;
// decide whether to executeQuery or executeUpdate based on sqlType
if (sqlParameters.getSqlType() == SqlType.QUERY.ordinal()) {
// query statements need to be convert to JsonArray and inserted into Alert to send
@ -208,6 +222,7 @@ public class SqlTask extends AbstractTask {
String updateResult = executeUpdate(connection, mainStatementsBinds, "main");
result = setNonQuerySqlReturn(updateResult, sqlParameters.getLocalParams());
}
// deal out params
sqlParameters.dealOutParam(result);
@ -319,6 +334,25 @@ public class SqlTask extends AbstractTask {
}
}
// see https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/jdbcdriver/#java
private String executeFlinkSQLQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception {
try (Statement statement = connection.createStatement()) {
log.info("{} statement execute flink sql query, for sql: {}", handlerType, sqlBinds.getSql());
ResultSet resultSet = statement.executeQuery(sqlBinds.getSql());
return resultProcess(resultSet);
}
}
private String executeFlinkSQLQuery(Connection connection, List<SqlBinds> statementsBinds, String handlerType) throws Exception {
String result = "";
for (SqlBinds sqlBind : statementsBinds) {
result = executeFlinkSQLQuery(connection, sqlBind, handlerType);
log.info("{} statement execute update result: {}, for sql: {}", handlerType, result,
sqlBind.getSql());
}
return String.valueOf(result);
}
private String executeUpdate(Connection connection, List<SqlBinds> statementsBinds,
String handlerType) throws Exception {
int result = 0;

2
dolphinscheduler-ui/src/service/modules/data-source/types.ts
View File

@ -42,6 +42,7 @@ type IDataBase =
| 'ZEPPELIN'
| 'SAGEMAKER'
| 'K8S'
| 'FLINK'
type IDataBaseLabel =
| 'MYSQL'
@ -65,6 +66,7 @@ type IDataBaseLabel =
| 'ZEPPELIN'
| 'SAGEMAKER'
| 'K8S'
| 'FLINK'
interface IDataSource {
id?: number

11
dolphinscheduler-ui/src/views/datasource/list/detail.tsx
View File

@ -546,7 +546,8 @@ const DetailModal = defineComponent({
<NFormItem
v-show={
(!showMode || detailForm.mode === 'password') &&
detailForm.type != 'K8S'
detailForm.type != 'K8S' &&
detailForm.type != 'FLINK'
}
label={t('datasource.user_name')}
path='userName'
@ -564,7 +565,8 @@ const DetailModal = defineComponent({
<NFormItem
v-show={
(!showMode || detailForm.mode === 'password') &&
detailForm.type != 'K8S'
detailForm.type != 'K8S' &&
detailForm.type != 'FLINK'
}
label={t('datasource.user_password')}
path='password'
@ -592,7 +594,10 @@ const DetailModal = defineComponent({
/>
</NFormItem>
<NFormItem
v-show={showDataBaseName}
v-show={
showDataBaseName &&
detailForm.type != 'FLINK'
}
label={t('datasource.database_name')}
path='database'
show-require-mark={requiredDataBase}

11
dolphinscheduler-ui/src/views/datasource/list/use-form.ts
View File

@ -121,7 +121,8 @@ export function useForm(id?: number) {
if (
!state.detailForm.userName &&
state.detailForm.type !== 'AZURESQL' &&
state.detailForm.type !== 'K8S'
state.detailForm.type !== 'K8S' &&
state.detailForm.type !== 'FLINK'
) {
return new Error(t('datasource.user_name_tips'))
}
@ -268,7 +269,8 @@ export function useForm(id?: number) {
type === 'SSH' ||
type === 'ZEPPELIN' ||
type === 'SAGEMAKER' ||
type === 'K8S'
type === 'K8S' ||
type === 'FLINK'
) {
state.showDataBaseName = false
state.requiredDataBase = false
@ -458,6 +460,11 @@ export const datasourceType: IDataBaseOptionKeys = {
value: 'K8S',
label: 'K8S',
defaultPort: 6443
},
FLINK: {
value: 'FLINK',
label: 'FLINK',
defaultPort: 8083
}
}

5
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-datasource.ts
View File

@ -157,6 +157,11 @@ export function useDatasource(
id: 25,
code: 'SAGEMAKER',
disabled: false
},
{
id: 26,
code: 'FLINK',
disabled: false
}
]