[mysqld]
server-id = 1
log-bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
gtid_mode = ON
enforce_gtid_consistency = ON

# 重启MySQL
# systemctl restart mysqld

# 验证binlog配置
$ mysql -u root -p -e “SHOW VARIABLES LIKE ‘binlog_format’;”

# 输出示例：
+—————+——-+
| Variable_name | Value |
+—————+——-+
| binlog_format | ROW |
+—————+——-+

# 创建CDC用户
$ mysql -u root -p

mysql> CREATE USER ‘fgedu’@’%’ IDENTIFIED BY ‘fgedu_mysql_2024’;
mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘fgedu’@’%’;
mysql> FLUSH PRIVILEGES;

# 创建测试表
mysql> USE fgedudb;
mysql> CREATE TABLE fgedu_employees (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100),
department VARCHAR(50),
salary DECIMAL(10,2),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
mysql> INSERT INTO fgedu_employees (name, department, salary) VALUES
(‘张三’, ‘技术部’, 15000.00),
(‘李四’, ‘销售部’, 12000.00),
(‘王五’, ‘财务部’, 13000.00);

— 创建MySQL CDC源表
CREATE TABLE mysql_cdc_source (
id INT,
name STRING,
department STRING,
salary DECIMAL(10,2),
created_at TIMESTAMP,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
‘connector’ = ‘mysql-cdc’,
‘hostname’ = ‘192.168.1.51’,
‘port’ = ‘3306’,
‘username’ = ‘fgedu’,
‘password’ = ‘fgedu_mysql_2024’,
‘database-name’ = ‘fgedudb’,
‘table-name’ = ‘fgedu_employees’,
‘server-time-zone’ = ‘Asia/Shanghai’,
‘scan.incremental.snapshot.enabled’ = ‘true’,
‘scan.incremental.snapshot.chunk.size’ = ‘8096’
);

— 创建Kafka目标表
CREATE TABLE kafka_sink (
id INT,
name STRING,
department STRING,
salary DECIMAL(10,2),
created_at TIMESTAMP,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
‘connector’ = ‘kafka’,
‘topic’ = ‘mysql-cdc-fgedudb’,
‘properties.bootstrap.servers’ = ‘192.168.1.51:9092’,
‘format’ = ‘json’,
‘json.timestamp-format.standard’ = ‘ISO_8601’
);

— 执行同步
INSERT INTO kafka_sink SELECT * FROM mysql_cdc_source;

# 提交任务到Flink
$ flink run -d /data/flink/lib/flink-connector-mysql-cdc-3.1.0.jar -j /data/flink/jobs/mysql-cdc-sync.sql

# 输出示例：
Job has been submitted with JobID 1a2b3c4d5e6f7g8h9i0j

# 查看任务状态
$ flink list

# 输出示例：
Waiting for response…
—————— Running/Restarting Jobs ——————-
01.04.2024 10:00:00 : job_id (RUNNING) : MySQL CDC Sync Job
————————————————————–
No scheduled jobs.

# 输出示例：
{“id”:1,”name”:”张三”,”department”:”技术部”,”salary”:15000.00,”created_at”:”2024-04-05T10:00:00″}
{“id”:2,”name”:”李四”,”department”:”销售部”,”salary”:12000.00,”created_at”:”2024-04-05T10:00:00″}
{“id”:3,”name”:”王五”,”department”:”财务部”,”salary”:13000.00,”created_at”:”2024-04-05T10:00:00″}

# 在MySQL中插入新数据
$ mysql -u fgedu -p -e “INSERT INTO fgedudb.fgedu_employees (name, department, salary) VALUES (‘赵六’, ‘市场部’, 14000.00);”

# 再次消费Kafka数据验证增量同步
$ kafka-console-consumer.sh –bootstrap-server 192.168.1.51:9092 –topic mysql-cdc-fgedudb

# 输出示例：
{“id”:4,”name”:”赵六”,”department”:”市场部”,”salary”:14000.00,”created_at”:”2024-04-05T10:05:00″}

# 更新数据验证
$ mysql -u fgedu -p -e “UPDATE fgedudb.fgedu_employees SET salary = 16000.00 WHERE id = 1;”

