[Feature][Connector-V2][Maxcompute] Add Maxcompute source & sink conn…

…ector (#3640)

docs/en/connector-v2/sink/Maxcompute.md

@@ -0,0 +1,79 @@
+# Maxcompute
+
+> Maxcompute sink connector
+
+## Description
+
+Used to read data from Maxcompute.
+
+## Key features
+
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+- [ ] [schema projection](../../concept/connector-v2-features.md)
+
+## Options
+
+| name              | type     | required  | default value |
+|-------------------|----------|-----------|---------------|
+| accessId          | string   | yes       | -             |
+| accesskey         | string   | yes       | -             |
+| endpoint          | string   | yes       | -             |
+| project           | string   | yes       | -             |
+| table_name        | string   | yes       | -             |
+| partition_spec    | string   | no        | -             |
+| overwrite         | boolean  | no        | false         |
+| common-options    | string   | no        |               |
+
+### accessId [string]
+
+`accessId` Your Maxcompute accessId which cloud be access from Alibaba Cloud.
+
+### accesskey [string]
+
+`accesskey` Your Maxcompute accessKey which cloud be access from Alibaba Cloud.
+
+### endpoint [string]
+
+`endpoint` Your Maxcompute endpoint start with http.
+
+### project [string]
+
+`project` Your Maxcompute project which is created in Alibaba Cloud.
+
+### table_name [string]
+
+`table_name` Target Maxcompute table name eg: fake.
+
+### partition_spec [string]
+
+`partition_spec` This spec of Maxcompute partition table eg:ds='20220101'.
+
+### overwrite [boolean]
+
+`overwrite` Whether to overwrite the table or partition, default: false.
+
+### common options 
+
+Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details.
+
+## Examples
+
+```hocon
+sink {
+  Maxcompute {
+    accessId="<your access id>"
+    accesskey="<your access Key>"
+    endpoint="<http://service.odps.aliyun.com/api>"
+    project="<your project>"
+    table_name="<your table name>"
+    #partition_spec="<your partition spec>"
+    #overwrite = false
+  }
+}
+```
+
+## Changelog
+
+### next version
+
+- [Feature] Add Maxcompute Sink Connector([3640](https://github.com/apache/incubator-seatunnel/pull/3640))

docs/en/connector-v2/source/Maxcompute.md

@@ -0,0 +1,82 @@
+# Maxcompute
+
+> Maxcompute source connector
+
+## Description
+
+Used to read data from Maxcompute.
+
+## Key features
+
+- [x] [batch](../../concept/connector-v2-features.md)
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+- [ ] [schema projection](../../concept/connector-v2-features.md)
+- [x] [parallelism](../../concept/connector-v2-features.md)
+- [ ] [support user-defined split](../../concept/connector-v2-features.md)
+
+## Options
+
+| name            | type   | required  | default value |
+|-----------------|--------|-----------|---------------|
+| accessId        | string | yes       | -             |
+| accesskey       | string | yes       | -             |
+| endpoint        | string | yes       | -             |
+| project         | string | yes       | -             |
+| table_name      | string | yes       | -             |
+| partition_spec  | string | no        | -             |
+| split_row       | int    | no        | 10000         |
+| common-options  | string | no        |               |
+
+### accessId [string]
+
+`accessId` Your Maxcompute accessId which cloud be access from Alibaba Cloud.
+
+### accesskey [string]
+
+`accesskey` Your Maxcompute accessKey which cloud be access from Alibaba Cloud.
+
+### endpoint [string]
+
+`endpoint` Your Maxcompute endpoint start with http.
+
+### project [string]
+
+`project` Your Maxcompute project which is created in Alibaba Cloud.
+
+### table_name [string]
+
+`table_name` Target Maxcompute table name eg: fake.
+
+### partition_spec [string]
+
+`partition_spec` This spec of Maxcompute partition table eg:ds='20220101'.
+
+### split_row [int]
+
+`split_row` Number of rows per split, default: 10000.
+
+### common options 
+
+Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details.
+
+## Examples
+
+```hocon
+source {
+  Maxcompute {
+    accessId="<your access id>"
+    accesskey="<your access Key>"
+    endpoint="<http://service.odps.aliyun.com/api>"
+    project="<your project>"
+    table_name="<your table name>"
+    #partition_spec="<your partition spec>"
+    #split_row = 10000
+  }
+}
+```
+
+## Changelog
+
+### next version
+
+- [Feature] Add Maxcompute Source Connector([3640](https://github.com/apache/incubator-seatunnel/pull/3640))

plugin-mapping.properties

@@ -160,4 +160,6 @@ seatunnel.source.Notion = connector-http-notion
 seatunnel.sink.RabbitMQ = connector-rabbitmq
 seatunnel.source.RabbitMQ = connector-rabbitmq
 seatunnel.source.OpenMldb = connector-openmldb
+seatunnel.source.Maxcompute = connector-maxcompute
+seatunnel.sink.Maxcompute = connector-maxcompute
 seatunnel.source.MySQL-CDC = connector-cdc-mysql

seatunnel-connectors-v2/connector-maxcompute/pom.xml

@@ -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">
+    <parent>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>connector-maxcompute</artifactId>
+
+    <properties>
+        <maxcompute.version>0.31.3</maxcompute.version>
+        <commons.lang3.version>3.4</commons.lang3.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun.odps</groupId>
+            <artifactId>odps-sdk-core</artifactId>
+            <version>${maxcompute.version}-public</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${commons.lang3.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>

seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/config/MaxcomputeConfig.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.seatunnel.connectors.seatunnel.maxcompute.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+import java.io.Serializable;
+
+public class MaxcomputeConfig implements Serializable {
+    private static final int SPLIT_ROW_DEFAULT = 10000;
+    public static final Option<String> ACCESS_ID = Options.key("accessId")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("Your Maxcompute accessId which cloud be access from Alibaba Cloud");
+    public static final Option<String> ACCESS_KEY = Options.key("accesskey")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("Your Maxcompute accessKey which cloud be access from Alibaba Cloud");
+    public static final Option<String> ENDPOINT = Options.key("endpoint")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("Your Maxcompute endpoint start with http");
+    public static final Option<String> PROJECT = Options.key("project")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("Your Maxcompute project which is created in Alibaba Cloud");
+    public static final Option<String> TABLE_NAME = Options.key("table_name")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("Target Maxcompute table name eg: fake");
+    public static final Option<String> PARTITION_SPEC = Options.key("partition_spec")
+        .stringType()
+        .noDefaultValue()
+        .withDescription("This spec of Maxcompute partition table.");
+    public static final Option<Integer> SPLIT_ROW = Options.key("split_row")
+        .intType()
+        .defaultValue(SPLIT_ROW_DEFAULT)
+        .withDescription("Number of rows per split. default: 10000");
+    public static final Option<Boolean> OVERWRITE = Options.key("overwrite")
+        .booleanType()
+        .defaultValue(false)
+        .withDescription("Whether to overwrite the table or partition");
+}

seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/exception/MaxcomputeConnectorException.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    https://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.seatunnel.connectors.seatunnel.maxcompute.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class MaxcomputeConnectorException extends SeaTunnelRuntimeException {
+
+    public MaxcomputeConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage) {
+        super(seaTunnelErrorCode, errorMessage);
+    }
+
+    public MaxcomputeConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage, Throwable cause) {
+        super(seaTunnelErrorCode, errorMessage, cause);
+    }
+
+    public MaxcomputeConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
+        super(seaTunnelErrorCode, cause);
+    }
+}

seatunnel-connectors-v2/connector-maxcompute/src/main/java/org/apache/seatunnel/connectors/seatunnel/maxcompute/sink/MaxcomputeSink.java

@@ -0,0 +1,67 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.maxcompute.sink;
+
+import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
+import org.apache.seatunnel.connectors.seatunnel.maxcompute.util.MaxcomputeUtil;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.google.auto.service.AutoService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@AutoService(SeaTunnelSink.class)
+public class MaxcomputeSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
+    private static final Logger LOG = LoggerFactory.getLogger(MaxcomputeSink.class);
+    private Config pluginConfig;
+    private SeaTunnelRowType typeInfo;
+
+    @Override
+    public String getPluginName() {
+        return "Maxcompute";
+    }
+
+    @Override
+    public void prepare(Config pluginConfig) throws PrepareFailException {
+        this.pluginConfig = pluginConfig;
+        MaxcomputeUtil.initTableOrPartition(pluginConfig);
+    }
+
+    @Override
+    public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
+        this.typeInfo = seaTunnelRowType;
+    }
+
+    @Override
+    public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
+        return this.typeInfo;
+    }
+
+    @Override
+    public AbstractSinkWriter<SeaTunnelRow, Void> createWriter(SinkWriter.Context context) {
+        return new MaxcomputeWriter(this.typeInfo, this.pluginConfig);
+    }
+}