[Improve][Paimon] Replace the test case by fakesource and add doc

apache · Mar 11, 2024 · 937c7aa · 937c7aa
1 parent 244d927
commit 937c7aa
Show file tree

Hide file tree

Showing 8 changed files with 457 additions and 355 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/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkFactory.java b/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSinkFactory.java
@@ -49,10 +49,11 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder()
-                .required(PaimonConfig.WAREHOUSE)
-                .required(PaimonConfig.DATABASE)
-                .required(PaimonConfig.TABLE)
-                .optional(PaimonConfig.HDFS_SITE_PATH)
+                .required(PaimonConfig.WAREHOUSE, PaimonConfig.DATABASE, PaimonConfig.TABLE)
+                .optional(
+                        PaimonConfig.HDFS_SITE_PATH,
+                        PaimonSinkConfig.SCHEMA_SAVE_MODE,
+                        PaimonSinkConfig.DATA_SAVE_MODE)
                 .build();
     }