# Kafka输出示例：
{“id”:1,”name”:”张三”,”department”:”技术部”,”salary”:16000.00,”created_at”:”2024-04-05T10:00:00″}

# 删除数据验证
$ mysql -u fgedu -p -e “DELETE FROM fgedudb.fgedu_employees WHERE id = 3;”

# Kafka输出示例：
{“id”:3,”name”:”王五”,”department”:”财务部”,”salary”:13000.00,”created_at”:”2024-04-05T10:00:00″,”op”:”d”}

mysql> USE fgedudb;
mysql> CREATE TABLE fgedu_test_full (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100),
value INT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

mysql> INSERT INTO fgedu_test_full (name, value) VALUES
(‘test1’, 100),
(‘test2’, 200),
(‘test3’, 300),
(‘test4’, 400),
(‘test5’, 500);

# 创建同步任务
# vi /data/flink/jobs/test-full-sync.sql

CREATE TABLE test_full_source (
id INT,
name STRING,
value INT,
created_at TIMESTAMP,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
‘connector’ = ‘mysql-cdc’,
‘hostname’ = ‘192.168.1.51’,
‘port’ = ‘3306’,
‘username’ = ‘fgedu’,
‘password’ = ‘fgedu_mysql_2024’,
‘database-name’ = ‘fgedudb’,
‘table-name’ = ‘fgedu_test_full’
);

CREATE TABLE test_full_sink (
id INT,
name STRING,
value INT,
created_at TIMESTAMP,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
‘connector’ = ‘print’
);

INSERT INTO test_full_sink SELECT * FROM test_full_source;

# 提交任务
$ flink run -d /data/flink/jobs/test-full-sync.sql

# 查看任务日志
$ tail -f /data/flink/log/flink-*-taskexecutor-*.log | grep test_full

# 输出示例：
test_full_sink: +I[1, test1, 100, 2024-04-05T10:00:00]
test_full_sink: +I[2, test2, 200, 2024-04-05T10:00:00]
test_full_sink: +I[3, test3, 300, 2024-04-05T10:00:00]
test_full_sink: +I[4, test4, 400, 2024-04-05T10:00:00]
test_full_sink: +I[5, test5, 500, 2024-04-05T10:00:00]

# 查看任务日志
$ tail -f /data/flink/log/flink-*-taskexecutor-*.log | grep test_full

# 输出示例：
test_full_sink: +I[6, test6, 600, 2024-04-05T10:05:00]

# 更新数据
mysql> UPDATE fgedu_test_full SET value = 150 WHERE id = 1;

# 输出示例：
test_full_sink: -U[1, test1, 100, 2024-04-05T10:00:00]
test_full_sink: +U[1, test1, 150, 2024-04-05T10:00:00]

# 删除数据
mysql> DELETE FROM fgedu_test_full WHERE id = 5;

# 输出示例：
test_full_sink: -D[5, test5, 500, 2024-04-05T10:00:00]

# 输出示例：
—————— Running/Restarting Jobs ——————-
01.04.2024 10:00:00 : abc123def456 (RUNNING) : test-full-sync
————————————————————–

# 创建savepoint
$ flink savepoint abc123def456 /data/flink/savepoints/test-full-sync

# 输出示例：
Triggering savepoint for job abc123def456.
Waiting for response…
Savepoint completed. Path: file:///data/flink/savepoints/test-full-sync/savepoint-abc123-1234567890

# 停止任务
$ flink stop abc123def456

# 输出示例：
Suspending job abc123def456 with a savepoint.

# 在MySQL中插入数据（任务停止期间）
mysql> INSERT INTO fgedu_test_full (name, value) VALUES (‘test7’, 700);
mysql> INSERT INTO fgedu_test_full (name, value) VALUES (‘test8’, 800);

# 从savepoint恢复任务
$ flink run -s /data/flink/savepoints/test-full-sync/savepoint-abc123-1234567890 -d /data/flink/jobs/test-full-sync.sql

# 输出示例：
Job has been submitted with JobID xyz789abc123

# 查看任务日志验证数据同步
$ tail -f /data/flink/log/flink-*-taskexecutor-*.log | grep test_full

