obsidian-sync/work常用/中间件/seatunnel/MySql-CDC.md

### 一、环境准备
#### 1.mysql 需开启binlog及gtid，seatunnel-cdc会伪装成mysql从库实现数据同步
mysql 8 版本参考配置如下：
```text
[mysqld]
# binlog
server-id = 1
log-bin = mysql-bin
expire_logs_days = 7
binlog_format = row
max_binlog_size = 100m 
# enable gtid mode
gtid_mode = on
enforce_gtid_consistency = on
```

执行SQL验证是否开启binlog
```sql
show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
```
![[Pasted image 20250227171647.png]]
#### 2.开启gtid

按顺序执行以下语句
```sql
set global gtid_mode = OFF_PERMISSIVE;  
set global gtid_mode = ON_PERMISSIVE;  
set global enforce_gtid_consistency = ON;  
set global gtid_mode=ON;
```

### 二、source配置
```text
source {
  MySQL-CDC {
    base-url = "jdbc:mysql://172.31.51.244:3306/bjpt_v4_wlmq"
    username = "manager"
    password = "manager2!@#"
  	database-names = ["bjpt_v4_wlmq"]
    table-names = ["bjpt_v4_wlmq.ryhcs_t_bqgl"]
    startup.mode = "initial"
  	schema-changes.enabled = true
  	server-id = "6500-7500"
  	table-names-config = [
      {
        table = "bjpt_v4_wlmq.ryhcs_t_bqgl"
        primaryKeys = ["ID"]
      }
    ]
  }
}
```

### 三、示例
```hocon
env {
  parallelism = 1
  job.mode = "STREAMING"
}
source {
  MySQL-CDC {
    base-url = "jdbc:mysql://10.2.0.7:3306/bjpt_hekou_v3"
    username = "root"
    password = "root@123"
	database-names = ["bjpt_hekou_v3"]
    table-names = ["bjpt_hekou_v3.ms_t_crjry"]
    startup.mode = "initial"
	schema-changes.enabled = true
	server-id = "6500-7500"
	table-names-config = [
      {
        table = "bjpt_hekou_v3.ms_t_crjry"
        primaryKeys = ["WYBS"]
      }
    ]
  }
}
sink {
	Doris {
        fenodes = "172.31.51.142:8030"
        query-port = 9030
        username = admin
        password = "6G_FahdUxAh@K"
        schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
        database = "bjbj"
        table = "cdc_test_ms_t_crjry"
        sink.enable-2pc = "true"
        sink.enable-delete = "true"
        sink.label-prefix = "cdc_test_ms_t_crjry"
        doris.config = {
            format="json"
            read_json_by_line="true"
        }
    }
}
```

```text
env {
  parallelism = 1
  job.mode = "STREAMING"
}
source {
  MySQL-CDC {
    base-url = "jdbc:mysql://172.31.51.244:3306/bjpt_hekou_v3"
    username = "manager"
    password = "manager2!@#"
	database-names = ["bjpt_hekou_v3"]
    table-names = ["bjpt_hekou_v3.ms_t_crjry","bjpt_hekou_v3.ms_t_crjjtgj","bjpt_hekou_v3.ms_t_crjry_jhw"]
    startup.mode = "initial"
	schema-changes.enabled = true
	server-id = "6500-7500"
	table-names-config = [
      {
        table = "bjpt_hekou_v3.ms_t_crjry"
        primaryKeys = ["WYBS"]
      },
	        {
        table = "bjpt_hekou_v3.ms_t_crjjtgj"
        primaryKeys = ["WYBS"]
      },
	        {
        table = "bjpt_hekou_v3.ms_t_crjry_jhw"
        primaryKeys = ["WYBS"]
      }
    ]
  }
}
sink {
	Doris {
        fenodes = "172.31.51.142:8030"
        query-port = 9030
        username = admin
        password = "6G_FahdUxAh@K"
        schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
        database = "bjbj"
        table = "${table_name}_cdc_test"
        sink.enable-2pc = "true"
        sink.enable-delete = "true"
        sink.label-prefix = "cdc_test_ms"
        doris.config = {
            format="json"
            read_json_by_line="true"
        }
    }
}
```