source {
# 数据源配置
Jdbc {
result_table_name = “source_table”
driver = “com.mysql.cj.jdbc.Driver”
url = “jdbc:mysql://192.168.1.51:3306/fgedu_db”
user = “root”
password = “root123”
query = “SELECT * FROM orders WHERE create_time >= ‘2026-01-01′”
}
}

transform {
# 数据转换配置（可选）
Filter {
source_table_name = “source_table”
result_table_name = “filtered_table”
fields = [“id”, “order_no”, “amount”, “status”]
}
}

sink {
# 数据目标配置
Jdbc {
source_table_name = “filtered_table”
driver = “com.mysql.cj.jdbc.Driver”
url = “jdbc:mysql://192.168.1.52:3306/fgedu_dw”
user = “root”
password = “root123”
table = “orders”
primary_keys = [“id”]
batch_size = 1000
}
}

环境配置说明

# 运行模式
job.mode = “BATCH” # BATCH或STREAMING

# Checkpoint配置
checkpoint.interval = 10000 # Checkpoint间隔（毫秒）
checkpoint.timeout = 60000 # Checkpoint超时时间

# 重试配置
retry.times = 3 # 失败重试次数
retry.interval = 5000 # 重试间隔（毫秒）

# 资源配置
job.resource.cpu = 2 # CPU核心数
job.resource.memory = 4096 # 内存大小（MB）
}

env {
parallelism = 4
job.mode = “BATCH”
checkpoint.interval = 10000
}

source {
Jdbc {
result_table_name = “orders_source”
driver = “com.mysql.cj.jdbc.Driver”
url = “jdbc:mysql://192.168.1.51:3306/fgedu_db?useSSL=false&serverTimezone=Asia/Shanghai”
user = “seatunnel”
password = “seatunnel123”
query = “””
SELECT
id, order_no, user_id, product_id,
quantity, amount, status,
create_time, update_time
FROM orders
WHERE update_time >= DATE_SUB(NOW(), INTERVAL 1 DAY)
“””
partition_column = “id”
partition_num = 4
}
}

transform {
Filter {
source_table_name = “orders_source”
result_table_name = “orders_filtered”
fields = [“id”, “order_no”, “user_id”, “product_id”, “quantity”, “amount”, “status”, “create_time”]
}

Sql {
source_table_name = “orders_filtered”
result_table_name = “orders_transformed”
query = “””
SELECT
id, order_no, user_id, product_id,
quantity, amount,
CASE status
WHEN 0 THEN ‘pending’
WHEN 1 THEN ‘paid’
WHEN 2 THEN ‘shipped’
WHEN 3 THEN ‘completed’
ELSE ‘unknown’
END as status,
create_time
FROM orders_filtered
“””
}
}

sink {
Jdbc {
source_table_name = “orders_transformed”
driver = “com.mysql.cj.jdbc.Driver”
url = “jdbc:mysql://192.168.1.52:3306/fgedu_dw?useSSL=false&serverTimezone=Asia/Shanghai”
user = “seatunnel”
password = “seatunnel123”
table = “orders”
primary_keys = [“id”]
batch_size = 2000
max_retries = 3
is_exactly_once = “true”
xa_data_source_class_name = “com.mysql.cj.jdbc.MysqlXADataSource”
}
}

步骤2：运行同步任务

输出示例如下：
2026-04-04 10:00:00.000 [main] INFO org.apache.seatunnel.core.starter.Seatunnel – Start running SeaTunnel job…
2026-04-04 10:00:00.100 [main] INFO org.apache.seatunnel.engine.server.SeaTunnelServer – Starting SeaTunnel server…
2026-04-04 10:00:01.000 [main] INFO org.apache.seatunnel.engine.server.TaskExecutionService – Task execution service started
2026-04-04 10:00:02.000 [main] INFO org.apache.seatunnel.engine.server.SeaTunnelServer – Job fgedu_mysql_sync_001 submitted
2026-04-04 10:00:05.000 [main] INFO org.apache.seatunnel.core.starter.Seatunnel – Job Status: RUNNING
2026-04-04 10:00:30.000 [main] INFO org.apache.seatunnel.core.starter.Seatunnel – Job Status: FINISHED
2026-04-04 10:00:30.100 [main] INFO org.apache.seatunnel.core.starter.Seatunnel –
***********************************************
Job Statistics
***********************************************
Total Read Count: 100000
Total Write Count: 100000
Total Failed Count: 0
Total Time: 30000ms
***********************************************

# 查看任务日志
$ tail -f logs/seatunnel.log

步骤3：增量同步配置

env {
parallelism = 2
job.mode = “STREAMING”
checkpoint.interval = 5000
}

source {
MySQL-CDC {
result_table_name = “orders_cdc”
base-url = “jdbc:mysql://192.168.1.51:3306/fgedu_db”
username = “seatunnel”
password = “seatunnel123”
database-names = [“fgedu_db”]
table-names = [“fgedu_db.orders”, “fgedu_db.users”]
server-id = 5400
server-time-zone = “Asia/Shanghai”
}
}

sink {
Jdbc {
source_table_name = “orders_cdc”
driver = “com.mysql.cj.jdbc.Driver”
url = “jdbc:mysql://192.168.1.52:3306/fgedu_dw”
user = “seatunnel”
password = “seatunnel123”
table = “${table_name}”
primary_keys = [“id”]
batch_size = 1000
}
}

