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

use connection center to manage spark conn

This commit is contained in:
EricGao888 2024-05-29 12:02:19 +08:00
parent 8824b5bab2
commit 0d2be37691
25 changed files with 766 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dolphinscheduler-bom/pom.xml
View File

@ -395,7 +395,8 @@
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>${mysql-connector.version}</version>
<scope>test</scope>
<!-- TODO: Revert this!!!-->
<!-- <scope>test</scope>-->
</dependency>
<dependency>

57
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/pom.xml Normal file
View File

@ -0,0 +1,57 @@
<?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-aliyunserverlessspark</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-datasource-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>emr_serverless_spark20230808</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>credentials-java</artifactId>
<version>0.3.0</version>
</dependency>
</dependencies>
</project>

59
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkClientWrapper.java Normal file
View File

@ -0,0 +1,59 @@
/*
* 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.aliyunserverlessspark;
import static com.google.common.base.Preconditions.checkNotNull;
import lombok.extern.slf4j.Slf4j;
import com.aliyun.emr_serverless_spark20230808.Client;
import com.aliyun.teaopenapi.models.Config;
@Slf4j
public class AliyunServerlessSparkClientWrapper implements AutoCloseable {
private Client aliyunServerlessSparkClient;
public AliyunServerlessSparkClientWrapper(
String accessKeyId,
String accessKeySecret,
String regionId)
throws Exception {
checkNotNull(accessKeyId, accessKeySecret, regionId);
String endpoint = String.format("emr-serverless-spark.%s.aliyuncs.com", regionId);
Config config = new Config()
.setEndpoint(endpoint)
.setAccessKeyId(accessKeyId)
.setAccessKeySecret(accessKeySecret);
aliyunServerlessSparkClient = new Client(config);
}
public boolean checkConnect(String accessKeyId, String accessKeySecret, String regionId) {
try {
// If the login fails, an exception will be thrown directly
return true;
} catch (Exception e) {
log.info("spark client failed to connect to the server");
return false;
}
}
@Override
public void close() throws Exception {
}
}

37
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannel.java Normal file
View File

@ -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.datasource.aliyunserverlessspark;
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 AliyunServerlessSparkDataSourceChannel implements DataSourceChannel {
@Override
public AdHocDataSourceClient createAdHocDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
throw new UnsupportedOperationException("Aliyun Serverless Spark AdHocDataSourceClient is not supported");
}
@Override
public PooledDataSourceClient createPooledDataSourceClient(BaseConnectionParam baseConnectionParam, DbType dbType) {
throw new UnsupportedOperationException("Aliyun Serverless Spark AdHocDataSourceClient is not supported");
}
}

39
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannelFactory.java Normal file
View File

@ -0,0 +1,39 @@
/*
* 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.aliyunserverlessspark;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
import org.apache.dolphinscheduler.spi.datasource.DataSourceChannelFactory;
import org.apache.dolphinscheduler.spi.enums.DbType;
import com.google.auto.service.AutoService;
@AutoService(DataSourceChannelFactory.class)
public class AliyunServerlessSparkDataSourceChannelFactory implements DataSourceChannelFactory {
@Override
public DataSourceChannel create() {
return new AliyunServerlessSparkDataSourceChannel();
}
@Override
public String getName() {
return DbType.ALIYUN_SERVERLESS_SPARK.getName();
}
}

40
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkUtils.java Normal file
View File

@ -0,0 +1,40 @@
/*
* 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.aliyunserverlessspark;
import org.apache.dolphinscheduler.plugin.datasource.aliyunserverlessspark.param.AliyunServerlessSparkConnectionParam;
import com.aliyun.emr_serverless_spark20230808.Client;
import com.aliyun.teaopenapi.models.Config;
public class AliyunServerlessSparkUtils {
private AliyunServerlessSparkUtils() {
throw new IllegalStateException("Utility class");
}
public static Client getAliyunServerlessSparkClient(AliyunServerlessSparkConnectionParam connectionParam) throws Exception {
String endpoint = String.format("emr-serverless-spark.%s.aliyuncs.com", connectionParam.getRegionId());
Config config = new Config()
.setEndpoint(endpoint)
.setAccessKeyId(connectionParam.getAccessKeyId())
.setAccessKeySecret(connectionParam.getAccessKeySecret());
return new Client(config);
}
}

35
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkConnectionParam.java Normal file
View File

@ -0,0 +1,35 @@
/*
* 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.aliyunserverlessspark.param;
import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
import lombok.Data;
import com.fasterxml.jackson.annotation.JsonInclude;
@Data
@JsonInclude(JsonInclude.Include.NON_NULL)
public class AliyunServerlessSparkConnectionParam implements ConnectionParam {
protected String accessKeyId;
protected String accessKeySecret;
protected String regionId;
}

38
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceParamDTO.java Normal file
View File

@ -0,0 +1,38 @@
/*
* 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.aliyunserverlessspark.param;
import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
import org.apache.dolphinscheduler.spi.enums.DbType;
import lombok.Data;
@Data
public class AliyunServerlessSparkDataSourceParamDTO extends BaseDataSourceParamDTO {
protected String accessKeyId;
protected String accessKeySecret;
protected String regionId;
@Override
public DbType getType() {
return DbType.ALIYUN_SERVERLESS_SPARK;
}
}

144
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceProcessor.java Normal file
View File

@ -0,0 +1,144 @@
/*
* 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.aliyunserverlessspark.param;
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.plugin.datasource.aliyunserverlessspark.AliyunServerlessSparkClientWrapper;
import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
import org.apache.dolphinscheduler.spi.enums.DbType;
import org.apache.commons.lang3.StringUtils;
import java.sql.Connection;
import java.text.MessageFormat;
import lombok.extern.slf4j.Slf4j;
import com.google.auto.service.AutoService;
@AutoService(DataSourceProcessor.class)
@Slf4j
public class AliyunServerlessSparkDataSourceProcessor extends AbstractDataSourceProcessor {
@Override
public BaseDataSourceParamDTO castDatasourceParamDTO(String paramJson) {
return JSONUtils.parseObject(paramJson, AliyunServerlessSparkDataSourceParamDTO.class);
}
@Override
public void checkDatasourceParam(BaseDataSourceParamDTO datasourceParamDTO) {
AliyunServerlessSparkDataSourceParamDTO aliyunServerlessSparkDataSourceParamDTO = (AliyunServerlessSparkDataSourceParamDTO) datasourceParamDTO;
if (StringUtils.isEmpty(aliyunServerlessSparkDataSourceParamDTO.getRegionId()) ||
StringUtils.isEmpty(aliyunServerlessSparkDataSourceParamDTO.getAccessKeyId()) ||
StringUtils.isEmpty(aliyunServerlessSparkDataSourceParamDTO.getRegionId())) {
throw new IllegalArgumentException("spark datasource param is not valid");
}
}
@Override
public String getDatasourceUniqueId(ConnectionParam connectionParam, DbType dbType) {
AliyunServerlessSparkConnectionParam baseConnectionParam = (AliyunServerlessSparkConnectionParam) connectionParam;
return MessageFormat.format(
"{0}@{1}@{2}@{3}",
dbType.getName(),
baseConnectionParam.getRegionId(),
PasswordUtils.encodePassword(baseConnectionParam.getAccessKeyId()),
PasswordUtils.encodePassword(baseConnectionParam.getAccessKeySecret()));
}
@Override
public BaseDataSourceParamDTO createDatasourceParamDTO(String connectionJson) {
AliyunServerlessSparkConnectionParam connectionParams = (AliyunServerlessSparkConnectionParam) createConnectionParams(connectionJson);
AliyunServerlessSparkDataSourceParamDTO aliyunServerlessSparkDataSourceParamDTO = new AliyunServerlessSparkDataSourceParamDTO();
aliyunServerlessSparkDataSourceParamDTO.setAccessKeyId(connectionParams.getAccessKeyId());
aliyunServerlessSparkDataSourceParamDTO.setAccessKeySecret(connectionParams.getAccessKeySecret());
aliyunServerlessSparkDataSourceParamDTO.setRegionId(connectionParams.getRegionId());
return aliyunServerlessSparkDataSourceParamDTO;
}
@Override
public AliyunServerlessSparkConnectionParam createConnectionParams(BaseDataSourceParamDTO datasourceParam) {
AliyunServerlessSparkDataSourceParamDTO aliyunServerlessSparkDataSourceParamDTO = (AliyunServerlessSparkDataSourceParamDTO) datasourceParam;
AliyunServerlessSparkConnectionParam aliyunServerlessSparkConnectionParam = new AliyunServerlessSparkConnectionParam();
aliyunServerlessSparkConnectionParam.setAccessKeyId(aliyunServerlessSparkDataSourceParamDTO.getAccessKeyId());
aliyunServerlessSparkConnectionParam.setAccessKeySecret(aliyunServerlessSparkDataSourceParamDTO.getAccessKeySecret());
aliyunServerlessSparkConnectionParam.setRegionId(aliyunServerlessSparkDataSourceParamDTO.getRegionId());
return aliyunServerlessSparkConnectionParam;
}
@Override
public ConnectionParam createConnectionParams(String connectionJson) {
return JSONUtils.parseObject(connectionJson, AliyunServerlessSparkConnectionParam.class);
}
@Override
public String getDatasourceDriver() {
return "";
}
@Override
public String getValidationQuery() {
return "";
}
@Override
public String getJdbcUrl(ConnectionParam connectionParam) {
return "";
}
@Override
public Connection getConnection(ConnectionParam connectionParam) {
return null;
}
@Override
public boolean checkDataSourceConnectivity(ConnectionParam connectionParam) {
AliyunServerlessSparkConnectionParam baseConnectionParam = (AliyunServerlessSparkConnectionParam) connectionParam;
try (
AliyunServerlessSparkClientWrapper aliyunServerlessSparkClientWrapper =
new AliyunServerlessSparkClientWrapper(
baseConnectionParam.getAccessKeyId(),
baseConnectionParam.getAccessKeySecret(),
baseConnectionParam.getRegionId())
) {
return aliyunServerlessSparkClientWrapper.checkConnect(
baseConnectionParam.getAccessKeyId(),
baseConnectionParam.getAccessKeySecret(),
baseConnectionParam.getRegionId());
} catch (Exception e) {
log.error("spark client failed to connect to the server", e);
return false;
}
}
@Override
public DbType getDbType() {
return DbType.ALIYUN_SERVERLESS_SPARK;
}
@Override
public DataSourceProcessor create() {
return new AliyunServerlessSparkDataSourceProcessor();
}
}

107
dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/test/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceProcessorTest.java Normal file
View File

@ -0,0 +1,107 @@
///*
// * 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.aliyunserverlessspark;
//
//import org.apache.dolphinscheduler.plugin.datasource.zeppelin.param.AliyunServerlessSparkConnectionParam;
//import org.apache.dolphinscheduler.plugin.datasource.zeppelin.param.AliyunServerlessSparkDataSourceParamDTO;
//import org.apache.dolphinscheduler.plugin.datasource.zeppelin.param.AliyunServerlessSparkDataSourceProcessor;
//import org.apache.dolphinscheduler.spi.enums.DbType;
//
//import org.junit.jupiter.api.Assertions;
//import org.junit.jupiter.api.BeforeEach;
//import org.junit.jupiter.api.Test;
//import org.junit.jupiter.api.extension.ExtendWith;
//import org.mockito.MockedConstruction;
//import org.mockito.Mockito;
//import org.mockito.junit.jupiter.MockitoExtension;
//
//@ExtendWith(MockitoExtension.class)
//public class AliyunServerlessSparkDataSourceProcessorTest {
//
// private AliyunServerlessSparkDataSourceProcessor zeppelinDataSourceProcessor;
//
// private String connectJson =
// "{\"username\":\"lucky\",\"password\":\"123456\",\"restEndpoint\":\"https://dolphinscheduler.com:8080\"}";
//
// @BeforeEach
// public void init() {
// zeppelinDataSourceProcessor = new AliyunServerlessSparkDataSourceProcessor();
// }
//
// @Test
// void testCheckDatasourceParam() {
// AliyunServerlessSparkDataSourceParamDTO zeppelinDataSourceParamDTO = new AliyunServerlessSparkDataSourceParamDTO();
// Assertions.assertThrows(IllegalArgumentException.class,
// () -> zeppelinDataSourceProcessor.checkDatasourceParam(zeppelinDataSourceParamDTO));
// zeppelinDataSourceParamDTO.setRestEndpoint("http://dolphinscheduler.com:8080");
// Assertions.assertThrows(IllegalArgumentException.class,
// () -> zeppelinDataSourceProcessor.checkDatasourceParam(zeppelinDataSourceParamDTO));
// zeppelinDataSourceParamDTO.setUserName("root");
// Assertions
// .assertDoesNotThrow(() -> zeppelinDataSourceProcessor.checkDatasourceParam(zeppelinDataSourceParamDTO));
// }
//
// @Test
// void testGetDatasourceUniqueId() {
// AliyunServerlessSparkConnectionParam zeppelinConnectionParam = new AliyunServerlessSparkConnectionParam();
// zeppelinConnectionParam.setRestEndpoint("https://dolphinscheduler.com:8080");
// zeppelinConnectionParam.setUsername("root");
// zeppelinConnectionParam.setPassword("123456");
// Assertions.assertEquals("zeppelin@https://dolphinscheduler.com:8080@root@123456",
// zeppelinDataSourceProcessor.getDatasourceUniqueId(zeppelinConnectionParam, DbType.ZEPPELIN));
//
// }
//
// @Test
// void testCreateDatasourceParamDTO() {
// AliyunServerlessSparkDataSourceParamDTO zeppelinDataSourceParamDTO =
// (AliyunServerlessSparkDataSourceParamDTO) zeppelinDataSourceProcessor.createDatasourceParamDTO(connectJson);
// Assertions.assertEquals("lucky", zeppelinDataSourceParamDTO.getUserName());
// Assertions.assertEquals("123456", zeppelinDataSourceParamDTO.getPassword());
// Assertions.assertEquals("https://dolphinscheduler.com:8080", zeppelinDataSourceParamDTO.getRestEndpoint());
// }
//
// @Test
// void testCreateConnectionParams() {
// AliyunServerlessSparkDataSourceParamDTO zeppelinDataSourceParamDTO =
// (AliyunServerlessSparkDataSourceParamDTO) zeppelinDataSourceProcessor.createDatasourceParamDTO(connectJson);
// AliyunServerlessSparkConnectionParam zeppelinConnectionParam =
// zeppelinDataSourceProcessor.createConnectionParams(zeppelinDataSourceParamDTO);
// Assertions.assertEquals("lucky", zeppelinConnectionParam.getUsername());
// Assertions.assertEquals("123456", zeppelinConnectionParam.getPassword());
// Assertions.assertEquals("https://dolphinscheduler.com:8080", zeppelinConnectionParam.getRestEndpoint());
// }
//
// @Test
// void testTestConnection() {
// AliyunServerlessSparkDataSourceParamDTO zeppelinDataSourceParamDTO =
// (AliyunServerlessSparkDataSourceParamDTO) zeppelinDataSourceProcessor.createDatasourceParamDTO(connectJson);
// AliyunServerlessSparkConnectionParam connectionParam =
// zeppelinDataSourceProcessor.createConnectionParams(zeppelinDataSourceParamDTO);
// Assertions.assertFalse(zeppelinDataSourceProcessor.checkDataSourceConnectivity(connectionParam));
// try (
// MockedConstruction<ZeppelinClientWrapper> sshClientWrapperMockedConstruction =
// Mockito.mockConstruction(ZeppelinClientWrapper.class, (mock, context) -> {
// Mockito.when(
// mock.checkConnect(connectionParam.getUsername(), connectionParam.getPassword()))
// .thenReturn(true);
// })) {
// Assertions.assertTrue(zeppelinDataSourceProcessor.checkDataSourceConnectivity(connectionParam));
// }
// }
//}

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-aliyunserverlessspark</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>

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-aliyunserverlessspark</module>
</modules>
<dependencyManagement>

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

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

5
dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/pom.xml
View File

@ -54,11 +54,6 @@
<artifactId>dolphinscheduler-datasource-all</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.zeppelin</groupId>
<artifactId>zeppelin-client</artifactId>
<version>0.10.1</version>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>emr_serverless_spark20230808</artifactId>

33
dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkParameters.java
View File

@ -3,7 +3,9 @@ package org.apache.dolphinscheduler.plugin.task.aliyunserverlessspark;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import java.util.List;
@ -11,22 +13,31 @@ import java.util.List;
@Slf4j
public class AliyunServerlessSparkParameters extends AbstractParameters {
// connection configurations
String regionId;
String accessKeyId;
String accessKeySecret;
// private String regionId;
// private String accessKeyId;
// private String accessKeySecret;
// spark job configurations
String workspaceId;
String resourceQueueId;
String codeType;
String jobName;
String engineReleaseVersion;
String entryPoint;
String entryPointArguments;
String sparkSubmitParameters;
private String workspaceId;
private String resourceQueueId;
private String codeType;
private String jobName;
private String engineReleaseVersion;
private String entryPoint;
private String entryPointArguments;
private String sparkSubmitParameters;
boolean isProduction;
private int datasource;
private String type;
@Override
public boolean checkParameters() {
return true;
}
@Override
public ResourceParametersHelper getResources() {
ResourceParametersHelper resources = super.getResources();
resources.put(ResourceType.DATASOURCE, datasource);
return resources;
}
}

36
dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java
View File

@ -18,12 +18,18 @@
package org.apache.dolphinscheduler.plugin.task.aliyunserverlessspark;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.datasource.aliyunserverlessspark.param.AliyunServerlessSparkConnectionParam;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils;
import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask;
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.ResourceType;
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import org.apache.dolphinscheduler.spi.enums.DbType;
import com.aliyun.emr_serverless_spark20230808.Client;
@ -56,12 +62,20 @@ public class AliyunServerlessSparkTask extends AbstractRemoteTask {
private AliyunServerlessSparkParameters aliyunServerlessSparkParameters;
private AliyunServerlessSparkConnectionParam aliyunServerlessSparkConnectionParam;
private String jobRunId;
private RunState previousState;
private RunState currentState;
private String accessKeyId;
private String accessKeySecret;
private String regionId;
protected AliyunServerlessSparkTask(TaskExecutionContext taskExecutionContext) {
super(taskExecutionContext);
this.taskExecutionContext = taskExecutionContext;
@ -75,9 +89,19 @@ public class AliyunServerlessSparkTask extends AbstractRemoteTask {
throw new AliyunServerlessSparkTaskException("Aliyun-Serverless-Spark task parameters are not valid!");
}
String accessKeyId = aliyunServerlessSparkParameters.getAccessKeyId();
String accessKeySecret = aliyunServerlessSparkParameters.getAccessKeySecret();
String regionId = aliyunServerlessSparkParameters.getRegionId();
ResourceParametersHelper resourceParametersHelper = taskExecutionContext.getResourceParametersHelper();
DataSourceParameters dataSourceParameters = (DataSourceParameters) resourceParametersHelper.getResourceParameters(ResourceType.DATASOURCE, aliyunServerlessSparkParameters.getDatasource());
log.info("[debug111] dataSourceParameters - {}", dataSourceParameters);
log.info("[debug111] aliyunServerlessSparkParameters - {}", aliyunServerlessSparkParameters);
aliyunServerlessSparkConnectionParam = (AliyunServerlessSparkConnectionParam) DataSourceUtils
.buildConnectionParams(
DbType.valueOf(aliyunServerlessSparkParameters.getType()),
dataSourceParameters.getConnectionParams());
accessKeyId = aliyunServerlessSparkConnectionParam.getAccessKeyId();
accessKeySecret = aliyunServerlessSparkConnectionParam.getAccessKeySecret();
regionId = aliyunServerlessSparkConnectionParam.getRegionId();
try {
aliyunServerlessSparkClient = buildAliyunServerlessSparkClient(accessKeyId, accessKeySecret, regionId);
} catch (Exception e) {
@ -169,7 +193,7 @@ public class AliyunServerlessSparkTask extends AbstractRemoteTask {
private StartJobRunRequest buildStartJobRunRequest(AliyunServerlessSparkParameters aliyunServerlessSparkParameters) {
StartJobRunRequest startJobRunRequest = new StartJobRunRequest();
startJobRunRequest.setRegionId(aliyunServerlessSparkParameters.getRegionId());
startJobRunRequest.setRegionId(regionId);
startJobRunRequest.setResourceQueueId(aliyunServerlessSparkParameters.getResourceQueueId());
startJobRunRequest.setCodeType(aliyunServerlessSparkParameters.getCodeType());
startJobRunRequest.setName(aliyunServerlessSparkParameters.getJobName());
@ -198,13 +222,13 @@ public class AliyunServerlessSparkTask extends AbstractRemoteTask {
private GetJobRunRequest buildGetJobRunRequest(AliyunServerlessSparkParameters aliyunServerlessSparkParameters) {
GetJobRunRequest getJobRunRequest = new GetJobRunRequest();
getJobRunRequest.setRegionId(aliyunServerlessSparkParameters.getRegionId());
getJobRunRequest.setRegionId(regionId);
return getJobRunRequest;
}
private CancelJobRunRequest buildCancelJobRunRequest(AliyunServerlessSparkParameters aliyunServerlessSparkParameters) {
CancelJobRunRequest cancelJobRunRequest = new CancelJobRunRequest();
cancelJobRunRequest.setRegionId(aliyunServerlessSparkParameters.getRegionId());
cancelJobRunRequest.setRegionId(regionId);
return cancelJobRunRequest;
}
}

8
dolphinscheduler-ui/src/locales/en_US/datasource.ts
View File

@ -97,5 +97,11 @@ export default {
kubeConfig: 'kubeConfig',
kubeConfig_tips: 'Please input KubeConfig',
namespace: 'namespace',
namespace_tips: 'Please input namespace'
namespace_tips: 'Please input namespace',
access_key_id: 'Access Key Id',
access_key_id_tips: 'Please enter access key id',
access_Key_secret: 'Access Key Secret',
access_Key_secret_tips: 'Please enter access key secret',
region_id: 'Region Id',
region_id_tips: 'Please enter region id'
}

6
dolphinscheduler-ui/src/locales/zh_CN/datasource.ts
View File

@ -94,5 +94,9 @@ export default {
kubeConfig: 'kubeConfig',
kubeConfig_tips: '请输入KubeConfig',
namespace: 'namespace',
namespace_tips: '请输入namespace'
namespace_tips: '请输入namespace',
access_key_id: 'Access Key Id',
access_key_id_tips: '请输入access key id',
access_Key_secret: 'Access Key Secret',
access_Key_secret_tips: '请输入access key secret'
}

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

@ -42,6 +42,7 @@ type IDataBase =
| 'ZEPPELIN'
| 'SAGEMAKER'
| 'K8S'
| 'ALIYUN_SERVERLESS_SPARK'
type IDataBaseLabel =
| 'MYSQL'
@ -65,6 +66,7 @@ type IDataBaseLabel =
| 'ZEPPELIN'
| 'SAGEMAKER'
| 'K8S'
| 'ALIYUN_SERVERLESS_SPARK'
interface IDataSource {
id?: number
@ -94,6 +96,9 @@ interface IDataSource {
compatibleMode?: string
publicKey?: string
datawarehouse?: string
accessKeyId?: string
accessKeySecret?: string
regionId?: string
}
interface ListReq {

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

@ -155,6 +155,9 @@ const DetailModal = defineComponent({
showHost,
showPort,
showRestEndpoint,
showAccessKeyId,
showAccessKeySecret,
showRegionId,
showAwsRegion,
showCompatibleMode,
showConnectType,
@ -270,6 +273,51 @@ const DetailModal = defineComponent({
placeholder={t('datasource.zeppelin_rest_endpoint_tips')}
/>
</NFormItem>
<NFormItem
v-show={showAccessKeyId}
label={t('datasource.access_key_id')}
path='accessKeyId'
show-require-mark
>
<NInput
allowInput={this.trim}
class='input-access_key_id'
v-model={[detailForm.accessKeyId, 'value']}
type='text'
maxlength={255}
placeholder={t('datasource.access_key_id_tips')}
/>
</NFormItem>
<NFormItem
v-show={showAccessKeySecret}
label={t('datasource.access_key_secret')}
path='accessKeySecret'
show-require-mark
>
<NInput
allowInput={this.trim}
class='input-access_key_secret'
v-model={[detailForm.accessKeySecret, 'value']}
type='text'
maxlength={255}
placeholder={t('datasource.access_key_secret_tips')}
/>
</NFormItem>
<NFormItem
v-show={showRegionId}
label={t('datasource.region_id')}
path='regionId'
show-require-mark
>
<NInput
allowInput={this.trim}
class='input-region_id'
v-model={[detailForm.regionId, 'value']}
type='text'
maxlength={255}
placeholder={t('datasource.region_id_tips')}
/>
</NFormItem>
<NFormItem
v-show={showPort}
label={t('datasource.port')}
@ -546,7 +594,7 @@ const DetailModal = defineComponent({
<NFormItem
v-show={
(!showMode || detailForm.mode === 'password') &&
detailForm.type != 'K8S'
detailForm.type != 'K8S' && detailForm.type != 'ALIYUN_SERVERLESS_SPARK'
}
label={t('datasource.user_name')}
path='userName'
@ -564,7 +612,7 @@ const DetailModal = defineComponent({
<NFormItem
v-show={
(!showMode || detailForm.mode === 'password') &&
detailForm.type != 'K8S'
detailForm.type != 'K8S' && detailForm.type != 'ALIYUN_SERVERLESS_SPARK'
}
label={t('datasource.user_password')}
path='password'

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

@ -71,6 +71,9 @@ export function useForm(id?: number) {
showPublicKey: false,
showNamespace: false,
showKubeConfig: false,
showAccessKeyId: false,
showAccessKeySecret: false,
showRegionId: false,
rules: {
name: {
trigger: ['input'],
@ -121,7 +124,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 !== 'ALIYUN_SERVERLESS_SPARK'
) {
return new Error(t('datasource.user_name_tips'))
}
@ -268,7 +272,8 @@ export function useForm(id?: number) {
type === 'SSH' ||
type === 'ZEPPELIN' ||
type === 'SAGEMAKER' ||
type === 'K8S'
type === 'K8S' ||
type === 'ALIYUN_SERVERLESS_SPARK'
) {
state.showDataBaseName = false
state.requiredDataBase = false
@ -282,7 +287,7 @@ export function useForm(id?: number) {
state.showPort = false
state.showRestEndpoint = true
}
if (type === 'SAGEMAKER' || type === 'K8S') {
if (type === 'SAGEMAKER' || type === 'K8S' || type == 'ALIYUN_SERVERLESS_SPARK') {
state.showHost = false
state.showPort = false
}
@ -290,6 +295,11 @@ export function useForm(id?: number) {
state.showNamespace = true
state.showKubeConfig = true
}
if (type === 'ALIYUN_SERVERLESS_SPARK') {
state.showAccessKeyId = true
state.showAccessKeySecret = true
state.showRegionId = true
}
} else {
state.showDataBaseName = true
state.requiredDataBase = true
@ -458,6 +468,11 @@ export const datasourceType: IDataBaseOptionKeys = {
value: 'K8S',
label: 'K8S',
defaultPort: 6443
},
ALIYUN_SERVERLESS_SPARK: {
value: 'ALIYUN_SERVERLESS_SPARK',
label: 'ALIYUN_SERVERLESS_SPARK',
defaultPort: 0
}
}

102
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-aliyun-serverless-spark.ts
View File

@ -23,59 +23,59 @@ export function useAliyunServerlessSpark(model: { [field: string]: any }): IJson
return [
// mandatory field
{
type: 'input',
field: 'regionId',
name: t('project.node.region_id'),
props: {
placeholder: t('project.node.region_id_tips')
},
validate: {
trigger: ['input', 'blur'],
required: true,
validator(validate: any, value: string) {
if (!value) {
return new Error(t('project.node.region_id_tips'))
}
}
}
},
// {
// type: 'input',
// field: 'regionId',
// name: t('project.node.region_id'),
// props: {
// placeholder: t('project.node.region_id_tips')
// },
// validate: {
// trigger: ['input', 'blur'],
// required: true,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.region_id_tips'))
// }
// }
// }
// },
{
type: 'input',
field: 'accessKeyId',
name: t('project.node.access_key_id'),
props: {
placeholder: t('project.node.access_key_id_tips')
},
validate: {
trigger: ['input', 'blur'],
required: true,
validator(validate: any, value: string) {
if (!value) {
return new Error(t('project.node.access_key_id_tips'))
}
}
}
},
// {
// type: 'input',
// field: 'accessKeyId',
// name: t('project.node.access_key_id'),
// props: {
// placeholder: t('project.node.access_key_id_tips')
// },
// validate: {
// trigger: ['input', 'blur'],
// required: true,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.access_key_id_tips'))
// }
// }
// }
// },
{
type: 'input',
field: 'accessKeySecret',
name: t('project.node.access_key_secret'),
props: {
placeholder: t('project.node.access_key_secret_tips')
},
validate: {
trigger: ['input', 'blur'],
required: true,
validator(validate: any, value: string) {
if (!value) {
return new Error(t('project.node.access_key_secret_tips'))
}
}
}
},
// {
// type: 'input',
// field: 'accessKeySecret',
// name: t('project.node.access_key_secret'),
// props: {
// placeholder: t('project.node.access_key_secret_tips')
// },
// validate: {
// trigger: ['input', 'blur'],
// required: true,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.access_key_secret_tips'))
// }
// }
// }
// },
{
type: 'input',

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: 27,
code: 'ALIYUN_SERVERLESS_SPARK',
disabled: false
}
]

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

@ -345,9 +345,9 @@ export function formatParams(data: INodeData): {
}
if (data.taskType === 'ALIYUN_SERVERLESS_SPARK') {
taskParams.regionId = data.regionId
taskParams.accessKeyId = data.accessKeyId
taskParams.accessKeySecret = data.accessKeySecret
// taskParams.regionId = data.regionId
// taskParams.accessKeyId = data.accessKeyId
// taskParams.accessKeySecret = data.accessKeySecret
taskParams.workspaceId = data.workspaceId
taskParams.resourceQueueId = data.resourceQueueId
taskParams.codeType = data.codeType
@ -357,6 +357,7 @@ export function formatParams(data: INodeData): {
taskParams.entryPointArguments = data.entryPointArguments
taskParams.sparkSubmitParameters = data.sparkSubmitParameters
taskParams.isProduction = data.isProduction
taskParams.type = data.type
}
if (data.taskType === 'K8S') {

2
dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-aliyun-serverless-spark.ts
View File

@ -65,7 +65,7 @@ export function useAliyunServerlessSpark({
...Fields.useFailed(),
Fields.useDelayTime(model),
...Fields.useTimeoutAlarm(model),
// ...Fields.useDatasource(model),
...Fields.useDatasource(model),
...Fields.useAliyunServerlessSpark(model),
Fields.usePreTasks()
] as IJsonItem[],