diff --git a/dolphinscheduler-bom/pom.xml b/dolphinscheduler-bom/pom.xml index 10c4f0e4a1..4149c4cf5b 100644 --- a/dolphinscheduler-bom/pom.xml +++ b/dolphinscheduler-bom/pom.xml @@ -395,7 +395,8 @@ mysql mysql-connector-java ${mysql-connector.version} - test + + diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/pom.xml b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/pom.xml new file mode 100644 index 0000000000..80226fe367 --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/pom.xml @@ -0,0 +1,57 @@ + + + + 4.0.0 + + org.apache.dolphinscheduler + dolphinscheduler-datasource-plugin + dev-SNAPSHOT + + + dolphinscheduler-datasource-aliyunserverlessspark + jar + ${project.artifactId} + + + + org.apache.dolphinscheduler + dolphinscheduler-spi + provided + + + + org.apache.dolphinscheduler + dolphinscheduler-datasource-api + ${project.version} + + + + com.aliyun + emr_serverless_spark20230808 + 1.0.0 + + + + com.aliyun + credentials-java + 0.3.0 + + + + diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkClientWrapper.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkClientWrapper.java new file mode 100644 index 0000000000..078597d33e --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkClientWrapper.java @@ -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 { + + } +} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannel.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannel.java new file mode 100644 index 0000000000..3821a1fdbe --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannel.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.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"); + } +} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannelFactory.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannelFactory.java new file mode 100644 index 0000000000..851110aeab --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceChannelFactory.java @@ -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(); + } + +} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkUtils.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkUtils.java new file mode 100644 index 0000000000..45799d172e --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkUtils.java @@ -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); + } + +} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkConnectionParam.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkConnectionParam.java new file mode 100644 index 0000000000..b5e70721c0 --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkConnectionParam.java @@ -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; +} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceParamDTO.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceParamDTO.java new file mode 100644 index 0000000000..5af436ca18 --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceParamDTO.java @@ -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; + } +} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceProcessor.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceProcessor.java new file mode 100644 index 0000000000..c2579c6eae --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/param/AliyunServerlessSparkDataSourceProcessor.java @@ -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(); + } +} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/test/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceProcessorTest.java b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/test/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceProcessorTest.java new file mode 100644 index 0000000000..79c26e8b2f --- /dev/null +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-aliyunserverlessspark/src/test/java/org/apache/dolphinscheduler/plugin/datasource/aliyunserverlessspark/AliyunServerlessSparkDataSourceProcessorTest.java @@ -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 sshClientWrapperMockedConstruction = +// Mockito.mockConstruction(ZeppelinClientWrapper.class, (mock, context) -> { +// Mockito.when( +// mock.checkConnect(connectionParam.getUsername(), connectionParam.getPassword())) +// .thenReturn(true); +// })) { +// Assertions.assertTrue(zeppelinDataSourceProcessor.checkDataSourceConnectivity(connectionParam)); +// } +// } +//} diff --git a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml index effe3c9abb..b62a3ce414 100644 --- a/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml +++ b/dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-all/pom.xml @@ -158,5 +158,10 @@ dolphinscheduler-datasource-hana ${project.version} + + org.apache.dolphinscheduler + dolphinscheduler-datasource-aliyunserverlessspark + ${project.version} + diff --git a/dolphinscheduler-datasource-plugin/pom.xml b/dolphinscheduler-datasource-plugin/pom.xml index c30a6b4258..1f712364d9 100644 --- a/dolphinscheduler-datasource-plugin/pom.xml +++ b/dolphinscheduler-datasource-plugin/pom.xml @@ -56,6 +56,7 @@ dolphinscheduler-datasource-sagemaker dolphinscheduler-datasource-k8s dolphinscheduler-datasource-hana + dolphinscheduler-datasource-aliyunserverlessspark diff --git a/dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/enums/DbType.java b/dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/enums/DbType.java index 882b170e11..360e788cb3 100644 --- a/dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/enums/DbType.java +++ b/dolphinscheduler-spi/src/main/java/org/apache/dolphinscheduler/spi/enums/DbType.java @@ -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 DB_TYPE_MAP = Arrays.stream(DbType.values()).collect(toMap(DbType::getCode, Functions.identity())); @EnumValue diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/pom.xml b/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/pom.xml index e2d14e6a9f..7a61f116c8 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/pom.xml +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/pom.xml @@ -54,11 +54,6 @@ dolphinscheduler-datasource-all ${project.version} - - org.apache.zeppelin - zeppelin-client - 0.10.1 - com.aliyun emr_serverless_spark20230808 diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkParameters.java index 971d9e01f7..7b9415b38f 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkParameters.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkParameters.java @@ -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; + } } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java index 897ea29bc3..069511c0d2 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-aliyunserverlessspark/src/main/java/org/apache/dolphinscheduler/plugin/task/aliyunserverlessspark/AliyunServerlessSparkTask.java @@ -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; } } diff --git a/dolphinscheduler-ui/src/locales/en_US/datasource.ts b/dolphinscheduler-ui/src/locales/en_US/datasource.ts index e9b799b16e..8a877b9864 100644 --- a/dolphinscheduler-ui/src/locales/en_US/datasource.ts +++ b/dolphinscheduler-ui/src/locales/en_US/datasource.ts @@ -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' } diff --git a/dolphinscheduler-ui/src/locales/zh_CN/datasource.ts b/dolphinscheduler-ui/src/locales/zh_CN/datasource.ts index 7aa797a591..d79544d174 100644 --- a/dolphinscheduler-ui/src/locales/zh_CN/datasource.ts +++ b/dolphinscheduler-ui/src/locales/zh_CN/datasource.ts @@ -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' } diff --git a/dolphinscheduler-ui/src/service/modules/data-source/types.ts b/dolphinscheduler-ui/src/service/modules/data-source/types.ts index 444f5293dd..200f1f8bad 100644 --- a/dolphinscheduler-ui/src/service/modules/data-source/types.ts +++ b/dolphinscheduler-ui/src/service/modules/data-source/types.ts @@ -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 { diff --git a/dolphinscheduler-ui/src/views/datasource/list/detail.tsx b/dolphinscheduler-ui/src/views/datasource/list/detail.tsx index 4842651290..53b9a3c94f 100644 --- a/dolphinscheduler-ui/src/views/datasource/list/detail.tsx +++ b/dolphinscheduler-ui/src/views/datasource/list/detail.tsx @@ -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')} /> + + + + + + + + +