Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Feature][Connector-V2][Maxcompute] Add Maxcompute source & sink connector #3640

Merged
merged 10 commits into from
Dec 11, 2022
Merged
79 changes: 79 additions & 0 deletions docs/en/connector-v2/sink/Maxcompute.md
Original file line number Diff line number Diff line change
@@ -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))
82 changes: 82 additions & 0 deletions docs/en/connector-v2/source/Maxcompute.md
Original file line number Diff line number Diff line change
@@ -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))
2 changes: 2 additions & 0 deletions plugin-mapping.properties
Original file line number Diff line number Diff line change
Expand Up @@ -157,4 +157,6 @@ seatunnel.source.Gitlab = connector-http-gitlab
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
61 changes: 61 additions & 0 deletions seatunnel-connectors-v2/connector-maxcompute/pom.xml
Original file line number Diff line number Diff line change
@@ -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>
Original file line number Diff line number Diff line change
@@ -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");
}
Original file line number Diff line number Diff line change
@@ -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);
}
}
Original file line number Diff line number Diff line change
@@ -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);
}
}
Loading