davidzollo commented on issue #8388:
URL: https://github.com/apache/seatunnel/issues/8388#issuecomment-2694512142
MySQL cdc to PostgreSQL
--------------------
```
env {
job.mode = "STREAMING"
job.name = "DEMO"
parallelism = 3
checkpoint.interval = 30000 # 30s
checkpoint.timeout = 30000 # 30s
job.retry.times = 3
job.retry.interval.seconds = 3 # 3s
}
source {
MySQL-CDC {
base-url =
"jdbc:mysql://192.168.8.101:3306/test?serverTimezone=Asia/Shanghai"
username = "test"
password = "123456"
database-names = ["test"]
# table-names = ["test.test_001","test.test_002"]
table-pattern = "test\\.test_.*" # The first dot is a normal string and
needs to be escaped; the second dot matches any single character
table-names-config = [
{"table":"test.test_002","primaryKeys":["id"]}
]
startup.mode = "initial" # First perform a full historical sync, then
incremental sync
snapshot.split.size = "8096"
snapshot.fetch.size = "1024"
server-id = "6500-8500"
connect.timeout.ms = 30000
connect.max-retries = 3
connection.pool.size = 20
exactly_once = false # For data analysis scenarios, disable
exactly-once consistency, allowing some duplication and loss to improve
performance
schema-changes.enabled = true # Enable schema evolution to avoid
frequent modifications of table fields, but may affect downstream tasks.
Supports add, rename, and drop operations.
}
}
sink {
jdbc {
url = "jdbc:postgresql://192.168.8.101:5432/test"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
generate_sink_sql = true
database = "test"
table = "${database_name}.${table_name}"
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
data_save_mode = "APPEND_DATA"
# enable_upsert = false
}
}
```
-------------------------------
```
env {
job.mode = "STREAMING"
job.name = "DEMO"
parallelism = 3
checkpoint.interval = 30000 # 30s
checkpoint.timeout = 30000 # 30s
job.retry.times = 3
job.retry.interval.seconds = 3 # 3s
}
source {
MySQL-CDC {
base-url =
"jdbc:mysql://192.168.8.101:3306/test?serverTimezone=Asia/Shanghai"
username = "test"
password = "123456"
database-names = ["test"]
# table-names = ["test.test_001","test.test_002"]
table-pattern = "test\\.test_.*" # 第一个点为普通字符串，需做转义； 第二个点表示匹配1个任意字符
table-names-config = [
{"table":"test.test_002","primaryKeys":["id"]}
]
startup.mode = "initial" # 先全量同步历史，再增量
snapshot.split.size = "8096"
snapshot.fetch.size = "1024"
server-id = "6500-8500"
connect.timeout.ms = 30000
connect.max-retries = 3
connection.pool.size = 20
exactly_once = false # 数据分析场景，关闭精确一致性，允许一定重复和缺失 提升性能
schema-changes.enabled = true # 模式演变 开启 避免反复修改表字段,但可能影响下游任务。add rename
drop
}
}
sink {
jdbc {
url = "jdbc:postgresql://192.168.8.101:5432/test"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
generate_sink_sql = true
database = "test"
table = "${database_name}.${table_name}"
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
data_save_mode = "APPEND_DATA"
# enable_upsert = false
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
