[Feature][Connector-V2] Support multi-table sink feature for paimon #…

apache · Mar 19, 2024 · b0abbd2 · b0abbd2
1 parent cb3db75
commit b0abbd2
Show file tree

Hide file tree

Showing 27 changed files with 1,652 additions and 128 deletions.
diff --git a/docs/en/connector-v2/sink/Paimon.md b/docs/en/connector-v2/sink/Paimon.md
@@ -4,48 +4,84 @@
 
 ## Description
 
-Write data to Apache Paimon.
+Sink connector for Apache Paimon. It can support cdc mode 、auto create table.
 
 ## Key features
 
 - [x] [exactly-once](../../concept/connector-v2-features.md)
 
 ## Options
 
-|      name      |  type  | required | default value |
-|----------------|--------|----------|---------------|
-| warehouse      | String | Yes      | -             |
-| database       | String | Yes      | -             |
-| table          | String | Yes      | -             |
-| hdfs_site_path | String | No       | -             |
+|       name       |  type  | required |        default value         |           Description           |
+|------------------|--------|----------|------------------------------|---------------------------------|
+| warehouse        | String | Yes      | -                            | Paimon warehouse path           |
+| database         | String | Yes      | -                            | The database you want to access |
+| table            | String | Yes      | -                            | The table you want to access    |
+| hdfs_site_path   | String | No       | -                            |                                 |
+| schema_save_mode | Enum   | no       | CREATE_SCHEMA_WHEN_NOT_EXIST | The schema save mode            |
+| data_save_mode   | Enum   | no       | APPEND_DATA                  | The data save mode              |
 
-### warehouse [string]
-
-Paimon warehouse path
-
-### database [string]
+## Examples
 
-The database you want to access
+### Single table
 
-### table [String]
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
 
-The table you want to access
+source {
+  Mysql-CDC {
+    base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+    username = "root"
+    password = "******"
+    table-names = ["seatunnel.role"]
+  }
+}
 
-## Examples
+transform {
+}
 
-```hocon
 sink {
   Paimon {
-    warehouse = "/tmp/paimon"
-    database = "default"
-    table = "st_test"
+    catalog_name="seatunnel_test"
+    warehouse="file:///tmp/seatunnel/paimon/hadoop-sink/"
+    database="seatunnel"
+    table="role"
   }
 }
 ```
 
-## Changelog
+### Multiple table
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  Mysql-CDC {
+    base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+    username = "root"
+    password = "******"
+    table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]
+  }
+}
 
-### next version
+transform {
+}
 
-- Add Paimon Sink Connector
+sink {
+  Paimon {
+    catalog_name="seatunnel_test"
+    warehouse="file:///tmp/seatunnel/paimon/hadoop-sink/"
+    database="${database_name}"
+    table="${table_name}"
+  }
+}
+```
 
diff --git a/docs/zh/connector-v2/sink/Paimon.md b/docs/zh/connector-v2/sink/Paimon.md
@@ -0,0 +1,87 @@
+# Paimon
+
+> Paimon 数据连接器
+
+## 描述
+
+Apache Paimon数据连接器。支持cdc写以及自动建表。
+
+## 主要特性
+
+- [x] [exactly-once](../../concept/connector-v2-features.md)
+
+## 连接器选项
+
+|        名称        |   类型   | 是否必须 |             默认值              |         描述         |
+|------------------|--------|------|------------------------------|--------------------|
+| warehouse        | String | Yes  | -                            | Paimon warehouse路径 |
+| database         | String | Yes  | -                            | 数据库名称              |
+| table            | String | Yes  | -                            | 表名                 |
+| hdfs_site_path   | String | No   | -                            |                    |
+| schema_save_mode | Enum   | no   | CREATE_SCHEMA_WHEN_NOT_EXIST | schema保存模式         |
+| data_save_mode   | Enum   | no   | APPEND_DATA                  | 数据保存模式             |
+
+## 示例
+
+### 单表
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  Mysql-CDC {
+    base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+    username = "root"
+    password = "******"
+    table-names = ["seatunnel.role"]
+  }
+}
+
+transform {
+}
+
+sink {
+  Paimon {
+    catalog_name="seatunnel_test"
+    warehouse="file:///tmp/seatunnel/paimon/hadoop-sink/"
+    database="seatunnel"
+    table="role"
+  }
+}
+```
+
+### 多表
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  Mysql-CDC {
+    base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+    username = "root"
+    password = "******"
+    table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]
+  }
+}
+
+transform {
+}
+
+sink {
+  Paimon {
+    catalog_name="seatunnel_test"
+    warehouse="file:///tmp/seatunnel/paimon/hadoop-sink/"
+    database="${database_name}"
+    table="${table_name}"
+  }
+}
+```
+
diff --git a/seatunnel-connectors-v2/connector-paimon/pom.xml b/seatunnel-connectors-v2/connector-paimon/pom.xml
@@ -34,6 +34,11 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
 
         <dependency>
             <groupId>org.apache.paimon</groupId>