# 输出示例：
test_full_sink: +I[7, test7, 700, 2024-04-05T10:10:00]
test_full_sink: +I[8, test8, 800, 2024-04-05T10:10:00]

# 增加TaskManager slots
taskmanager.numberOfTaskSlots: 16

# 增加默认并行度
parallelism.default: 8

# 在SQL中设置并行度
# vi /data/flink/jobs/mysql-cdc-sync-optimized.sql

— 设置作业配置
SET ‘execution.parallelism’ = ‘8’;
SET ‘execution.checkpointing.interval’ = ’30s’;
SET ‘execution.checkpointing.mode’ = ‘EXACTLY_ONCE’;
SET ‘execution.checkpointing.timeout’ = ’10min’;
SET ‘state.backend’ = ‘rocksdb’;
SET ‘state.backend.incremental’ = ‘true’;

— 创建源表（增加并行读取配置）
CREATE TABLE mysql_cdc_source (
id INT,
name STRING,
department STRING,
salary DECIMAL(10,2),
created_at TIMESTAMP,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
‘connector’ = ‘mysql-cdc’,
‘hostname’ = ‘192.168.1.51’,
‘port’ = ‘3306’,
‘username’ = ‘fgedu’,
‘password’ = ‘fgedu_mysql_2024’,
‘database-name’ = ‘fgedudb’,
‘table-name’ = ‘fgedu_employees’,
‘scan.incremental.snapshot.enabled’ = ‘true’,
‘scan.incremental.snapshot.chunk.size’ = ‘8096’,
‘scan.snapshot.fetch.size’ = ‘2048’
);

— 创建目标表
CREATE TABLE kafka_sink (
id INT,
name STRING,
department STRING,
salary DECIMAL(10,2),
created_at TIMESTAMP,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
‘connector’ = ‘kafka’,
‘topic’ = ‘mysql-cdc-fgedudb’,
‘properties.bootstrap.servers’ = ‘192.168.1.51:9092’,
‘format’ = ‘json’,
‘sink.partitioner’ = ‘fixed’
);

— 执行同步
INSERT INTO kafka_sink SELECT * FROM mysql_cdc_source;

# TaskManager内存配置
taskmanager.memory.process.size: 32768m
taskmanager.memory.flink.size: 24576m
taskmanager.memory.task.heap.size: 16384m
taskmanager.memory.managed.fraction: 0.4
taskmanager.memory.network.fraction: 0.15
taskmanager.memory.network.max: 2048m
taskmanager.memory.network.min: 512m

# JVM配置
env.java.opts.taskmanager: “-XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:+PrintGCDetails -XX:+PrintGCDateStamps”

# RocksDB状态后端配置
state.backend.rocksdb.localdir: /data/flink/rocksdb
state.backend.rocksdb.block.cache-size: 512m
state.backend.rocksdb.write.buffer.size: 64m
state.backend.rocksdb.write.batch.size: 2m

# 重启Flink集群
# stop-cluster.sh
# start-cluster.sh

# 检查点配置
execution.checkpointing.interval: 30000
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.timeout: 600000
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.min-pause: 500
execution.checkpointing.tolerable-failure-number: 3
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

# 状态后端配置
state.backend: rocksdb
state.backend.incremental: true
state.checkpoints.dir: hdfs://192.168.1.51:8020/flink/checkpoints
state.savepoints.dir: hdfs://192.168.1.51:8020/flink/savepoints

# 本地恢复
state.backend.local-recovery: true
taskmanager.state.local.root-dirs: /data/flink/local-recovery

# 重启策略
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 5
restart-strategy.fixed-delay.delay: 30s

# 在SQL中配置检查点
SET ‘execution.checkpointing.interval’ = ’30s’;
SET ‘execution.checkpointing.mode’ = ‘EXACTLY_ONCE’;
SET ‘execution.checkpointing.timeout’ = ’10min’;
SET ‘execution.checkpointing.min-pause’ = ‘500ms’;
SET ‘execution.checkpointing.max-concurrent-checkpoints’ = ‘1’;
SET ‘execution.checkpointing.externalized-checkpoint-retention’ = ‘RETAIN_ON_CANCELLATION’;
SET ‘state.backend’ = ‘rocksdb’;
SET ‘state.backend.incremental’ = ‘true’;