env {
parallelism = 4
job.mode = “STREAMING”
checkpoint.interval = 5000
}

source {
MySQL-CDC {
result_table_name = “source_data”
base-url = “jdbc:mysql://192.168.1.51:3306/fgedu_db”
username = “seatunnel”
password = “seatunnel123”
database-names = [“fgedu_db”]
table-names = [“fgedu_db.orders”]
server-id = 5401
server-time-zone = “Asia/Shanghai”
}
}

sink {
Kafka {
source_table_name = “source_data”
topic = “fgedu_orders”
bootstrap.servers = “192.168.1.51:9092,192.168.1.52:9092,192.168.1.53:9092”
format = json
semantics = EXACTLY_ONCE
partition = 3
replication.factor = 3
kafka.request.timeout.ms = 60000
kafka.max.block.ms = 120000
}
}

# 运行任务
$ sh bin/seatunnel.sh –config config/mysql_to_kafka.conf

输出示例如下：
2026-04-04 10:00:00.000 [main] INFO org.apache.seatunnel.core.starter.Seatunnel – Start CDC streaming job…
2026-04-04 10:00:01.000 [main] INFO org.apache.seatunnel.connectors.seatunnel.kafka.KafkaSinkWriter – Kafka producer initialized
2026-04-04 10:00:02.000 [main] INFO org.apache.seatunnel.core.starter.Seatunnel – Job Status: RUNNING

步骤2：Kafka到ClickHouse同步

env {
parallelism = 4
job.mode = “STREAMING”
checkpoint.interval = 10000
}

source {
Kafka {
result_table_name = “kafka_source”
topics = [“fgedu_orders”]
bootstrap.servers = “192.168.1.51:9092”
consumer.group = “seatunnel_clickhouse_group”
format = json
schema = “””
{
“type”: “object”,
“properties”: {
“id”: {“type”: “integer”},
“order_no”: {“type”: “string”},
“user_id”: {“type”: “integer”},
“amount”: {“type”: “number”},
“status”: {“type”: “string”},
“create_time”: {“type”: “string”}
}
}
“””
start_mode = earliest
}
}

transform {
Filter {
source_table_name = “kafka_source”
result_table_name = “filtered_data”
fields = [“id”, “order_no”, “user_id”, “amount”, “status”, “create_time”]
}
}

sink {
Clickhouse {
source_table_name = “filtered_data”
host = “192.168.1.51:8123”
database = “fgedu_dw”
table = “orders”
username = “default”
password = “clickhouse123”
bulk_size = 10000
split_mode = true
sharding_key = “id”
}
}

# 运行任务
$ sh bin/seatunnel.sh –config config/kafka_to_clickhouse.conf

seatunnel:
engine:
backup-count: 1
queue-type: blockingqueue
print-execution-info-interval: 60
slot-service:
dynamic-slot: true
checkpoint:
interval: 300000
timeout: 10000
storage:
type: hdfs
max-retained: 3
plugin-config:
namespace: /seatunnel/checkpoint
storage.type: hdfs
fs.defaultFS: hdfs://192.168.1.51:9000
hazelcast:
cluster-name: seatunnel-cluster
network:
join:
tcp-ip:
enabled: true
member-list:
– 192.168.1.51:5801
– 192.168.1.52:5801
– 192.168.1.53:5801
port:
auto-increment: false
port: 5801

# 在所有节点上启动SeaTunnel Server
$ sh bin/seatunnel-cluster.sh -d

# 查看集群状态
$ curl http://192.168.1.51:5801/hazelcast/rest/maps/overview

输出示例如下：
{
“clusterName”: “seatunnel-cluster”,
“members”: [
{“address”: “192.168.1.51:5801”, “status”: “ACTIVE”},
{“address”: “192.168.1.52:5801”, “status”: “ACTIVE”},
{“address”: “192.168.1.53:5801”, “status”: “ACTIVE”}
],
“slotCount”: 12,
“usedSlotCount”: 2
}

步骤2：提交任务到集群

# 查看运行中的任务
$ curl http://192.168.1.51:5801/hazelcast/rest/maps/running-jobs

输出示例如下：
[
{
“jobId”: “fgedu_sync_001”,
“jobName”: “mysql_to_mysql”,
“status”: “RUNNING”,
“createTime”: “2026-04-04T10:00:00”,
“progress”: {
“totalRead”: 50000,
“totalWrite”: 48000
}
}
]

# 停止任务
$ curl -X POST http://192.168.1.51:5801/hazelcast/rest/maps/stop-job/fgedu_sync_001

步骤3：监控集群

seatunnel:
engine:
metrics:
enabled: true
port: 5802
reporter: prometheus

# 访问Prometheus指标
$ curl http://192.168.1.51:5802/metrics

输出示例如下：
# HELP seatunnel_job_count Total job count
# TYPE seatunnel_job_count gauge
seatunnel_job_count{status=”running”} 2.0
seatunnel_job_count{status=”finished”} 15.0
seatunnel_job_count{status=”failed”} 1.0

# HELP seatunnel_slot_count Total slot count
# TYPE seatunnel_slot_count gauge
seatunnel_slot_count{status=”total”} 12.0
seatunnel_slot_count{status=”used”} 4.0

# HELP seatunnel_record_count Total record count
# TYPE seatunnel_record_count counter
seatunnel_record_count{job=”fgedu_sync_001″,type=”read”} 100000.0
seatunnel_record_count{job=”fgedu_sync_001″,type=”write”} 99500.0