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] [Amazondynamodb Connector]add amazondynamodb source & sink connnector #3166

Merged
merged 28 commits into from
Nov 4, 2022
Merged
Show file tree
Hide file tree
Changes from 23 commits
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
a692d35
add amazondynamodb connnector
liugddx Oct 24, 2022
08908af
add amazondynamodb source
liugddx Oct 27, 2022
d3fa1b9
add amazondynamodbSource
liugddx Oct 28, 2022
7ed060b
add Amazondynamodb Source
liugddx Oct 28, 2022
5b0ab48
fix Amazondynamodb Source bug
liugddx Oct 29, 2022
bc881a5
Merge remote-tracking branch 'upstream/dev' into feature-amazondynamodb
liugddx Oct 29, 2022
2a6d350
add Amazondynamodb sink connector
liugddx Oct 29, 2022
4714319
fix some bug
liugddx Oct 30, 2022
9f2d18b
add doc
liugddx Oct 30, 2022
24071a3
add e2e test
liugddx Oct 30, 2022
05c05d9
fix e2e test error
liugddx Oct 30, 2022
b586ce9
fix e2e test error
liugddx Oct 30, 2022
c0a9751
fix e2e test error
liugddx Oct 30, 2022
91ab78b
fix e2e test error
liugddx Oct 30, 2022
38692ef
fix e2e test error
liugddx Oct 30, 2022
efebf11
fix e2e test error
liugddx Oct 30, 2022
cd1ec33
Merge branch 'dev' into feature-amazondynamodb
liugddx Oct 31, 2022
73bb6df
add deserializer and serializer
liugddx Oct 31, 2022
39414cb
remove unnecessary code
liugddx Oct 31, 2022
964d580
Merge remote-tracking branch 'upstream/dev' into feature-amazondynamodb
liugddx Oct 31, 2022
4e1e17c
add batch write.
liugddx Oct 31, 2022
d1e4f1f
modify sink doc
liugddx Nov 1, 2022
aaf3d1d
fix cv bug.
liugddx Nov 2, 2022
5a7ff67
fix some error
liugddx Nov 2, 2022
b26981a
fix bug.
liugddx Nov 3, 2022
1e6e186
fix bug.
liugddx Nov 3, 2022
97d8026
Merge remote-tracking branch 'upstream/dev' into feature-amazondynamodb
liugddx Nov 3, 2022
fca481e
fix ci error
liugddx Nov 3, 2022
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 69 additions & 0 deletions docs/en/connector-v2/sink/Amazondynamodb.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@

# Amazondynamodb

> Amazondynamodb sink connector

## Description

Write data to `Amazondynamodb`

## Key features

- [ ] [exactly-once](../../concept/connector-v2-features.md)
- [ ] [schema projection](../../concept/connector-v2-features.md)

## Options

| name | type | required | default value |
|----------------- | ------ |----------| ------------- |
| url | string | yes | - |
| region | string | yes | - |
| access_key_id | string | yes | - |
| secret_access_key| string | yes | - |
| table | string | yes | - |
| batch_size | string | no | 25 |
| batch_interval_ms| string | no | 1000 |
| common-options | | no | - |

### url [string]

url to write to Amazondynamodb.

### region [string]

The region of Amazondynamodb.

### accessKeyId [string]

The access id of Amazondynamodb.

### secretAccessKey [string]

The access secret of Amazondynamodb.

### table [string]

The table of Amazondynamodb.

### common options

Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details.

## Example

```bash
Amazondynamodb {
url = "http://127.0.0.1:8000"
region = "us-east-1"
accessKeyId = "dummy-key"
secretAccessKey = "dummy-secret"
table = "TableName"
}
```

## Changelog

### next version

- Add Amazondynamodb Sink Connector

108 changes: 108 additions & 0 deletions docs/en/connector-v2/source/Amazondynamodb.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
# Amazondynamodb

> Amazondynamodb source connector
## Description

Read data from Amazondynamodb.

## Key features

- [x] [batch](../../concept/connector-v2-features.md)
- [ ] [stream](../../concept/connector-v2-features.md)
- [ ] [exactly-once](../../concept/connector-v2-features.md)
- [x] [schema projection](../../concept/connector-v2-features.md)
- [ ] [parallelism](../../concept/connector-v2-features.md)
- [ ] [support user-defined split](../../concept/connector-v2-features.md)

## Options

| name | type | required | default value |
| ---------------- | ------ | -------- | ------------- |
| url | string | yes | - |
| region | string | yes | - |
| access_key_id | string | yes | - |
| secret_access_key| string | yes | - |
| table | string | yes | - |
| schema | object | yes | - |
| common-options | | yes | - |

### url [string]

url to read to Amazondynamodb.

### region [string]

The region of Amazondynamodb.

### accessKeyId [string]

The access id of Amazondynamodb.

### secretAccessKey [string]

The access secret of Amazondynamodb.

### table [string]

The table of Amazondynamodb.

### schema [object]

#### fields [Config]

Amazon Dynamodb is a NOSQL database service of support keys-value storage and document data structure,there is no way to get the data type.Therefore, we must configure schma.

