This is an automated email from the ASF dual-hosted git repository.
zhangshenghang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 13547dfa8a [Docs][Connector-V2] Improve MongoDB OpenGauss and
PostgreSQL CDC docs (#11324)
13547dfa8a is described below
commit 13547dfa8a81ba5af71107523026feadcfd6ea74
Author: Daniel Carter <[email protected]>
AuthorDate: Sat Jul 4 09:54:50 2026 +0800
[Docs][Connector-V2] Improve MongoDB OpenGauss and PostgreSQL CDC docs
(#11324)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/source/MongoDB-CDC.md | 101 ++++++++++++++++++--------
docs/en/connectors/source/Opengauss-CDC.md | 49 ++++++++++---
docs/en/connectors/source/PostgreSQL-CDC.md | 58 +++++++++++----
docs/zh/connectors/source/MongoDB-CDC.md | 107 +++++++++++++++++++---------
docs/zh/connectors/source/Opengauss-CDC.md | 47 +++++++++---
docs/zh/connectors/source/PostgreSQL-CDC.md | 54 ++++++++++----
6 files changed, 306 insertions(+), 110 deletions(-)
diff --git a/docs/en/connectors/source/MongoDB-CDC.md
b/docs/en/connectors/source/MongoDB-CDC.md
index 03d3bc221d..165cef3005 100644
--- a/docs/en/connectors/source/MongoDB-CDC.md
+++ b/docs/en/connectors/source/MongoDB-CDC.md
@@ -33,15 +33,15 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
## Availability Settings
-1.MongoDB version: MongoDB version >= 4.0.
+1. MongoDB version: MongoDB version >= 4.0.
-2.Cluster deployment: replica sets or sharded clusters.
+2. Cluster deployment: replica sets or sharded clusters.
-3.Storage Engine: WiredTiger Storage Engine.
+3. Storage Engine: WiredTiger Storage Engine.
-4.Permissions:changeStream and read
+4. Permissions: `changeStream` and `read`.
-```
+```javascript
// 1) Switch to the target database
use <DB_NAME>
@@ -112,25 +112,26 @@ For specific types in MongoDB, we use Extended JSON
format to map them to Seatun
## Source Options
-| Name | Type | Required | Default |
Description
|
-|------------------------------------|--------|----------|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| hosts | String | Yes | - | The
comma-separated list of hostname and port pairs of the MongoDB servers, or a
standard MongoDB connection URI using `mongodb://` or `mongodb+srv://`. eg.
`localhost:27017,localhost:27018` or `mongodb+srv://cluster.example.net`
|
-| username | String | No | - | Name of
the database user to be used when connecting to MongoDB.
|
-| password | String | No | - | Password
to be used when connecting to MongoDB.
|
-| database | List | Yes | - | Name of
the database to watch for changes. If not set then all databases will be
captured. The database also supports regular expressions to monitor multiple
databases matching the regular expression. eg. `db1,db2`.
|
-| collection | List | Yes | - | Name of
the collection in the database to watch for changes. If not set then all
collections will be captured. The collection also supports regular expressions
to monitor multiple collections matching fully-qualified collection
identifiers. eg. `db1.coll1,db2.coll2`. |
-| schema | | no | - | The
structure of the data, including field names and field types, use single table
cdc. For more details, please refer to [Schema
Feature](../../introduction/concepts/schema-feature.md).
|
-| tables_configs | | no | - | The
structure of the data, including field names and field types, use muliti table
cdc.
|
-| connection.options | String | No | - | The
ampersand-separated connection options of MongoDB. eg.
`replicaSet=test&connectTimeoutMS=300000`.
|
-| batch.size | Long | No | 1024 | The
cursor batch size.
|
-| poll.max.batch.size | Enum | No | 1024 | Maximum
number of change stream documents to include in a single batch when polling for
new data.
|
-| poll.await.time.ms | Long | No | 1000 | The
amount of time to wait before checking for new results on the change stream.
|
-| heartbeat.interval.ms | String | No | 0 | The
length of time in milliseconds between sending heartbeat messages. Use 0 to
disable.
|
-| incremental.snapshot.chunk.size.mb | Long | No | 64 | The chunk
size mb of incremental snapshot.
|
-| startup.mode | Enum | No | INITIAL | Optional
startup mode for MongoDB CDC consumer, valid enumerations are `initial`,
`latest` and `timestamp`. See the [Startup Mode](#startup-mode) section below.
|
-| startup.timestamp | Long | No | - | Start
from the specified epoch timestamp (in milliseconds). Only used when
`startup.mode` is `timestamp`.
|
-| exactly_once | Boolean| No | false | Enable
exactly once semantic. Enabling this may cause an out-of-memory risk during the
large table snapshot stage in recovery.
|
-| common-options | | No | - | Source
plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
|
+| Name | Type | Required | Default |
Description
|
+|------------------------------------|---------|----------|---------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| hosts | String | Yes | - | The
comma-separated list of hostname and port pairs of the MongoDB servers, or a
standard MongoDB connection URI using `mongodb://` or `mongodb+srv://`. For
example: `localhost:27017,localhost:27018` or
`mongodb+srv://cluster.example.net`. |
+| username | String | No | - | Name of
the database user to be used when connecting to MongoDB. Required only when
MongoDB authentication is enabled.
|
+| password | String | No | - | Password
to be used when connecting to MongoDB. Required only when MongoDB
authentication is enabled.
|
+| database | List | Yes | - | Database
names to watch for changes. Regular expressions are supported, for example
`["inventory"]` or `["db.*"]`.
|
+| collection | List | Yes | - |
Collection names to watch for changes. Each value should use the fully
qualified `database.collection` format, for example `["inventory.products",
"inventory.orders"]`. Regular expressions are supported.
|
+| schema | Config | No | - | Schema
for one collection, including field names, field types, and optional primary
key. For more details, see [Schema
Feature](../../introduction/concepts/schema-feature.md).
|
+| tables_configs | List | No | - | Schema
list for multiple collections. Each item contains a `schema` block. The number
and order of `tables_configs` items must match the `collection` list.
|
+| connection.options | String | No | - |
Ampersand-separated MongoDB connection options. For example:
`replicaSet=test&connectTimeoutMS=300000`.
|
+| batch.size | Integer | No | 1024 | Cursor
batch size used when reading snapshot data.
|
+| poll.max.batch.size | Integer | No | 1024 | Maximum
number of change stream documents to include in one poll batch.
|
+| poll.await.time.ms | Integer | No | 1000 | Time in
milliseconds to wait before checking for new change stream results.
|
+| heartbeat.interval.ms | Integer | No | 0 | Time in
milliseconds between heartbeat messages. Use `0` to disable heartbeat messages.
|
+| incremental.snapshot.chunk.size.mb | Integer | No | 64 | Chunk
size, in MB, for incremental snapshot reading.
|
+| startup.mode | Enum | No | INITIAL | Optional
startup mode for MongoDB CDC consumer. Valid values are `initial`, `latest`,
and `timestamp`. See the [Startup Mode](#startup-mode) section below.
|
+| startup.timestamp | Long | No | - | Start
from the specified epoch timestamp in milliseconds. Only used when
`startup.mode` is `timestamp`.
|
+| exactly_once | Boolean | No | false | Enable
exactly-once semantics. Enabling this may increase memory usage during large
table snapshot recovery.
|
+| debezium | Config | No | - |
Pass-through Debezium properties used by the embedded engine.
|
+| common-options | | No | - | Source
plugin common parameters. For details, see [Source Common
Options](../common-options/source-common-options.md).
|
### Startup Mode
@@ -168,7 +169,7 @@ source {
> 1.If the collection changes at a slow pace, it is strongly recommended to
> set an appropriate value greater than 0 for the heartbeat.interval.ms
> parameter. When we recover a Seatunnel job from a checkpoint or savepoint,
> the heartbeat events can push the resumeToken forward to avoid its
> expiration.<br/>
> 2.MongoDB has a limit of 16MB for a single document. Change documents
> include additional information, so even if the original document is not
> larger than 15MB, the change document may exceed the 16MB limit, resulting
> in the termination of the Change Stream operation.<br/>
> 3.It is recommended to use immutable shard keys. In MongoDB, shard keys
> allow modifications after transactions are enabled, but changing the shard
> key can cause frequent shard migrations, resulting in additional performance
> overhead. Additionally, modifying the shard key can also cause the Update
> Lookup feature to become ineffective, leading to inconsistent results in CDC
> (Change Data Capture) scenarios.<br/>
-> 4.`schema` `tables_configs` are mutually exclusive, and one must be
configured at a time.
+> 4. `schema` and `tables_configs` are mutually exclusive. Use `schema` for
one collection and `tables_configs` for multiple collections.
## Change Streams
@@ -177,7 +178,7 @@ Applications can use change streams to subscribe to all
data changes on a single
**Lookup Full Document for Update Operations** is a feature provided by
**Change Stream** which can configure the change stream to return the most
current majority-committed version of the updated document. Because of this
feature, we can easily collect the latest full document and convert the change
log to Changelog Stream.
-The format of the data captured by delete events in change streams: [delete
envet](https://www.mongodb.com/docs/v5.0/reference/change-events/delete/)
+The format of the data captured by delete events in change streams: [delete
event](https://www.mongodb.com/docs/v5.0/reference/change-events/delete/)
```
{
"_id": { <Resume Token> },
@@ -217,6 +218,10 @@ source {
password = stpw
schema = {
table = "inventory.products"
+ primaryKey {
+ name = "id"
+ columnNames = ["_id"]
+ }
fields {
"_id" : string,
"name" : string,
@@ -227,7 +232,6 @@ source {
}
}
-# Console printing of the read Mongodb data
sink {
Console {
parallelism = 1
@@ -270,7 +274,7 @@ sink {
jdbc {
url = "jdbc:mysql://mysql_cdc_e2e:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "st_user"
+ username = "st_user"
password = "seatunnel"
generate_sink_sql = true
@@ -284,7 +288,7 @@ sink {
## Multi-table Synchronization
-The following example demonstrates how to create a data synchronization job
that read the cdc data of multiple library tables mongodb and prints it on the
local client:
+The following example demonstrates how to read CDC data from multiple MongoDB
collections and write each collection to the matching MySQL table. The sink
table uses `${table_name}`, so `inventory.products` and `inventory.orders` are
routed to their own target tables.
```hocon
env {
@@ -329,9 +333,44 @@ source {
}
}
-# Console printing of the read Mongodb data
sink {
- Console {
+ jdbc {
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mongodb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user"
+ password = "seatunnel"
+ generate_sink_sql = true
+ database = mongodb_cdc
+ table = "${table_name}"
+ primary_keys = ["_id"]
+ }
+}
+```
+
+## CDC Metadata Fields
+
+MongoDB CDC exposes metadata fields that can be used by the `Metadata`
transform:
+
+| Field | Type | Description |
+|-------|------|-------------|
+| database | STRING | Source database name. |
+| table | STRING | Source collection name. |
+| rowKind | STRING | Change type, such as insert, update, or delete. |
+| ts_ms | LONG | Source event timestamp in milliseconds. |
+| delay | LONG | Delay between event time and processing time in milliseconds.
|
+
+Example:
+
+```hocon
+transform {
+ Metadata {
+ metadata_fields {
+ Database = database
+ Table = table
+ RowKind = rowKind
+ EventTime = ts_ms
+ Delay = delay
+ }
}
}
```
diff --git a/docs/en/connectors/source/Opengauss-CDC.md
b/docs/en/connectors/source/Opengauss-CDC.md
index f46407eb23..2d166a4647 100644
--- a/docs/en/connectors/source/Opengauss-CDC.md
+++ b/docs/en/connectors/source/Opengauss-CDC.md
@@ -20,8 +20,8 @@ import ChangeLog from
'../changelog/connector-cdc-opengauss.md';
## Description
-The Opengauss CDC connector allows for reading snapshot data and incremental
data from Opengauss database. This document
-describes how to set up the Opengauss CDC connector to run SQL queries against
Opengauss databases.
+The Opengauss CDC connector allows for reading snapshot data and incremental
data from Opengauss databases. This document
+describes how to set up the Opengauss CDC connector.
## Using steps
@@ -71,13 +71,14 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename ||
' REPLICA IDENTITY FU
| url | String | Yes | - |
The URL of the JDBC connection. Refer to a case:
`jdbc:postgresql://localhost:5432/postgres_cdc?loggerLevel=OFF`.
[...]
| username | String | Yes | - |
Username of the database to use when connecting to the database server.
[...]
| password | String | Yes | - |
Password to use when connecting to the database server.
[...]
-| database-names | List | No | - |
Database name of the database to monitor.
[...]
-| table-names | List | Yes | - |
Table name of the database to monitor. The table name needs to include the
database name, for example: `database_name.table_name`
[...]
-| table-names-config | List | No | - |
Table config list. for example: [{"table":
"db1.schema1.table1","primaryKeys":["key1"]}]
[...]
+| database-names | List | No | - |
Database names to monitor.
[...]
+| table-names | List | Yes, if
`table-pattern` is not used | - | Tables to monitor. Use the fully
qualified `database.schema.table` format, for example:
`opengauss_cdc.inventory.orders`.
[...]
+| table-pattern | String | Yes, if `table-names`
is not used | - | Regular expression for tables to monitor. Use the
fully qualified table name in the pattern, for example:
`opengauss_cdc\\.inventory\\..*`. `table-names` and `table-pattern` are
mutually exclusive.
[...]
+| table-names-config | List | No | - |
Per-table config list. Example: `[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`. Use `primaryKeys` for tables without
a physical primary key. `snapshotSplitColumn` must be a unique key; otherwise
SeaTunnel ignores it and selects a split column internally.
[...]
| startup.mode | Enum | No | INITIAL |
Optional startup mode for Opengauss CDC consumer, valid enumerations are
`initial`, `earliest`, `latest`. <br/> `initial`: Synchronize historical data
at startup, and then synchronize incremental data.<br/> `earliest`: Startup
from the earliest offset possible.<br/> `latest`: Startup from the latest
offset.
[...]
| snapshot.split.size | Integer | No | 8096 |
The split size (number of rows) of table snapshot, captured tables are split
into multiple splits when read the snapshot of table.
[...]
| snapshot.fetch.size | Integer | No | 1024 |
The maximum fetch size for per poll when read table snapshot.
[...]
-| slot.name | String | No | - |
The name of the Opengauss logical decoding slot that was created for streaming
changes from a particular plug-in for a particular database/schema. The server
uses this slot to stream events to the connector that you are configuring.
Default is seatunnel.
[...]
+| slot.name | String | No | seatunnel
| The Opengauss logical decoding slot name. Use a different slot name for each
CDC job that reads from the same Opengauss instance.
[...]
| decoding.plugin.name | String | No | pgoutput |
The name of the Postgres logical decoding plug-in installed on the
server,Supported values are decoderbufs, wal2json, wal2json_rds,
wal2json_streaming,wal2json_rds_streaming and pgoutput.
[...]
| server-time-zone | String | No | UTC |
The session time zone in database server. If not set, then
ZoneId.systemDefault() is used to determine the server time zone.
[...]
| connect.timeout.ms | Duration | No | 30000 |
The maximum time that the connector should wait after trying to connect to the
database server before timing out.
[...]
@@ -88,7 +89,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| sample-sharding.threshold | Integer | No | 1000 |
This configuration specifies the threshold of estimated shard count to trigger
the sample sharding strategy. When the distribution factor is outside the
bounds specified by `chunk-key.even-distribution.factor.upper-bound` and
`chunk-key.even-distribution.factor.lower-bound`, and the estimated shard count
(calculated as approximate row count / chunk size) exceeds this threshold, the
sample sharding strategy wil [...]
| inverse-sampling.rate | Integer | No | 1000 |
The inverse of the sampling rate used in the sample sharding strategy. For
example, if this value is set to 1000, it means a 1/1000 sampling rate is
applied during the sampling process. This option provides flexibility in
controlling the granularity of the sampling, thus affecting the final number of
shards. It's especially useful when dealing with very large datasets where a
lower sampling rate is preferred. [...]
| split.allow-sampling | Boolean | No | true |
Whether to allow sampling-based sharding strategy. When set to false, the
system will fall back to unevenly-sized chunk splitting (iterative query
approach) regardless of the shard count. The default value is true. |
-| exactly_once | Boolean | No | false |
Enable exactly once semantic.
[...]
+| exactly_once | Boolean | No | false |
Enable exactly-once semantics during the initial snapshot phase. This option is
only available when `startup.mode` is `initial`.
[...]
| format | Enum | No | DEFAULT |
Optional output format for Opengauss CDC, valid enumerations are `DEFAULT`,
`COMPATIBLE_DEBEZIUM_JSON`.
[...]
| debezium | Config | No | - |
Pass-through [Debezium's
properties](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
to Debezium Embedded Engine which is used to capture data changes from
Opengauss server.
[...]
| common-options | | no | - |
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
[...]
@@ -116,10 +117,10 @@ source {
username = "gaussdb"
password = "openGauss@123"
database-names = ["opengauss_cdc"]
- schema-names = ["inventory"]
table-names =
["opengauss_cdc.inventory.opengauss_cdc_table_1","opengauss_cdc.inventory.opengauss_cdc_table_2"]
url = "jdbc:postgresql://opengauss_cdc_e2e:5432/opengauss_cdc"
decoding.plugin.name = "pgoutput"
+ slot.name = "seatunnel_opengauss_cdc"
}
}
@@ -156,11 +157,11 @@ source {
username = "gaussdb"
password = "openGauss@123"
database-names = ["opengauss_cdc"]
- schema-names = ["inventory"]
table-names = ["opengauss_cdc.inventory.full_types_no_primary_key"]
url =
"jdbc:postgresql://opengauss_cdc_e2e:5432/opengauss_cdc?loggerLevel=OFF"
decoding.plugin.name = "pgoutput"
exactly_once = true
+ slot.name = "seatunnel_opengauss_cdc"
table-names-config = [
{
table = "opengauss_cdc.inventory.full_types_no_primary_key"
@@ -171,6 +172,34 @@ source {
}
```
+## CDC Metadata Fields
+
+Opengauss CDC exposes metadata fields that can be used by the `Metadata`
transform:
+
+| Field | Type | Description |
+|-------|------|-------------|
+| database | STRING | Source database name. |
+| table | STRING | Source table name. |
+| rowKind | STRING | Change type, such as insert, update, or delete. |
+| ts_ms | LONG | Source event timestamp in milliseconds. |
+| delay | LONG | Delay between event time and processing time in milliseconds.
|
+
+Example:
+
+```hocon
+transform {
+ Metadata {
+ metadata_fields {
+ Database = database
+ Table = table
+ RowKind = rowKind
+ EventTime = ts_ms
+ Delay = delay
+ }
+ }
+}
+```
+
## Changelog
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md
b/docs/en/connectors/source/PostgreSQL-CDC.md
index 3a79cf32c8..ddaed8cb0d 100644
--- a/docs/en/connectors/source/PostgreSQL-CDC.md
+++ b/docs/en/connectors/source/PostgreSQL-CDC.md
@@ -20,8 +20,8 @@ import ChangeLog from
'../changelog/connector-cdc-postgres.md';
## Description
-The Postgre CDC connector allows for reading snapshot data and incremental
data from Postgre database. This document
-describes how to set up the Postgre CDC connector to run SQL queries against
Postgre databases.
+The PostgreSQL CDC connector allows for reading snapshot data and incremental
data from PostgreSQL databases. This document
+describes how to set up the PostgreSQL CDC connector.
## Supported DataSource Info
@@ -93,13 +93,14 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| url | String | Yes | - |
The URL of the JDBC connection. Refer to a case:
`jdbc:postgresql://localhost:5432/postgres_cdc?loggerLevel=OFF`.
[...]
| username | String | Yes | - |
Name of the database to use when connecting to the database server.
[...]
| password | String | Yes | - |
Password to use when connecting to the database server.
[...]
-| database-names | List | No | - |
Database name of the database to monitor.
[...]
-| table-names | List | Yes | - |
Table name of the database to monitor. The table name needs to include the
database name, for example: `database_name.table_name`
[...]
-| table-names-config | List | No | - |
Table config list. for example: [{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]. The snapshotSplitColumn option must
be configured with a unique key. If a non-unique column is provided, the
configuration is ignored and SeaTunnel automatically selects an appropriate
split column internally.
[...]
+| database-names | List | No | - |
Database names to monitor.
[...]
+| table-names | List | Yes, if
`table-pattern` is not used | - | Tables to monitor. Use the fully
qualified `database.schema.table` format, for example:
`postgres_cdc.inventory.orders`.
[...]
+| table-pattern | String | Yes, if `table-names`
is not used | - | Regular expression for tables to monitor. Use the
fully qualified table name in the pattern, for example:
`postgres_cdc\\.inventory\\..*`. `table-names` and `table-pattern` are mutually
exclusive.
[...]
+| table-names-config | List | No | - |
Per-table config list. Example: `[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`. Use `primaryKeys` for tables without
a physical primary key. `snapshotSplitColumn` must be a unique key; otherwise
SeaTunnel ignores it and selects a split column internally.
[...]
| startup.mode | Enum | No | INITIAL |
Optional startup mode for PostgreSQL CDC consumer, valid enumerations are
`initial`, `earliest` and `latest`. <br/> `initial`: Synchronize historical
data at startup, and then synchronize incremental data.<br/> `earliest`:
Startup from the earliest offset possible.<br/> `latest`: Startup from the
latest offset.
[...]
| snapshot.split.size | Integer | No | 8096 |
The split size (number of rows) of table snapshot, captured tables are split
into multiple splits when read the snapshot of table.
[...]
| snapshot.fetch.size | Integer | No | 1024 |
The maximum fetch size for per poll when read table snapshot.
[...]
-| slot.name | String | No | - |
The name of the PostgreSQL logical decoding slot that was created for streaming
changes from a particular plug-in for a particular database/schema. The server
uses this slot to stream events to the connector that you are configuring.
Default is seatunnel.
[...]
+| slot.name | String | No | seatunnel
| The PostgreSQL logical decoding slot name. Use a different slot name for each
CDC job that reads from the same PostgreSQL instance.
[...]
| decoding.plugin.name | String | No | pgoutput |
The name of the Postgres logical decoding plug-in installed on the
server,Supported values are decoderbufs, wal2json, wal2json_rds,
wal2json_streaming,wal2json_rds_streaming and pgoutput.
[...]
| server-time-zone | String | No | UTC |
The session time zone in database server. If not set, then
ZoneId.systemDefault() is used to determine the server time zone.
[...]
| connect.timeout.ms | Duration | No | 30000 |
The maximum time that the connector should wait after trying to connect to the
database server before timing out.
[...]
@@ -110,7 +111,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| sample-sharding.threshold | Integer | No | 1000 |
This configuration specifies the threshold of estimated shard count to trigger
the sample sharding strategy. When the distribution factor is outside the
bounds specified by `chunk-key.even-distribution.factor.upper-bound` and
`chunk-key.even-distribution.factor.lower-bound`, and the estimated shard count
(calculated as approximate row count / chunk size) exceeds this threshold, the
sample sharding strategy wil [...]
| inverse-sampling.rate | Integer | No | 1000 |
The inverse of the sampling rate used in the sample sharding strategy. For
example, if this value is set to 1000, it means a 1/1000 sampling rate is
applied during the sampling process. This option provides flexibility in
controlling the granularity of the sampling, thus affecting the final number of
shards. It's especially useful when dealing with very large datasets where a
lower sampling rate is preferred. [...]
| split.allow-sampling | Boolean | No | true |
Whether to allow sampling-based sharding strategy. When set to false, the
system will fall back to unevenly-sized chunk splitting (iterative query
approach) regardless of the shard count. The default value is true. |
-| exactly_once | Boolean | No | false |
Enable exactly once semantic.
[...]
+| exactly_once | Boolean | No | false |
Enable exactly-once semantics during the initial snapshot phase. This option is
only available when `startup.mode` is `initial`.
[...]
| format | Enum | No | DEFAULT |
Optional output format for PostgreSQL CDC, valid enumerations are `DEFAULT`,
`COMPATIBLE_DEBEZIUM_JSON`.
[...]
| require-replica-identity-full | Boolean | No | true |
Require the table to have REPLICA IDENTITY FULL. When set to false, allows
tables with other replica identity settings, but UPDATE/DELETE events may not
contain the previous state. This should only be used for append-only tables
(e.g., outbox pattern). Default is true for backward compatibility.
[...]
| debezium | Config | No | - |
Pass-through [Debezium's
properties](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
to Debezium Embedded Engine which is used to capture data changes from
PostgreSQL server.
[...]
@@ -136,13 +137,14 @@ env {
source {
Postgres-CDC {
- plugin_output = "customers_Postgre_cdc"
+ plugin_output = "customers_postgres_cdc"
username = "postgres"
password = "postgres"
database-names = ["postgres_cdc"]
- schema-names = ["inventory"]
- table-names =
["postgres_cdc.inventory.postgres_cdc_table_1,postgres_cdc.inventory.postgres_cdc_table_2"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1",
"postgres_cdc.inventory.postgres_cdc_table_2"]
url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = "seatunnel_postgres_cdc"
}
}
@@ -152,7 +154,7 @@ transform {
sink {
jdbc {
- plugin_input = "customers_Postgre_cdc"
+ plugin_input = "customers_postgres_cdc"
url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
driver = "org.postgresql.Driver"
username = "postgres"
@@ -173,15 +175,15 @@ sink {
```
source {
Postgres-CDC {
- plugin_output = "customers_mysql_cdc"
+ plugin_output = "customers_postgres_cdc"
username = "postgres"
password = "postgres"
database-names = ["postgres_cdc"]
- schema-names = ["inventory"]
table-names = ["postgres_cdc.inventory.full_types_no_primary_key"]
url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
decoding.plugin.name = "decoderbufs"
- exactly_once = false
+ exactly_once = true
+ slot.name = "seatunnel_postgres_cdc"
table-names-config = [
{
table = "postgres_cdc.inventory.full_types_no_primary_key"
@@ -192,6 +194,34 @@ source {
}
```
+## CDC Metadata Fields
+
+PostgreSQL CDC exposes metadata fields that can be used by the `Metadata`
transform:
+
+| Field | Type | Description |
+|-------|------|-------------|
+| database | STRING | Source database name. |
+| table | STRING | Source table name. |
+| rowKind | STRING | Change type, such as insert, update, or delete. |
+| ts_ms | LONG | Source event timestamp in milliseconds. |
+| delay | LONG | Delay between event time and processing time in milliseconds.
|
+
+Example:
+
+```hocon
+transform {
+ Metadata {
+ metadata_fields {
+ Database = database
+ Table = table
+ RowKind = rowKind
+ EventTime = ts_ms
+ Delay = delay
+ }
+ }
+}
+```
+
## FAQ
### What PostgreSQL permissions are required for CDC?
diff --git a/docs/zh/connectors/source/MongoDB-CDC.md
b/docs/zh/connectors/source/MongoDB-CDC.md
index 9f0362608e..f6e791e7e2 100644
--- a/docs/zh/connectors/source/MongoDB-CDC.md
+++ b/docs/zh/connectors/source/MongoDB-CDC.md
@@ -27,21 +27,21 @@ MongoDB CDC连接器允许从MongoDB数据库读取快照数据和增量数据
为了使用Mongodb CDC连接器，需要以下依赖关系。
它们可以通过install-plugin.sh或Maven中央存储库下载。
-| 数据源 | 支持的版本 | Dependency
|
-|------------|--------------------|-------------------------------------------------------------------------------------------|
-| MongoDB | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-cdc-mongodb)
|
+| 数据源 | 支持的版本 | 依赖
|
+|--------|------------|---------------------------------------------------------------------------------------|
+| MongoDB | universal |
[下载](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-cdc-mongodb)
|
## 可用性设置
-1.MongoDB版本：MongoDB版本>=4.0。
+1. MongoDB 版本：MongoDB 版本 >= 4.0。
-2.集群部署：副本集或分片集群。
+2. 集群部署：副本集或分片集群。
-3.存储引擎：WiredTiger存储引擎。
+3. 存储引擎：WiredTiger 存储引擎。
-4.权限：更改流和读取
+4. 权限：`changeStream` 和 `read`。
-```
+```javascript
// 1) 切换到目标数据库
use <DB_NAME>
@@ -112,25 +112,26 @@ db.grantRolesToUser("<USER_NAME>", ["<ROLE_NAME>"])
## 源配置项
-| Name | 类型 | 必须 | 默认值 | 描述
|
-|------------------------------------|--------|----------|-------|---------------------------------------------------------------------------------------|
-| hosts | String | 是 | - |
MongoDB服务器的主机名和端口对的逗号分隔列表，也可以是使用 `mongodb://` 或 `mongodb+srv://` 的标准 MongoDB 连接
URI。如 `localhost:27017,localhost:27018` 或 `mongodb+srv://cluster.example.net`
|
-| username | String | 否 | - |
连接到MongoDB时要使用的数据库用户的名称。
|
-| password | String | 否 | - |
连接到MongoDB时使用的密码。
|
-| database | List | 是 | - |
要监视更改的数据库的名称。如果未设置，则将捕获所有数据库。该数据库还支持正则表达式，以监视与正则表达式匹配的多个数据库。例如db1、db2。
|
-| collection | List | 是 | - |
要监视更改的数据库中集合的名称。如果未设置，则将捕获所有集合。该集合还支持正则表达式来监视与完全限定的集合标识符匹配的多个集合。例如db1.coll1、db2.coll2。
|
-| schema | | 否 | - |
数据的结构，包括字段名和字段类型，使用单表cdc。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
-| tables_configs | | 否 | - |
数据的结构，包括字段名和字段类型，使用多表cdc。
|
-| connection.options | String | 否 | - |
与号分隔了MongoDB的连接选项。如。 `replicaSet=test&connectTimeoutMS=300000`.
|
-| batch.size | Long | 否 | 1024 | 批量大小。
|
-| poll.max.batch.size | Enum | 否 | 1024 |
轮询新数据时，单个批中包含的更改流文档的最大数量。
|
-| poll.await.time.ms | Long | 否 | 1000 |
在检查更改流上的新结果之前等待的时间量。
|
-| heartbeat.interval.ms | String | 否 | 0 |
发送心跳消息之间的时间长度（毫秒）。使用0禁用。
|
-| incremental.snapshot.chunk.size.mb | Long | 否 | 64 |
增量快照的块大小（mb）。
|
-| startup.mode | Enum | 否 | INITIAL | MongoDB
CDC 消费者的可选启动模式，有效枚举为 `initial`、`latest` 和 `timestamp`。详见下方[启动模式](#启动模式)章节。
|
-| startup.timestamp | Long | 否 | - |
从指定的纪元时间戳（毫秒）开始消费。仅在 `startup.mode` 为 `timestamp` 时使用。
|
-| exactly_once | Boolean| 否 | false |
启用精确一次语义，若开启在大表快照阶段恢复时会有内存溢出风险。
|
-| common-options | | 否 | - | 源插件常用参数，请参考
[Source Common Options](../common-options/source-common-options.md)
|
+| 名称 | 类型 | 必填 | 默认值 | 描述
|
+|------------------------------------|---------|------|---------|----------------------------------------------------------------------------------------------------------------------------|
+| hosts | String | 是 | - | MongoDB
服务器的主机名和端口列表，也可以是使用 `mongodb://` 或 `mongodb+srv://` 的标准 MongoDB 连接
URI。例如：`localhost:27017,localhost:27018` 或 `mongodb+srv://cluster.example.net`。
|
+| username | String | 否 | - | 连接 MongoDB
时使用的数据库用户名。仅在 MongoDB 开启认证时需要。
|
+| password | String | 否 | - | 连接 MongoDB
时使用的密码。仅在 MongoDB 开启认证时需要。
|
+| database | List | 是 | - |
要监听的数据库名称，支持正则表达式，例如 `["inventory"]` 或 `["db.*"]`。
|
+| collection | List | 是 | - |
要监听的集合名称，建议使用完整的 `database.collection` 格式，例如 `["inventory.products",
"inventory.orders"]`，支持正则表达式。 |
+| schema | Config | 否 | - | 单集合的 Schema
配置，包含字段名、字段类型和可选主键。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。 |
+| tables_configs | List | 否 | - | 多集合的 Schema
配置列表，每个元素包含一个 `schema` 块。`tables_configs` 的数量和顺序必须与 `collection` 列表一致。
|
+| connection.options | String | 否 | - | 使用 `&` 分隔的
MongoDB 连接参数，例如 `replicaSet=test&connectTimeoutMS=300000`。
|
+| batch.size | Integer | 否 | 1024 |
读取快照数据时的游标批大小。
|
+| poll.max.batch.size | Integer | 否 | 1024 |
轮询新数据时，单个批次中包含的变更流文档最大数量。
|
+| poll.await.time.ms | Integer | 否 | 1000 |
检查变更流新结果前等待的时间，单位为毫秒。
|
+| heartbeat.interval.ms | Integer | 否 | 0 |
发送心跳消息的间隔，单位为毫秒。设置为 `0` 表示禁用心跳。
|
+| incremental.snapshot.chunk.size.mb | Integer | 否 | 64 |
增量快照读取时的分片大小，单位为 MB。
|
+| startup.mode | Enum | 否 | INITIAL | MongoDB CDC
的启动模式，可选值为 `initial`、`latest` 和 `timestamp`。详见下方[启动模式](#启动模式)。
|
+| startup.timestamp | Long | 否 | - |
从指定的毫秒级时间戳开始消费。仅在 `startup.mode` 为 `timestamp` 时使用。
|
+| exactly_once | Boolean | 否 | false |
启用精确一次语义。开启后，大表快照阶段恢复时可能增加内存使用。
|
+| debezium | Config | 否 | - | 透传给内嵌
Debezium 引擎的配置。
|
+| common-options | | 否 | - | 源插件通用参数，请参考
[源通用选项](../common-options/source-common-options.md)。
|
### 启动模式
@@ -168,7 +169,7 @@ source {
> 1.如果集合更改速度较慢，强烈建议为heartbeat.interval.ms参数设置一个大于0的适当值。当我们从检查点或保存点恢复Seatunnel作业时，心跳事件可以向前推resumeToken以避免其过期。<br/>
> 2.MongoDB对单个文档的限制为16MB。变更文档包含其他信息，因此即使原始文档不超过15MB，变更文档也可能超过16MB的限制，从而导致变更流操作终止。<br/>
> 3.建议使用不可变分片键。在MongoDB中，分片键允许在启用事务后进行修改，但更改分片键可能会导致频繁的分片迁移，从而导致额外的性能开销。此外，修改分片键也可能导致更新查找功能失效，从而导致CDC（变更数据捕获）场景中的结果不一致。<br/>
-> 4.“schema”和“tables_config”是互斥的，必须一次配置一个。
+> 4. `schema` 和 `tables_configs` 互斥。单集合使用 `schema`，多集合使用 `tables_configs`。
## 更新数据的流
@@ -177,7 +178,7 @@ source {
**查找更新操作的完整文档**是**更改流**提供的一项功能，它可以配置更改流以返回更新文档的最新多数提交版本。由于此功能，我们可以轻松收集最新的完整文档，并将更改日志转换为Changelog流。
-更新流中删除事件捕获的数据格式：[delete
envet](https://www.mongodb.com/docs/v5.0/reference/change-events/delete/)
+更新流中删除事件捕获的数据格式：[delete
事件](https://www.mongodb.com/docs/v5.0/reference/change-events/delete/)
```
{
"_id": { <Resume Token> },
@@ -217,6 +218,10 @@ source {
password = stpw
schema = {
table = "inventory.products"
+ primaryKey {
+ name = "id"
+ columnNames = ["_id"]
+ }
fields {
"_id" : string,
"name" : string,
@@ -227,7 +232,6 @@ source {
}
}
-# 控制台打印读取的Mongodb数据
sink {
Console {
parallelism = 1
@@ -270,7 +274,7 @@ sink {
jdbc {
url = "jdbc:mysql://mysql_cdc_e2e:3306"
driver = "com.mysql.cj.jdbc.Driver"
- user = "st_user"
+ username = "st_user"
password = "seatunnel"
generate_sink_sql = true
@@ -284,7 +288,7 @@ sink {
## 多表同步
-以下示例演示了如何创建数据同步作业，该作业读取多个库表mongodb的cdc数据并将其打印到本地客户端：
+以下示例演示了如何读取多个 MongoDB 集合的 CDC 数据，并将每个集合写入对应的 MySQL 表。Sink 表名使用
`${table_name}`，因此 `inventory.products` 和 `inventory.orders` 会分别写入自己的目标表。
```hocon
env {
@@ -329,9 +333,44 @@ source {
}
}
-# 控制台打印读取的Mongodb数据
sink {
- Console {
+ jdbc {
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mongodb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user"
+ password = "seatunnel"
+ generate_sink_sql = true
+ database = mongodb_cdc
+ table = "${table_name}"
+ primary_keys = ["_id"]
+ }
+}
+```
+
+## CDC 元数据字段
+
+MongoDB CDC 会提供以下元数据字段，可配合 `Metadata` 转换使用：
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| database | STRING | 源数据库名称。 |
+| table | STRING | 源集合名称。 |
+| rowKind | STRING | 变更类型，例如 insert、update 或 delete。 |
+| ts_ms | LONG | 源事件时间，单位为毫秒。 |
+| delay | LONG | 事件时间和处理时间之间的延迟，单位为毫秒。 |
+
+示例：
+
+```hocon
+transform {
+ Metadata {
+ metadata_fields {
+ Database = database
+ Table = table
+ RowKind = rowKind
+ EventTime = ts_ms
+ Delay = delay
+ }
}
}
```
diff --git a/docs/zh/connectors/source/Opengauss-CDC.md
b/docs/zh/connectors/source/Opengauss-CDC.md
index 4c9d4667b1..c3c0cfe874 100644
--- a/docs/zh/connectors/source/Opengauss-CDC.md
+++ b/docs/zh/connectors/source/Opengauss-CDC.md
@@ -20,7 +20,7 @@ import ChangeLog from
'../changelog/connector-cdc-opengauss.md';
## 描述
-Opengauss CDC连接器允许从Opengauss数据库读取快照数据和增量数据。这个文档描述如何设置Opengauss
CDC连接器以在Opengauss database中运行SQL查询。
+Opengauss CDC 连接器允许从 Opengauss 数据库读取快照数据和增量数据。本文档介绍如何配置 Opengauss CDC 连接器。
## 使用步骤
@@ -70,13 +70,14 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename ||
' REPLICA IDENTITY FU
| url | 字符串 | 是 | - |
JDBC连接的URL. 参考:
`jdbc:postgresql://localhost:5432/postgres_cdc?loggerLevel=OFF`.
|
| username | 字符串 | 是 | - |
连接数据库的用户名
|
| password | 字符串 | 是 | - |
连接数据库的密码
|
-| database-names | 列表 | 否 | - |
监控的数据库名称
|
-| table-names | 列表 | 是 | - |
监控的数据表名称. 表名需要包含数据库名称, 例如: `database_name.table_name`
|
-| table-names-config | 列表 | 否 | - |
表配置的列表集合. 例如: [{"table": "db1.schema1.table1","primaryKeys":["key1"]}]
|
+| database-names | 列表 | 否 | - |
需要监控的数据库名称。
|
+| table-names | 列表 | 二选一 | - |
需要监控的表。请使用完整的 `database.schema.table` 格式，例如：`opengauss_cdc.inventory.orders`。
|
+| table-pattern | 字符串 | 二选一 | - |
需要监控的表名正则表达式。正则需要匹配完整表名，例如：`opengauss_cdc\\.inventory\\..*`。`table-names` 和
`table-pattern` 互斥。
|
+| table-names-config | 列表 | 否 | - |
表级配置列表。例如：`[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`。无物理主键表可通过 `primaryKeys`
指定唯一键。`snapshotSplitColumn` 必须是唯一键，否则 SeaTunnel 会忽略该配置并自动选择拆分列。
|
| startup.mode | 枚举 | 否 | INITIAL |
Opengauss CDC消费者的可选启动模式, 有效的枚举是`initial`, `earliest`, `latest`. <br/>
`initial`: 启动时同步历史数据，然后同步增量数据 <br/> `earliest`: 从可能的最早偏移量启动 <br/> `latest`:
从最近的偏移量启动 |
| snapshot.split.size | 整型 | 否 | 8096 |
表快照的分割大小（行数），在读取表的快照时，捕获的表被分割成多个split
|
| snapshot.fetch.size | 整型 | 否 | 1024 |
读取表快照时，每次轮询的最大读取大小
|
-| slot.name | 字符串 | 否 | - |
Opengauss逻辑解码插槽的名称，该插槽是为特定数据库/模式的特定插件的流式更改而创建的。服务器使用此插槽将事件流传输到正在配置的连接器。默认值为seatunnel
|
+| slot.name | 字符串 | 否 | seatunnel |
Opengauss 逻辑解码槽名称。同一个 Opengauss 实例上如果有多个 CDC 任务，请为每个任务配置不同的 `slot.name`。
|
| decoding.plugin.name | 字符串 | 否 | pgoutput |
安装在服务器上的Postgres逻辑解码插件的名称，支持的值是decoderbufs、wal2json、wal2json_rds、wal2json_streaming、wal2json_rds_streaming和pgoutput
|
| server-time-zone | 字符串 | 否 | UTC |
数据库服务器中的会话时区。如果没有设置，则使用ZoneId.systemDefault()来确定服务器的时区
|
| connect.timeout.ms | 时间间隔 | 否 | 30000 |
在尝试连接数据库服务器之后，连接器在超时之前应该等待的最大时间
|
@@ -87,7 +88,7 @@ select 'ALTER TABLE ' || schemaname || '.' || tablename || '
REPLICA IDENTITY FU
| sample-sharding.threshold | 整型 | 否 | 1000 |
此配置指定了用于触发采样分片策略的估计分片数的阈值。当分布因子超出了由`chunk-key.even-distribution.factor.upper-bound
`和`chunk-key.even-distribution.factor.lower-bound`，并且估计的分片计数(以近似的行数/块大小计算)超过此阈值，则将使用样本分片策略。这有助于更有效地处理大型数据集。默认值为1000个分片。
|
| inverse-sampling.rate | 整型 | 否 | 1000 |
采样分片策略中使用的采样率的倒数。例如，如果该值设置为1000，则意味着在采样过程中应用了1/1000的采样率。该选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。当处理非常大的数据集时，它特别有用，其中首选较低的采样率。缺省值为1000。
|
| split.allow-sampling | 布尔 | 否 | true |
是否允许使用基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都会回退到非均匀分片方式（迭代查询方式）。默认值为 true。
|
-| exactly_once | 布尔 | 否 | false |
启用exactly once语义
|
+| exactly_once | 布尔 | 否 | false |
在初始快照阶段启用精确一次语义。仅当 `startup.mode` 为 `initial` 时可用。
|
| format | 枚举 | 否 | DEFAULT |
Opengauss CDC可选的输出格式, 有效的枚举是`DEFAULT`, `COMPATIBLE_DEBEZIUM_JSON`.
|
| debezium | 配置 | 否 | - | 将
[Debezium的属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
传递到Debezium嵌入式引擎，该引擎用于捕获来自Opengauss服务的数据更改 |
| common-options | | 否 | - |
源码插件通用参数, 请参考[Source Common
Options](../common-options/source-common-options.md)获取详情
|
@@ -115,10 +116,10 @@ source {
username = "gaussdb"
password = "openGauss@123"
database-names = ["opengauss_cdc"]
- schema-names = ["inventory"]
table-names =
["opengauss_cdc.inventory.opengauss_cdc_table_1","opengauss_cdc.inventory.opengauss_cdc_table_2"]
url = "jdbc:postgresql://opengauss_cdc_e2e:5432/opengauss_cdc"
decoding.plugin.name = "pgoutput"
+ slot.name = "seatunnel_opengauss_cdc"
}
}
@@ -155,11 +156,11 @@ source {
username = "gaussdb"
password = "openGauss@123"
database-names = ["opengauss_cdc"]
- schema-names = ["inventory"]
table-names = ["opengauss_cdc.inventory.full_types_no_primary_key"]
url =
"jdbc:postgresql://opengauss_cdc_e2e:5432/opengauss_cdc?loggerLevel=OFF"
decoding.plugin.name = "pgoutput"
exactly_once = true
+ slot.name = "seatunnel_opengauss_cdc"
table-names-config = [
{
table = "opengauss_cdc.inventory.full_types_no_primary_key"
@@ -170,6 +171,34 @@ source {
}
```
+## CDC 元数据字段
+
+Opengauss CDC 会提供以下元数据字段，可配合 `Metadata` 转换使用：
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| database | STRING | 源数据库名称。 |
+| table | STRING | 源表名称。 |
+| rowKind | STRING | 变更类型，例如 insert、update 或 delete。 |
+| ts_ms | LONG | 源事件时间，单位为毫秒。 |
+| delay | LONG | 事件时间和处理时间之间的延迟，单位为毫秒。 |
+
+示例：
+
+```hocon
+transform {
+ Metadata {
+ metadata_fields {
+ Database = database
+ Table = table
+ RowKind = rowKind
+ EventTime = ts_ms
+ Delay = delay
+ }
+ }
+}
+```
+
## 变更日志
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md
b/docs/zh/connectors/source/PostgreSQL-CDC.md
index 134afd2959..31ebd0bce2 100644
--- a/docs/zh/connectors/source/PostgreSQL-CDC.md
+++ b/docs/zh/connectors/source/PostgreSQL-CDC.md
@@ -20,7 +20,7 @@ import ChangeLog from
'../changelog/connector-cdc-postgres.md';
## 描述
-Postgre CDC 连接器允许从 Postgre 数据库读取快照数据和增量数据。本文件描述了如何设置 Postgre CDC 连接器，以便对
Postgre 数据库执行 SQL 查询。
+PostgreSQL CDC 连接器允许从 PostgreSQL 数据库读取快照数据和增量数据。本文档介绍如何配置 PostgreSQL CDC 连接器。
## 支持的数据源信息
@@ -92,12 +92,13 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| username | String | 是 | - |
连接到数据库服务器时使用的数据库名称。
[...]
| password | String | 是 | - |
连接到数据库服务器时使用的密码。
[...]
| database-names | List | 否 | - |
需要监控的数据库名称。
[...]
-| table-names | List | 是 | - |
需要监控的数据库表名称。表名称需要包含数据库名称，例如：`database_name.table_name`。
[...]
-| table-names-config | List | 否 | - |
表配置列表。例如： [{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]。 snapshotSplitColumn
选项必须配置为唯一键(主键或唯一索引)。 如果指定了非唯一列，该配置将被忽略，SeaTunnel 会在内部自动选择合适的拆分列。
[...]
+| table-names | List | 二选一 | - |
需要监控的表。请使用完整的 `database.schema.table` 格式，例如：`postgres_cdc.inventory.orders`。
[...]
+| table-pattern | String | 二选一 | - |
需要监控的表名正则表达式。正则需要匹配完整表名，例如：`postgres_cdc\\.inventory\\..*`。`table-names` 和
`table-pattern` 互斥。
[...]
+| table-names-config | List | 否 | - |
表级配置列表。例如：`[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]`。无物理主键表可通过 `primaryKeys`
指定唯一键。`snapshotSplitColumn` 必须是唯一键，否则 SeaTunnel 会忽略该配置并自动选择拆分列。
[...]
| startup.mode | List | 否 | INITIAL |
PostgreSQL CDC 消费者的可选启动模式，有效枚举为 `initial`、`earliest` 和 `latest`。<br/>
`initial`: 启动时同步历史数据，然后同步增量数据。<br/> `earliest`: 从可能的最早偏移量启动。<br/> `latest`:
从最新偏移量启动。
[...]
| snapshot.split.size | Integer | 否 | 8096 |
表快照的拆分大小（行数），捕获的表在读取表快照时被拆分成多个拆分。
[...]
| snapshot.fetch.size | Integer | 否 | 1024 |
读取表快照时每次轮询的最大获取大小。
[...]
-| slot.name | String | 否 | - |
为特定数据库/模式创建的用于流式传输更改的 PostgreSQL 逻辑解码槽的名称。服务器使用此槽将事件流式传输到您正在配置的连接器。默认值为
seatunnel。
|
+| slot.name | String | 否 | seatunnel |
PostgreSQL 逻辑解码槽名称。同一个 PostgreSQL 实例上如果有多个 CDC 任务，请为每个任务配置不同的 `slot.name`。
|
| decoding.plugin.name | String | 否 | pgoutput |
安装在服务器上的 Postgres 逻辑解码插件的名称，支持的值有
decoderbufs、wal2json、wal2json_rds、wal2json_streaming、wal2json_rds_streaming 和
pgoutput。
[...]
| server-time-zone | String | 否 | UTC |
数据库服务器中的会话时区。如果未设置，则使用 ZoneId.systemDefault() 来确定服务器时区。
[...]
| connect.timeout.ms | Duration | 否 | 30000 |
连接器在尝试连接到数据库服务器后应等待的最大时间，以防超时。
[...]
@@ -108,7 +109,7 @@ ALTER TABLE your_table_name REPLICA IDENTITY FULL;
| sample-sharding.threshold | Integer | 否 | 1000 |
此配置指定触发采样分片策略的估计分片数量阈值。当分布因子超出由
`chunk-key.even-distribution.factor.upper-bound` 和
`chunk-key.even-distribution.factor.lower-bound` 指定的范围，且估计的分片数量（计算为近似行数 /
块大小）超过此阈值时，将使用采样分片策略。这可以帮助更有效地处理大数据集。默认值为 1000 个分片。
|
| inverse-sampling.rate | Integer | 否 | 1000 |
在采样分片策略中使用的采样率的倒数。例如，如果此值设置为 1000，则意味着在采样过程中应用 1/1000
的采样率。此选项提供了控制采样粒度的灵活性，从而影响最终的分片数量。在处理非常大数据集时，较低的采样率尤为有用。默认值为 1000。
|
| split.allow-sampling | Boolean | 否 | true |
是否允许基于采样的分片策略。当设置为 false 时，无论预估分片数是否超过阈值，系统都将回退到非均匀分片方式（迭代查询方式）。默认值为 true。 |
-| exactly_once | Boolean | 否 | false |
启用精确一次语义。
[...]
+| exactly_once | Boolean | 否 | false |
在初始快照阶段启用精确一次语义。仅当 `startup.mode` 为 `initial` 时可用。
[...]
| format | Enum | 否 | DEFAULT |
PostgreSQL CDC 的可选输出格式，有效枚举为 `DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`。
[...]
| require-replica-identity-full | Boolean | 否 | true |
要求表具有 REPLICA IDENTITY FULL。设置为 false 时，允许表使用其他副本标识设置，但 UPDATE/DELETE
事件可能不包含之前的状态。此选项仅应用于仅追加的表（例如 outbox 模式）。默认为 true 以保持向后兼容性。
[...]
| debezium | Config | 否 | - | 将
[Debezium
的属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/postgresql.adoc#connector-configuration-properties)
传递给用于捕获 PostgreSQL 服务器数据更改的 Debezium 嵌入式引擎。
[...]
@@ -134,13 +135,14 @@ env {
source {
Postgres-CDC {
- plugin_output = "customers_Postgre_cdc"
+ plugin_output = "customers_postgres_cdc"
username = "postgres"
password = "postgres"
database-names = ["postgres_cdc"]
- schema-names = ["inventory"]
- table-names =
["postgres_cdc.inventory.postgres_cdc_table_1,postgres_cdc.inventory.postgres_cdc_table_2"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1",
"postgres_cdc.inventory.postgres_cdc_table_2"]
url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = "seatunnel_postgres_cdc"
}
}
@@ -150,7 +152,7 @@ transform {
sink {
jdbc {
- plugin_input = "customers_Postgre_cdc"
+ plugin_input = "customers_postgres_cdc"
url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
driver = "org.postgresql.Driver"
username = "postgres"
@@ -171,15 +173,15 @@ sink {
```
source {
Postgres-CDC {
- plugin_output = "customers_mysql_cdc"
+ plugin_output = "customers_postgres_cdc"
username = "postgres"
password = "postgres"
database-names = ["postgres_cdc"]
- schema-names = ["inventory"]
table-names = ["postgres_cdc.inventory.full_types_no_primary_key"]
url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
decoding.plugin.name = "decoderbufs"
- exactly_once = false
+ exactly_once = true
+ slot.name = "seatunnel_postgres_cdc"
table-names-config = [
{
table = "postgres_cdc.inventory.full_types_no_primary_key"
@@ -190,6 +192,34 @@ source {
}
```
+## CDC 元数据字段
+
+PostgreSQL CDC 会提供以下元数据字段，可配合 `Metadata` 转换使用：
+
+| 字段 | 类型 | 说明 |
+|------|------|------|
+| database | STRING | 源数据库名称。 |
+| table | STRING | 源表名称。 |
+| rowKind | STRING | 变更类型，例如 insert、update 或 delete。 |
+| ts_ms | LONG | 源事件时间，单位为毫秒。 |
+| delay | LONG | 事件时间和处理时间之间的延迟，单位为毫秒。 |
+
+示例：
+
+```hocon
+transform {
+ Metadata {
+ metadata_fields {
+ Database = database
+ Table = table
+ RowKind = rowKind
+ EventTime = ts_ms
+ Delay = delay
+ }
+ }
+}
+```
+
## 常见问题
### PostgreSQL CDC 需要哪些权限？
