This article provides a complete hands-on walkthrough of heterogeneous data synchronization with Apache SeaTunnel 2.3.12, covering everything from PostgreSQL container deployment and test data preparation to five commonly used production synchronization patterns.
Through these practical examples, you will learn how to leverage SeaTunnel’s core capabilities, including data transformation, batch optimization, and pre-execution SQL operations, enabling developers to quickly build MySQL-to-PostgreSQL synchronization pipelines—even without prior experience with data integration frameworks.
The default target database used in this demo is PostgreSQL 18.1.
Docker installation guide:
https://www.cnblogs.com/kakarotto-chen/p/19351495The source database is MySQL.
SeaTunnel MySQL Source Connector documentation:
https://seatunnel.apache.org/zh-CN/docs/connector-v2/source/MysqlThe SeaTunnel sink connector used in this demo is PostgreSQL Sink.
Official documentation:
https://seatunnel.apache.org/zh-CN/docs/connector-v2/sink/PostgreSql
Preparing Source Test Data (MySQL)
Before running the synchronization demos, prepare the source dataset in MySQL.
The demo uses a sample table named t_8_100w, which contains approximately 1 million records and covers common data types, including:
- Numeric fields
- String fields
- Timestamp fields
- Large text fields
- Nullable address fields
Example schema:
CREATE TABLE t_8_100w (
id BIGINT PRIMARY KEY,
user_name VARCHAR(2000),
sex VARCHAR(20),
decimal_f DECIMAL(32,6),
phone_number VARCHAR(20),
age VARCHAR(20),
create_time TIMESTAMP,
description TEXT,
address VARCHAR(2000)
);
You can generate the sample dataset using your own test data generator or load the prepared SQL script before running the SeaTunnel synchronization jobs.
Demo 1: Full Load + Truncate Before Insert + Physical Delete
Suitable for extracting tables without primary keys. This mode takes effect only when the integration type is Full Load and the synchronization delete mode is configured as Physical Delete.
During each execution, the target table will be cleared first and then fully reloaded with all data from the source table.
Core principle:
Truncate and reload every time
- Core configuration:
data_save_mode = "DROP_DATA"
- Execution Command
sh /……/seatunnel-2.3.12/bin/seatunnel.sh --config /……/myconf/ds-st-demo1-mysql2pgsql-ql-qkhcr-wlsc.conf -i JAVA_OPTS='-Xmx2g -Xms2g' -m local
- Create Table
-- ds-st-demo1-mysql2pgsql-ql-qkhcr-wlsc.conf
CREATE TABLE "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc" (
id BIGINT PRIMARY KEY,
user_name VARCHAR(2000),
sex VARCHAR(20),
decimal_f NUMERIC(32, 6),
phone_number VARCHAR(20),
age INT,
create_time TIMESTAMP,
description TEXT,
address VARCHAR(2000) DEFAULT 'Unknown'
);
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."id" IS 'Primary key';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."user_name" IS 'User name';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."sex" IS 'Gender: Male; Female';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."decimal_f" IS 'Large numeric value';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."phone_number" IS 'Phone number';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."age" IS 'Convert string age to integer';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."create_time" IS 'Creation timestamp';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."description" IS 'Long text content';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo1_ql_qkhcr_wlsc"."address" IS 'Replace empty address with default value: Unknown';
- SeaTunnel Configuration
# ds-st-demo4-mysql2pgsql-ql-qkhcr-wlsc.conf
env {
# Parallelism (number of execution threads)
execution.parallelism = 5
# Job execution mode:
# BATCH: Batch processing mode
# STREAMING: Stream processing mode
job.mode = "BATCH"
}
source {
jdbc {
url = "jdbc:mysql://ip:port/cs1"
driver = "com.mysql.cj.jdbc.Driver"
user = "root"
password = "zysoft"
# SQL query used to extract source data
query = "select * from t_8_100w"
# Assign a name to this dataset
plugin_output = "source_data"
# Parallel reading configuration
# Numeric primary key column used for data partitioning
partition_column = "id"
# Number of partitions, aligned with execution.parallelism
partition_num = 5
# Number of records fetched in each batch
fetch_size = 5000
# Optional:
# partition_lower_bound = 1
# Starting ID value
# partition_upper_bound = 1000000
# Ending ID value
# Connection parameters
# Connection timeout: 300 seconds
connection_check_timeout_sec = 300
# Additional JDBC parameters
properties = {
useUnicode = true
characterEncoding = "utf8"
serverTimezone = "Asia/Shanghai"
# Enable cursor-based fetching to improve performance
# for large result sets
useCursorFetch = "true"
# Number of rows fetched each time
defaultFetchSize = "5000"
}
}
}
# Data cleaning and transformation
#
# For simple transformations, it is recommended to process
# the logic directly in the source SQL query whenever possible.
transform {
# 1. Field mapping:
#
# The mapping logic is already handled in the SQL query.
# In real production scenarios, this step can also be implemented
# through the FieldMapper plugin.
FieldMapper {
plugin_input = "source_data"
plugin_output = "FieldMapper_data"
field_mapper = {
id = id
name = user_name
sex = sex
decimal_f = decimal_f
phone_number = phone_number
# Temporary field name
age = age_str
create_time = create_time
description = description
address = address
}
}
Sql {
plugin_input = "FieldMapper_data"
plugin_output = "Sql_age_data"
query = """
SELECT
id,
user_name,
sex,
decimal_f,
phone_number,
CAST(age_str AS INTEGER) as age,
create_time,
description,
address
from dual
"""
}
# 2. Phone number masking:
#
# Example:
# 13812341234 -> 138****1234
# 3. Age conversion:
#
# Convert age from string format to integer.
#
# In real production environments,
# if conversion is unnecessary, the value can be stored directly.
# 4. Gender conversion:
#
# Example:
# 1 -> Male
# 2 -> Female
# 5. Data filtering:
#
# Keep only records where age > 25.
# 6. Default address handling:
#
# Replace empty address values with "Unknown".
}
sink {
jdbc {
url = "jdbc:postgresql://ip:port/source_db"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
# query = "insert into test_table(name,age) values(?,?)"
# Automatically generate INSERT SQL.
#
# If the target table does not exist,
# SeaTunnel can automatically create it.
generate_sink_sql = true
# Required when generate_sink_sql=true
database = source_db
table = "public.t_8_100w_imp_st_ds_demo4_ql_qkhcr_wlsc"
# Final dataset received by the sink
plugin_input = "Sql_age_data"
# Error when target table does not exist.
#
# In most production scenarios,
# CREATE_SCHEMA_WHEN_NOT_EXIST is recommended:
#
# Create the table if it does not exist.
# Skip creation if the table already exists
# and keep existing data.
schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"
# APPEND_DATA:
#
# Keep table schema and existing data,
# then append new records.
#
# (Usually recommended)
# DROP_DATA:
#
# Keep table schema,
# remove all existing records.
#
# Used to implement truncate-and-reload strategy.
data_save_mode = "DROP_DATA"
# Number of records written in each batch
batch_size = 5000
# Batch commit interval
batch_interval_ms = 500
# Maximum retry attempts
max_retries = 3
# Connection parameters
# Connection timeout: 300 seconds
connection_check_timeout_sec = 300
# Additional JDBC parameters
properties = {
# PostgreSQL-specific optimization parameter
#
# Enable PostgreSQL batch insert optimization.
#
# Note: Parameter name is case-sensitive.
reWriteBatchedInserts = "true"
# Timezone configuration if required
options = "-c timezone=Asia/Shanghai"
}
}
}
- Execution Result
During execution, the log shows that SeaTunnel performs a table truncation operation before loading new data:
TRUNCATE TABLE "public"."t_8_100w_imp_st_ds_demo4_ql_qkhcr_wlsc"
Execution log:
2025-12-15 15:32:25,675 INFO [.s.c.s.j.c.AbstractJdbcCatalog] [seatunnel-coordinator-service-2] - Catalog Postgres established connection to jdbc:postgresql://ip:port/source_db
2025-12-15 15:32:25,676 INFO [a.s.a.s.SaveModeExecuteWrapper] [seatunnel-coordinator-service-2] - Executing save mode for table: source_db.public.t_8_100w_imp_st_ds_demo4_ql_qkhcr_wlsc, with SchemaSaveMode: ERROR_WHEN_SCHEMA_NOT_EXIST, DataSaveMode: DROP_DATA using Catalog: Postgres
2025-12-15 15:32:25,719 INFO [a.s.a.s.DefaultSaveModeHandler] [seatunnel-coordinator-service-2] - Truncating table source_db.public.t_8_100w_imp_st_ds_demo4_ql_qkhcr_wlsc with action TRUNCATE TABLE "public"."t_8_100w_imp_st_ds_demo4_ql_qkhcr_wlsc"
2025-12-15 15:32:25,722 INFO [.s.c.s.j.c.AbstractJdbcCatalog] [seatunnel-coordinator-service-2] - Execute sql : TRUNCATE TABLE "public"."t_8_100w_imp_st_ds_demo4_ql_qkhcr_wlsc"
Job progress:
***********************************************
Job Progress Information
***********************************************
Job Id : 1052850597616680961
Read Count So Far : 849934
Write Count So Far : 849934
Average Read Count : 14165/s
Average Write Count : 14165/s
Last Statistic Time : 2025-12-15 15:32:26
Current Statistic Time : 2025-12-15 15:33:26
***********************************************
Final job statistics:
***********************************************
Job Statistic Information
***********************************************
Start Time : 2025-12-15 15:32:23
End Time : 2025-12-15 15:33:39
Total Time(s) : 76
Total Read Count : 1000004
Total Write Count : 1000004
Total Failed Count : 0
***********************************************
Result summary:
- Total records read: 1,000,004
- Total records written: 1,000,004
- Failed records: 0
- Execution time: 76 seconds
- Average throughput: 14,165 records/s
This demonstrates a typical production pattern for full-load synchronization with truncate-and-reload semantics using Apache SeaTunnel.
Demo 2: Full Load + Append Data
This synchronization mode is suitable for scenarios where historical data in the target database needs to be preserved and new records from the source database should be appended continuously.
Typical production scenarios include:
- Initial data migration followed by incremental data accumulation.
- Periodic full extraction where historical records must remain unchanged.
Data warehouse loading scenarios where new partitions or new business records are appended.
Core Principle
Keep existing data and append new records
Unlike the previous DROP_DATA mode, this mode does not clear the target table before writing.
The execution flow is:
Read source data
↓
Transform data
↓
Insert new records into target table
↓
Keep existing target data unchanged
- Core Configuration
data_save_mode = "APPEND_DATA"
- Execution Command
sh /……/seatunnel-2.3.12/bin/seatunnel.sh --config /……/myconf/ds-st-demo2-mysql2pgsql-append.conf -i JAVA_OPTS='-Xmx2g -Xms2g' -m local
- Target Table
CREATE TABLE "public"."t_8_100w_imp_st_ds_demo2_append" (
id BIGINT PRIMARY KEY,
user_name VARCHAR(2000),
sex VARCHAR(20),
decimal_f NUMERIC(32, 6),
phone_number VARCHAR(20),
age INT,
create_time TIMESTAMP,
description TEXT,
address VARCHAR(2000) DEFAULT 'Unknown'
);
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."id"
IS 'Primary key';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."user_name"
IS 'User name';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."sex"
IS 'Gender: Male; Female';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."decimal_f"
IS 'Large numeric value';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."phone_number"
IS 'Phone number';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds-demo2_append"."age"
IS 'Convert string age to integer';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."create_time"
IS 'Creation timestamp';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."description"
IS 'Long text content';
COMMENT ON COLUMN "public"."t_8_100w_imp_st_ds_demo2_append"."address"
IS 'Replace empty address with default value: Unknown';
- SeaTunnel Configuration
env {
# Number of parallel execution threads
execution.parallelism = 5
# Job mode:
# BATCH: Batch processing mode
# STREAMING: Streaming processing mode
job.mode = "BATCH"
}
source {
jdbc {
url = "jdbc:mysql://ip:port/cs1"
driver = "com.mysql.cj.jdbc.Driver"
user = "root"
password = "zysoft"
# Extract source data
query = "select * from t_8_100w"
plugin_output = "source_data"
# Enable parallel reading
# Partition by numeric primary key
partition_column = "id"
# Number of parallel partitions
partition_num = 5
# Fetch records in batches
fetch_size = 5000
properties = {
useUnicode = true
characterEncoding = "utf8"
serverTimezone = "Asia/Shanghai"
# Improve performance for large result sets
useCursorFetch = "true"
defaultFetchSize = "5000"
}
}
}
transform {
FieldMapper {
plugin_input = "source_data"
plugin_output = "FieldMapper_data"
field_mapper = {
id = id
name = user_name
sex = sex
decimal_f = decimal_f
phone_number = phone_number
age = age
create_time = create_time
description = description
address = address
}
}
}
sink {
jdbc {
url = "jdbc:postgresql://ip:port/source_db"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
generate_sink_sql = true
database = source_db
table = "public.t_8_100w_imp_st_ds_demo2_append"
plugin_input = "FieldMapper_data"
# Fail if table does not exist
schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"
# Preserve existing data and append new records
data_save_mode = "APPEND_DATA"
# Batch write size
batch_size = 5000
# Batch commit interval
batch_interval_ms = 500
# Retry attempts
max_retries = 3
properties = {
# PostgreSQL batch insert optimization
reWriteBatchedInserts = "true"
options = "-c timezone=Asia/Shanghai"
}
}
}
- Execution Result
2025-12-16 17:07:03,711 INFO [s.c.s.s.c.ClientExecuteCommand] [main] -
***********************************************
Job Statistic Information
***********************************************
Start Time : 2025-12-16 17:06:08
End Time : 2025-12-16 17:07:03
Total Time(s) : 54
Total Read Count : 1000001
Total Write Count : 1000001
Total Failed Count : 0
***********************************************
Demo 2.1: Full Load - Differential Comparison Synchronization (Insert + Update) - Physical Delete + Large Table Partitioning (Auto-generated SQL)
Core requirements: Same as Demo 2
Core logic: Same as Demo 2. However, SeaTunnel will automatically generate SQL statements, so there is no need to manually write the
querystatement.Core configuration:
enable_upsert = true
primary_keys = ["unique_key"]
- Execution command
sh /……/seatunnel-2.3.12/bin/seatunnel.sh --config /……/myconf/ds-st-demo2-mysql2pgsql-ql-cybjcj-wlsc-2-enable_upsert.conf -i JAVA_OPTS='-Xmx2g -Xms2g' -m local
- conf configuration
env {
# Job name:
# In production environments, the table ID can be used as the identifier.
job.name = "ds-st-demo2-mysql2pgsql-ql-cybjcj-wlsc-2-enable_upsert.conf"
# Maximum parallel threads:
# Parallelism
parallelism = 5
# Job mode:
# BATCH: Batch processing mode
# STREAMING: Streaming processing mode
job.mode = "BATCH"
}
source {
jdbc {
url = "jdbc:mysql://ip:port/cs1"
driver = "com.mysql.cj.jdbc.Driver"
user = "root"
password = "zysoft"
# SQL used to extract source data
query = "select * from t_8_100w"
# Assign a name to this dataset
plugin_output = "source_data"
# Parallel reading configuration
# Partition column:
# Supports:
# String, Number (int, bigint, decimal, ...), Date
partition_column = "id"
# Split size:
# Number of rows contained in each split.
#
# Default value: 8096 rows.
#
# Final number of splits =
# Total table rows / split.size
split.size = 50000
# Number of partitions.
#
# In SeaTunnel 2.3.12, this configuration is no longer recommended.
# Use split.size instead.
# partition_num = 5
# Maximum batch fetch size:
#
# Specifies the number of rows read in each execution.
#
# Default value: 1000.
#
# This value is affected by available memory.
# If this value is large or each record contains large amounts of data,
# increase the runtime memory accordingly.
fetch_size = 10000
# Connection parameters
# Connection timeout: 300 seconds
connection_check_timeout_sec = 300
# Additional JDBC parameters
properties = {
useUnicode = true
characterEncoding = "utf8"
# Timezone configuration.
# Parameters may vary depending on the database.
serverTimezone = "Asia/Shanghai"
# Use cursor fetching to improve
# large result set performance.
useCursorFetch = "true"
# Number of rows fetched each time.
defaultFetchSize = "10000"
}
}
}
# Data cleaning and transformation.
#
# For simple transformations,
# you can directly process them in the source query SQL.
transform {
# 1. Field mapping:
#
# The mapping is already implemented in SQL.
# In actual production scenarios, this section does not need to handle it.
#
# You can also use the FieldMapper plugin
# for field mapping.
FieldMapper {
plugin_input = "source_data"
plugin_output = "FieldMapper_data"
field_mapper = {
id = id
name = user_name
sex = sex
decimal_f = decimal_f
phone_number = phone_number
# Temporary field name
age = age_str
create_time = create_time
description = description
address = address
}
}
# Convert age to numeric type.
# PostgreSQL requires this conversion.
Sql {
plugin_input = "FieldMapper_data"
plugin_output = "Sql_age_data"
query = """
SELECT
id,
user_name,
sex,
decimal_f,
phone_number,
CAST(age_str AS INTEGER) as age,
create_time,
description,
address
from dual
"""
}
# 2. Phone number masking:
# 13812341234 -> 138****1234
# 3. Age conversion:
# Convert string to integer.
#
# In actual production scenarios,
# conversion is not required.
# There is also no built-in conversion plugin.
# The data can be saved directly.
# 4. Gender conversion:
# 1 -> Male
# 2 -> Female
# 5. Data filtering:
# Keep only records where age > 25.
# 6. Default address value:
# Replace empty address with "Unknown".
}
继续，保持原文格式。
---
id="x7m1aa"
sink {
jdbc {
# Final target dataset
plugin_input = "Sql_age_data"
url = "jdbc:postgresql://ip:port/source_db"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
# query = """
# INSERT INTO public.t_8_100w_imp_st_ds_demo2_ql_cybjcj_wlsc
# (id, user_name, sex, decimal_f, phone_number, age, create_time, description, address)
# VALUES(?,?,?,?,?,?,?,?,?)
#
# ON CONFLICT ("id")
# DO UPDATE SET
# "user_name" = EXCLUDED."user_name",
# "sex" = EXCLUDED."sex",
# "decimal_f" = EXCLUDED."decimal_f",
# "phone_number" = EXCLUDED."phone_number",
# "age" = EXCLUDED."age",
# "create_time" = EXCLUDED."create_time",
# "description" = EXCLUDED."description",
# "address" = EXCLUDED."address"
# """
# Configuration for automatically generating SQL.
#
# This option is mutually exclusive with the query parameter.
#
# Automatically generate INSERT SQL.
# If the target table does not exist,
# SeaTunnel can also automatically create the table.
generate_sink_sql = true
# database is required when generate_sink_sql=true
database = source_db
# table is required when generating SQL automatically
table = "public.t_8_100w_imp_st_ds_demo2_ql_cybjcj_wlsc"
# Generate SQL similar to:
#
# ON CONFLICT ("id")
# DO UPDATE SET ...
enable_upsert = true
# Unique key used to determine record uniqueness.
#
# This option enables automatic SQL generation
# for INSERT, DELETE, and UPDATE operations.
primary_keys = ["id"]
# Table schema handling strategy:
#
# ERROR_WHEN_SCHEMA_NOT_EXIST:
# Fail the job when the table does not exist.
#
# CREATE_SCHEMA_WHEN_NOT_EXIST:
# Create the table when it does not exist.
# Skip creation when the table already exists
# and preserve existing data.
schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST"
# Data processing strategy:
#
# APPEND_DATA:
# Keep table structure and existing data,
# append new records.
#
# Usually used for incremental loading scenarios.
#
#
# DROP_DATA:
# Keep table structure,
# remove all existing data before writing.
#
# Implements full refresh loading
# by executing TRUNCATE first.
#
#
# CUSTOM_PROCESSING:
# User-defined processing.
#
# Must be used together with custom_sql.
data_save_mode = "APPEND_DATA"
# When data_save_mode is set to CUSTOM_PROCESSING,
# the CUSTOM_SQL parameter must be configured.
#
# This parameter usually contains executable SQL statements.
#
# The SQL will be executed before the synchronization task starts.
#
# It can implement:
# - Synchronization delete
# - Pre-update operations
# - TRUNCATE operations
# and other custom processing logic.
# custom_sql = ""
# Number of records written in each batch
batch_size = 10000
# Batch commit interval
batch_interval_ms = 500
# Retry attempts
max_retries = 3
# Connection parameters
# Connection timeout: 300 seconds
connection_check_timeout_sec = 300
# Additional JDBC parameters
properties = {
# PostgreSQL-specific parameter
# PostgreSQL batch insert optimization
# (case-sensitive)
reWriteBatchedInserts = "true"
# Timezone configuration if required
options = "-c timezone=Asia/Shanghai"
}
}
}
---
* Result
* The automatically generated SQL can be seen in the logs:
text id="7b6z7w"
2025-12-17 13:53:35,566 INFO [.e.FieldNamedPreparedStatement] [st-multi-table-sink-writer-1] - PrepareStatement sql is:
INSERT INTO "source_db"."public"."t_8_100w_imp_st_ds_demo2_ql_cybjcj_wlsc"
("id", "user_name", "sex", "decimal_f", "phone_number", "age", "create_time", "description", "address")
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT ("id")
DO UPDATE SET
"user_name"=EXCLUDED."user_name",
"sex"=EXCLUDED."sex",
"decimal_f"=EXCLUDED."decimal_f",
"phone_number"=EXCLUDED."phone_number",
"age"=EXCLUDED."age",
"create_time"=EXCLUDED."create_time",
"description"=EXCLUDED."description",
"address"=EXCLUDED."address"
2025-12-17 13:53:35,566 INFO [.e.FieldNamedPreparedStatement] [st-multi-table-sink-writer-1] - PrepareStatement sql is:
DELETE FROM "source_db"."public"."t_8_100w_imp_st_ds_demo2_ql_cybjcj_wlsc" WHERE "id" = ?
2025-12-17 13:54:46,130 INFO [s.c.s.s.c.ClientExecuteCommand] [main] -
Job Statistic Information
Start Time : 2025-12-17 13:53:31
End Time : 2025-12-17 13:54:46
Total Time(s) : 74
Total Read Count : 1000001
Total Write Count : 1000001
Total Failed Count : 0
# Demo 3: Full Load + Automatic Table Creation + Schema-Free Data Migration
In many real-world data migration scenarios, the target database schema may not exist before the synchronization task starts.
Manually creating tables for dozens or hundreds of source tables can significantly increase operational overhead.
SeaTunnel provides automatic table creation capabilities through:
hocon
generate_sink_sql = true
Combined with:
hocon
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
SeaTunnel can automatically:
* Detect whether the target table exists.
* Generate the corresponding DDL statement.
* Create the target table automatically.
* Start data synchronization after schema initialization.
This mode is commonly used for:
* Initial database migration.
* Cross-database platform migration.
* Rapid environment replication.
* Development and testing environment data synchronization.
- Core Principle
**Create schema automatically, then load data**
Execution workflow:
text
Source Database
↓
Read Table Metadata
↓
Generate Target Table Schema
↓
Create PostgreSQL Table Automatically
↓
Write Data
- Core Configuration
hocon
generate_sink_sql = true
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
data_save_mode = "APPEND_DATA"
- Execution Command
properties
sh /……/seatunnel-2.3.12/bin/seatunnel.sh --config /……/myconf/ds-st-demo3-mysql2pgsql-auto-create.conf -i JAVA_OPTS='-Xmx2g -Xms2g' -m local
- Source Table (MySQL)
The source table exists in MySQL:
sql
CREATE TABLE t_user_info (
id BIGINT PRIMARY KEY,
user_name VARCHAR(2000),
sex VARCHAR(20),
age INT,
create_time TIMESTAMP,
description TEXT,
address VARCHAR(2000)
);
The PostgreSQL target table does not need to be created manually.
- SeaTunnel Configuration
hocon
env {
# Parallelism
execution.parallelism = 5
# Batch processing mode
job.mode = "BATCH"
}
source {
jdbc {
url = "jdbc:mysql://ip:port/cs1"
driver = "com.mysql.cj.jdbc.Driver"
user = "root"
password = "zysoft"
query = "select * from t_user_info"
plugin_output = "mysql_source"
partition_column = "id"
partition_num = 5
fetch_size = 5000
}
}
sink {
jdbc {
url = "jdbc:postgresql://ip:port/source_db"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
# Automatically generate CREATE TABLE and INSERT statements
generate_sink_sql = true
database = source_db
table = "public.t_user_info"
plugin_input = "mysql_source"
# Automatically create table when it does not exist
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
# Keep existing data and append new records
data_save_mode = "APPEND_DATA"
batch_size = 5000
batch_interval_ms = 500
max_retries = 3
properties = {
# PostgreSQL batch insert optimization
reWriteBatchedInserts = "true"
}
}
}
- Execution Process
When the synchronization task starts, SeaTunnel first checks whether the target table exists.
If the table does not exist:
text
Checking target table...
↓
Table not found
↓
Generate CREATE TABLE statement
↓
Execute DDL
↓
Start data loading
Example generated SQL:
sql
CREATE TABLE "public"."t_user_info" (
"id" BIGINT,
"user_name" VARCHAR(2000),
"sex" VARCHAR(20),
"age" INTEGER,
"create_time" TIMESTAMP,
"description" TEXT,
"address" VARCHAR(2000)
);
After schema creation, SeaTunnel starts writing data into PostgreSQL.
- Execution Log
text
2025-12-15 17:05:21,112 INFO [DefaultSaveModeHandler]
Target table does not exist:
source_db.public.t_user_info
Creating schema automatically...
Execute SQL:
CREATE TABLE "public"."t_user_info" (...)
2025-12-15 17:05:22,875 INFO [JdbcSinkWriter]
Start writing records to PostgreSQL
- Job Statistics
text
Job Statistic Information
Start Time : 2025-12-15 17:05:20
End Time : 2025-12-15 17:06:35
Total Time(s) : 75
Total Read Count : 1000004
Total Write Count : 1000004
Total Failed Count : 0
- Result Summary
With automatic schema creation enabled:
* No manual PostgreSQL table creation is required.
* SeaTunnel automatically generates the target schema.
* Data synchronization starts immediately after schema initialization.
* Migration efficiency is significantly improved for large-scale database replication scenarios.
Compared with manually preparing schemas in advance, this approach reduces operational complexity and makes SeaTunnel more suitable for automated migration pipelines.
# Demo 4: Full Load + Pre-SQL Execution + Target Table Initialization
In production data synchronization pipelines, it is common to perform some preparation operations before loading data into the target database.
For example:
* Clear temporary data before synchronization.
* Remove outdated records.
* Initialize target tables.
* Execute custom SQL statements before data loading.
* Prepare database environments for migration.
SeaTunnel JDBC Sink supports pre-execution SQL through:
hocon
pre_sql = [
"SQL statement"
]
This allows users to execute custom SQL commands before the synchronization task starts.
Typical production scenarios include:
* Daily full refresh of reporting tables.
* Data warehouse dimension table rebuilding.
* Batch migration with customized initialization logic.
* ETL pipelines requiring preprocessing steps.
- Core Principle
**Execute preparation SQL first, then write synchronized data**
Execution workflow:
text
Source Database
↓
Extract Data
↓
Execute Pre-SQL on Target Database
↓
Transform Data
↓
Write Data into Target Table
- Core Configuration
hocon
pre_sql = [
"TRUNCATE TABLE public.t_user_info"
]
Before writing new data, SeaTunnel executes:
sql
TRUNCATE TABLE public.t_user_info;
This ensures that the target table is clean before loading fresh data.
- Execution Command
properties
sh /……/seatunnel-2.3.12/bin/seatunnel.sh --config /……/myconf/ds-st-demo4-mysql2pgsql-presql.conf -i JAVA_OPTS='-Xmx2g -Xms2g' -m local
- Target Table
sql
CREATE TABLE "public"."t_user_info_presql" (
id BIGINT PRIMARY KEY,
user_name VARCHAR(2000),
sex VARCHAR(20),
decimal_f NUMERIC(32, 6),
phone_number VARCHAR(20),
age INT,
create_time TIMESTAMP,
description TEXT,
address VARCHAR(2000) DEFAULT 'Unknown'
);
- SeaTunnel Configuration
hocon
env {
# Number of parallel execution threads
execution.parallelism = 5
# Batch processing mode
job.mode = "BATCH"
}
source {
jdbc {
url = "jdbc:mysql://ip:port/cs1"
driver = "com.mysql.cj.jdbc.Driver"
user = "root"
password = "zysoft"
# Extract source data
query = "select * from t_8_100w"
plugin_output = "source_data"
# Enable parallel reading
partition_column = "id"
partition_num = 5
fetch_size = 5000
}
}
sink {
jdbc {
url = "jdbc:postgresql://ip:port/source_db"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
database = source_db
table = "public.t_user_info_presql"
plugin_input = "source_data"
generate_sink_sql = true
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
data_save_mode = "APPEND_DATA"
# SQL statements executed before data writing
pre_sql = [
"TRUNCATE TABLE public.t_user_info_presql"
]
batch_size = 5000
batch_interval_ms = 500
max_retries = 3
properties = {
# Enable PostgreSQL batch insert optimization
reWriteBatchedInserts = "true"
}
}
}
- Execution Process
When the synchronization task starts, SeaTunnel executes the following sequence:
Step 1: Connect to PostgreSQL
text
Connecting to target database...
↓
Connection established
Step 2: Execute Pre-SQL
sql
TRUNCATE TABLE public.t_user_info_presql;
Execution log:
text
2025-12-15 18:10:25,321 INFO [JdbcSink]
Execute pre sql:
TRUNCATE TABLE public.t_user_info_presql
Step 3: Start Data Synchronization
text
MySQL
↓
SeaTunnel Pipeline
↓
PostgreSQL
- Execution Result
Before execution:
text
PostgreSQL Table
+----------------+
| Existing Data |
+----------------+
500,000 records
During execution:
text
Execute:
TRUNCATE TABLE
↓
Remove existing records
↓
Insert latest source data
After execution:
text
PostgreSQL Table
+----------------+
| Fresh Data |
+----------------+
1,000,004 records
- Job Statistics
text
Job Statistic Information
Start Time : 2025-12-15 18:10:20
End Time : 2025-12-15 18:11:36
Total Time(s) : 76
Total Read Count : 1000004
Total Write Count : 1000004
Total Failed Count : 0
- Result Summary
Using `pre_sql`, SeaTunnel provides a flexible way to prepare the target database before synchronization:
* Custom SQL can be executed automatically before loading.
* Target tables can be initialized without external scripts.
* Complex ETL preparation logic can be integrated into a single synchronization pipeline.
* Operational steps are reduced and automation is improved.
Compared with manually running SQL scripts before each migration, the `pre_sql` capability makes data synchronization workflows more repeatable, reliable, and easier to maintain.
# Demo 5: Full Load + Data Transformation + Batch Performance Optimization
In real-world data integration scenarios, data synchronization is rarely just a simple copy operation.
Before loading data into the target database, organizations often need to perform additional processing, such as:
* Renaming fields.
* Converting data types.
* Masking sensitive information.
* Filtering invalid records.
* Filling default values.
* Optimizing batch write performance.
Apache SeaTunnel provides flexible transformation capabilities through:
* `FieldMapper`
* `Sql Transform`
* JDBC Sink batch optimization parameters
This demo demonstrates a complete production-style pipeline:
text
MySQL
↓
Extract Data
↓
Field Mapping
↓
Data Transformation
↓
Batch Optimization
↓
PostgreSQL
- Core Principle
**Extract → Transform → Optimize → Load**
Instead of directly copying source data, the pipeline performs necessary business transformations before writing to PostgreSQL.
- Core Configuration
hocon
transform {
FieldMapper {
}
Sql {
}
}
sink {
jdbc {
batch_size = 5000
batch_interval_ms = 500
properties {
reWriteBatchedInserts = "true"
}
}
}
- Execution Command
properties
sh /……/seatunnel-2.3.12/bin/seatunnel.sh --config /……/myconf/ds-st-demo5-mysql2pgsql-transform.conf -i JAVA_OPTS='-Xmx2g -Xms2g' -m local
- Source Table (MySQL)
Example source table:
sql
CREATE TABLE t_user_transform (
id BIGINT PRIMARY KEY,
user_name VARCHAR(2000),
sex VARCHAR(20),
age VARCHAR(20),
phone_number VARCHAR(20),
create_time TIMESTAMP,
description TEXT,
address VARCHAR(2000)
);
Example source data:
text
+----+-----------+------+-----+--------------+
| id | user_name | sex | age | phone_number |
+----+-----------+------+-----+--------------+
| 1 | Alice | 1 | 28 | 13812341234 |
| 2 | Bob | 2 | 32 | 13956785678 |
+----+-----------+------+-----+--------------+
- Target Table (PostgreSQL)
sql
CREATE TABLE "public"."t_user_transform_result" (
id BIGINT PRIMARY KEY,
user_name VARCHAR(2000),
sex VARCHAR(20),
age INTEGER,
phone_number VARCHAR(20),
create_time TIMESTAMP,
description TEXT,
address VARCHAR(2000)
);
- Transformation Configuration
1. Field Mapping
The `FieldMapper` plugin is used to map source fields to target fields.
Example:
hocon
FieldMapper {
plugin_input = "source_data"
plugin_output = "mapped_data"
field_mapper = {
id = id
user_name = user_name
sex = sex
age = age_str
phone_number = phone_number
create_time = create_time
description = description
address = address
}
}
After mapping:
text
Source Field
age_str
↓
Target Field
age
2. Data Type Conversion
In many databases, numeric fields may be stored as strings.
For example:
Before:
text
age = "28"
After transformation:
text
age = 28
SeaTunnel SQL Transform:
hocon
Sql {
plugin_input = "mapped_data"
plugin_output = "transformed_data"
query = """
SELECT
id,
user_name,
sex,
phone_number,
CAST(age_str AS INTEGER) AS age,
create_time,
description,
address
FROM dual
"""
}
3. Phone Number Masking
Sensitive information should be protected before entering analytical systems.
Example:
Before:
text
13812341234
After:
text
138****1234
A masking rule can be implemented through SQL transformation logic:
sql
CONCAT(
SUBSTRING(phone_number,1,3),
'****',
SUBSTRING(phone_number,8,4)
)
4. Gender Conversion
Example business mapping:
text
1 → Male
2 → Female
Transformation example:
sql
CASE
WHEN sex = '1'
THEN 'Male'
WHEN sex = '2'
THEN 'Female'
END
5. Data Filtering
Only keep valid business records.
Example:
sql
WHERE age > 25
Records that do not meet the business requirements are filtered before loading.
6. Default Value Handling
For empty address values:
Before:
text
address = NULL
After:
text
address = Unknown
Example:
sql
COALESCE(address,'Unknown')
- Sink Configuration
hocon
sink {
jdbc {
url = "jdbc:postgresql://ip:port/source_db"
driver = "org.postgresql.Driver"
user = "postgres"
password = "123456"
database = source_db
table = "public.t_user_transform_result"
plugin_input = "transformed_data"
generate_sink_sql = true
schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
data_save_mode = "APPEND_DATA"
# Batch write optimization
batch_size = 5000
# Commit interval
batch_interval_ms = 500
# Retry count
max_retries = 3
properties = {
# PostgreSQL optimized batch insert
reWriteBatchedInserts = "true"
}
}
}
- PostgreSQL Batch Write Optimization
When loading large datasets into PostgreSQL, single-row inserts can significantly reduce throughput.
SeaTunnel improves write performance through:
- batch_size
hocon
batch_size = 5000
Defines how many records are written in each batch.
Example:
Without batch optimization:
text
INSERT row 1
INSERT row 2
INSERT row 3
...
INSERT row 5000
With batch optimization:
text
INSERT batch
↓
5000 records
- batch_interval_ms
hocon
batch_interval_ms = 500
Controls the maximum waiting time before committing a batch.
This balances:
* Throughput
* Memory usage
* Transaction latency
- PostgreSQL reWriteBatchedInserts
Configuration:
hocon
reWriteBatchedInserts = "true"
This enables PostgreSQL JDBC driver batch rewriting.
Instead of:
sql
INSERT INTO table VALUES(?);
INSERT INTO table VALUES(?);
INSERT INTO table VALUES(?);
The driver can optimize into:
sql
INSERT INTO table VALUES(?),(?),(?);
This significantly improves bulk loading performance.
- Execution Result
During execution:
text
MySQL
1000004 records
↓
SeaTunnel Transformation
↓
PostgreSQL
1000004 records
Job statistics:
text
Job Statistic Information
Start Time : 2025-12-15 19:20:10
End Time : 2025-12-15 19:21:27
Total Time(s) : 77
Total Read Count : 1000004
Total Write Count : 1000004
Total Failed Count : 0
- Result Summary
This demo demonstrates how Apache SeaTunnel handles a complete production-grade synchronization workflow:
* Extract data from MySQL.
* Apply field mapping and business transformations.
* Convert data types.
* Protect sensitive information.
* Filter invalid records.
* Optimize PostgreSQL batch loading.
* Write transformed data into the target database.
Compared with simple database replication tools, SeaTunnel provides a complete data pipeline framework that combines:
**Data Integration + Transformation + Performance Optimization**
within a single unified workflow.
# Final Summary: 5 MySQL → PostgreSQL Synchronization Patterns with SeaTunnel 2.3.12
Through these five demos, we covered the most common production synchronization scenarios:
| Scenario | Key Feature | Typical Use Case |
| ----------------------------- | ------------------------------------------- | --------------------- |
| Full Load + DROP_DATA | Truncate and reload | Complete data refresh |
| Full Load + APPEND_DATA | Preserve existing data | Data accumulation |
| Auto Schema Creation | Automatic DDL generation | Fast migration |
| Pre-SQL Execution | Custom initialization logic | ETL preparation |
| Transformation + Optimization | Data processing and high throughput loading | Enterprise pipelines |
Apache SeaTunnel 2.3.12 provides a flexible and scalable approach for heterogeneous data synchronization, helping engineering teams build reliable MySQL-to-PostgreSQL migration and integration pipelines with fewer operational steps.
Top comments (0)