[Feature][Connector-V2] [Amazondynamodb Connector]add amazondynamodb …

…source & sink connnector (#3166)
apache · Nov 4, 2022 · 183bac0 · 183bac0
1 parent edb1410
commit 183bac0
Show file tree

Hide file tree

Showing 23 changed files with 1,622 additions and 7 deletions.
diff --git a/docs/en/connector-v2/sink/Amazondynamodb.md b/docs/en/connector-v2/sink/Amazondynamodb.md
@@ -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
+
diff --git a/docs/en/connector-v2/source/Amazondynamodb.md b/docs/en/connector-v2/source/Amazondynamodb.md
@@ -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
diff --git a/plugin-mapping.properties b/plugin-mapping.properties
@@ -136,4 +136,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
diff --git a/seatunnel-connectors-v2/connector-amazondynamodb/pom.xml b/seatunnel-connectors-v2/connector-amazondynamodb/pom.xml
@@ -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>
diff --git a/...org/apache/seatunnel/connectors/seatunnel/amazondynamodb/config/AmazondynamodbConfig.java b/...org/apache/seatunnel/connectors/seatunnel/amazondynamodb/config/AmazondynamodbConfig.java
@@ -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";
+}
diff --git a/...che/seatunnel/connectors/seatunnel/amazondynamodb/config/AmazondynamodbSourceOptions.java b/...che/seatunnel/connectors/seatunnel/amazondynamodb/config/AmazondynamodbSourceOptions.java
@@ -0,0 +1,68 @@
+/*
+ * 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) {
+        this.url = config.getString(AmazondynamodbConfig.URL);
+        this.region = config.getString(AmazondynamodbConfig.REGION);
+        this.accessKeyId = config.getString(AmazondynamodbConfig.ACCESS_KEY_ID);
+        this.secretAccessKey = config.getString(AmazondynamodbConfig.SECRET_ACCESS_KEY);
+        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);
+        }
+    }
+}