担心 DataX 迁移到 Apache SeaTunnel 成本高？一篇指南手把手带你平滑切换

# 源码编译git clone https://github.com/apache/seatunnel-tools.gitcd seatunnel-toolsmvn clean package -pl x2seatunnel -DskipTests# 编译完成后，包位于 x2seatunnel/target/x2seatunnel-*.zip

# 基本用法：将 datax.json 转换为 seatunnel.conf./bin/x2seatunnel.sh \    -s examples/source/datax-mysql2hdfs.json \    -t examples/target/mysql2hdfs-result.conf \    -r examples/report/mysql2hdfs-report.md

{    "job": {        "setting": {            "speed": {                // [DataX] 全局并发通道数，控制同步速度                "channel": 1            }        },        "content": [            {                "reader": {                    // [DataX] 读取插件名称                    "name": "mysqlreader",                    "parameter": {                        "username": "root",                        "password": "root",                        // [DataX] 需要同步的列名                        "column": ["id", "name", "age"],                        "connection": [{                            // [DataX] 源表名                            "table": ["source_table"],                            // [DataX] JDBC 连接串                            "jdbcUrl": ["jdbc:mysql://localhost:3306/source_db"]                        }]                    }                },                "writer": {                    // [DataX] 写入插件名称                    "name": "mysqlwriter",                    "parameter": {                        // [DataX] 写入模式，支持 insert/replace/update                        "writeMode": "insert",                        "username": "root",                        "password": "root",                        "column": ["id", "name", "age"],                        "connection": [{                            // [DataX] 目标表名                            "table": ["target_table"],                            "jdbcUrl": ["jdbc:mysql://localhost:3306/target_db"]                        }]                    }                }            }        ]    }}

# 1. 环境配置 (对应 DataX 的 setting)env {  # [SeaTunnel] 任务并行度，对应 DataX 的 channel  execution.parallelism = 1  # [SeaTunnel] 任务模式：BATCH (离线批处理) 或 STREAMING (实时流处理)  job.mode = "BATCH"}# 2. Source 配置 (对应 DataX 的 reader)source {  Jdbc {    # [SeaTunnel] 驱动类名    driver = "com.mysql.cj.jdbc.Driver"    # [SeaTunnel] JDBC 连接串    url = "jdbc:mysql://localhost:3306/source_db"    user = "root"    password = "root"    # [SeaTunnel] 查询语句，支持灵活的 SQL 定义，替代 DataX 的 column + table 配置    query = "select id, name, age from source_table"    # [SeaTunnel] 关键配置：将读取到的数据注册为一个临时表，供后续 Sink 使用    result_table_name = "mysql_source"  }}# 3. Transform 配置 (可选，DataX 通常没有这一层)# transform {#   ...# }# 4. Sink 配置 (对应 DataX 的 writer)sink {  Jdbc {    driver = "com.mysql.cj.jdbc.Driver"    url = "jdbc:mysql://localhost:3306/target_db"    user = "root"    password = "root"    # [SeaTunnel] 关键配置：指定数据来源表，这里引用 Source 中定义的 result_table_name    source_table_name = "mysql_source"    # [SeaTunnel] 写入 SQL 模板    query = "insert into target_table (id, name, age) values (?, ?, ?)"  }}

# 方式一：本地开发模式 (Local)
# 适用于开发调试，直接在本地启动进程执行任务
./bin/seatunnel.sh --config ./config/mysql_to_mysql.conf -e local

# 方式二：集群生产模式 (Cluster - Zeta Engine)
# 适用于生产环境，将任务提交到已经启动的 SeaTunnel Zeta 集群
./bin/seatunnel.sh --config ./config/mysql_to_mysql.conf -e cluster

env {  # [CDC 必选] 开启实时流模式  job.mode = "STREAMING"  # [CDC 必选] 开启 Checkpoint (单位毫秒)，用于故障恢复和数据一致性保障  checkpoint.interval = 5000}source {  MySQL-CDC {    result_table_name = "mysql_cdc_source"    # 数据库连接配置    base-url = "jdbc:mysql://localhost:3306/source_db"    username = "root"    password = "root"    # [CDC] 指定需要监听的表，格式：database.table    table-names = ["source_db.source_table"]    # [CDC] 启动模式：    # initial: 先全量同步，再自动切换到增量 Binlog (最常用)    # latest: 只同步任务启动后的增量数据    startup.mode = "initial"  }}sink {  Jdbc {    source_table_name = "mysql_cdc_source"    driver = "com.mysql.cj.jdbc.Driver"    url = "jdbc:mysql://localhost:3306/target_db"    user = "root"    password = "root"    # [CDC 关键] 自动生成 SQL 以支持 INSERT/UPDATE/DELETE    generate_sink_sql = true    # [CDC 关键] 指定目标表的主键，用于确定更新/删除的行    primary_keys = ["id"]    # 目标库表名称    database = "target_db"    table = "target_table"  }}