such as:

```
schema {
fields {
id = int
key_aa = string
key_bb = string
}
}
```

### common options

Source Plugin common parameters, refer to [Source Plugin](common-options.md) for details

## Example

```bash
Amazondynamodb {
url = "http://127.0.0.1:8000"
region = "us-east-1"
accessKeyId = "dummy-key"
secretAccessKey = "dummy-secret"
table = "TableName"
schema = {
fields {
artist = string
c_map = "map<string, array<int>>"
c_array = "array<int>"
c_string = string
c_boolean = boolean
c_tinyint = tinyint
c_smallint = smallint
c_int = int
c_bigint = bigint
c_float = float
c_double = double
c_decimal = "decimal(30, 8)"
c_null = "null"
c_bytes = bytes
c_date = date
c_timestamp = timestamp
}
}
}
```

## Changelog

### next version

- Add Amazondynamodb Source Connector
4 changes: 3 additions & 1 deletion plugin-mapping.properties
Original file line number Diff line number Diff line change
Expand Up @@ -135,4 +135,6 @@ seatunnel.sink.MongoDB = connector-mongodb
seatunnel.source.Iceberg = connector-iceberg
seatunnel.source.InfluxDB = connector-influxdb
seatunnel.source.S3File = connector-file-s3
seatunnel.sink.S3File = connector-file-s3
seatunnel.sink.S3File = connector-file-s3
seatunnel.source.Amazondynamodb = connector-amazondynamodb
seatunnel.sink.Amazondynamodb = connector-amazondynamodb
64 changes: 64 additions & 0 deletions seatunnel-connectors-v2/connector-amazondynamodb/pom.xml
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
<?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-amazondynamodb</artifactId>

<properties>
<amazon.awssdk>2.18.1</amazon.awssdk>
</properties>

<dependencyManagement>
<dependencies>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>bom</artifactId>
<version>${amazon.awssdk}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>

<dependencies>
<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>dynamodb-enhanced</artifactId>
</dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>dynamodb</artifactId>
</dependency>
</dependencies>

</project>
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
/*
* 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.amazondynamodb.config;

import java.io.Serializable;

public class AmazondynamodbConfig implements Serializable {
public static final String URL = "url";
public static final String REGION = "region";
public static final String ACCESS_KEY_ID = "access_key_id";
public static final String SECRET_ACCESS_KEY = "secret_access_key";
public static final String TABLE = "table";
public static final String BATCH_SIZE = "batch_size";
public static final String DEFAULT_BATCH_INTERVAL_MS = "batch_interval_ms";
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
/*
* 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.amazondynamodb.config;

import org.apache.seatunnel.connectors.seatunnel.common.config.CommonConfig;

import org.apache.seatunnel.shade.com.typesafe.config.Config;

import lombok.AllArgsConstructor;
import lombok.Data;

import java.io.Serializable;

@Data
@AllArgsConstructor
public class AmazondynamodbSourceOptions implements Serializable {

private static final int DEFAULT_BATCH_SIZE = 25;
private static final int DEFAULT_BATCH_INTERVAL_MS = 1000;

private String url;

private String region;

private String accessKeyId;

private String secretAccessKey;

private String table;

private Config schema;

public int batchSize = DEFAULT_BATCH_SIZE;
public int batchIntervalMs = DEFAULT_BATCH_INTERVAL_MS;

public AmazondynamodbSourceOptions(Config config) {
if (config.hasPath(AmazondynamodbConfig.URL)) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seem like these config don't have default value, maybe you should use CheckConfigUtil.checkAllExists to make sure user didn't forget config it.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You already check it exist, No need to check again

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You already check it exist, No need to check again

Done.

this.url = config.getString(AmazondynamodbConfig.URL);
}
if (config.hasPath(AmazondynamodbConfig.REGION)) {
this.region = config.getString(AmazondynamodbConfig.REGION);
}
if (config.hasPath(AmazondynamodbConfig.ACCESS_KEY_ID)) {
this.accessKeyId = config.getString(AmazondynamodbConfig.ACCESS_KEY_ID);
}
if (config.hasPath(AmazondynamodbConfig.SECRET_ACCESS_KEY)) {
this.secretAccessKey = config.getString(AmazondynamodbConfig.SECRET_ACCESS_KEY);
}
if (config.hasPath(AmazondynamodbConfig.TABLE)) {
this.table = config.getString(AmazondynamodbConfig.TABLE);
}
if (config.hasPath(CommonConfig.SCHEMA)) {
this.schema = config.getConfig(CommonConfig.SCHEMA);
}
if (config.hasPath(AmazondynamodbConfig.BATCH_SIZE)) {
this.batchSize = config.getInt(AmazondynamodbConfig.BATCH_SIZE);
}
if (config.hasPath(AmazondynamodbConfig.DEFAULT_BATCH_INTERVAL_MS)) {
this.batchIntervalMs = config.getInt(AmazondynamodbConfig.DEFAULT_BATCH_INTERVAL_MS);
}
}
}
Loading