环境

mysql-connector-java-8.0.28.jar、connector-cdc-mysql

配置 

env {
  # You can set SeaTunnel environment configuration here
  execution.parallelism = 2
  job.mode = "STREAMING"
  # 10秒检查一次，可以适当加大这个值
  checkpoint.interval = 10000
  #execution.checkpoint.interval = 10000
  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
}

# 配置数据源
source {
  MySQL-CDC {
    # 数据库账号
    username = "root"
    password = "123456"
    # 源表，格式：数据库名.表名
    table-names = ["source.test_table"]
    base-url = "jdbc:mysql://10.10.10.11:3306/source"
  }
}

# 配置目标库
sink {

  jdbc {
    url = "jdbc:mysql://10.10.10.11:3306/sink"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "123456"
    
    generate_sink_sql = true
	# 目标数据库名
    database = "sink"
    # 目标表名
    table = "test_table"
    # 主键名称
    primary_keys = ["id"]
	field_ide = LOWERCASE
    schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
  }

}

 测试

 bin/seatunnel.sh --config  config/mysql_to_mysql_config -m local