Flink CDC 原理、实践和优化

CREATE TABLE `Data_Input` (
    id BIGINT,
    actor VARCHAR,
    alias VARCHAR,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'kafka',                             -- 可选 'kafka','kafka-0.11'. 注意选择对应的内置  Connector
    'topic' = 'YourDebeziumTopic',                     -- 替换为您要消费的 Topic
    'scan.startup.mode' = 'earliest-offset'            -- 可以是 latest-offset  earliest-offset  specific-offsets  group-offsets 的任何一种
    'properties.bootstrap.servers' = '10.0.1.2:9092',  -- 替换为您的 Kafka 连接地址
    'properties.group.id' = 'YourGroup',               -- 必选参数, 一定要指定 Group ID

    -- 定义数据格式 (Debezium JSON 格式)
    'format' = 'debezium-json',
    'debezium-json.schema-include' = 'false',
);

CREATE TABLE `Data_Output` (
    id BIGINT,
    actor VARCHAR,
    alias VARCHAR,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:postgresql://postgresql.example:50060/myDatabase?currentSchema=mySchema&reWriteBatchedInserts=true', -- 请替换为您的实际 PostgreSQL 连接参数
    'table-name' = 'MyTable',   -- 需要写入的数据表
    'username' = 'user',        -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'helloworld'   -- 数据库访问的密码
);

INSERT INTO `Data_Output` SELECT * FROM `Data_Input`;
复制

CREATE TABLE `Data_Input` (
    id BIGINT,
    actor VARCHAR,
    alias VARCHAR,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',    -- 可选 'mysql-cdc' 和 'postgres-cdc'
    'hostname' = '192.168.10.22',   -- 数据库的 IP
    'port' = '3306',                -- 数据库的访问端口
    'username' = 'debezium',        -- 数据库访问的用户名（需要提供 SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, SELECT, RELOAD 权限）
    'password' = 'hello@world!',    -- 数据库访问的密码
    'database-name' = 'YourData',   -- 需要同步的数据库
    'table-name' = 'YourTable'      -- 需要同步的数据表名
);

CREATE TABLE `Data_Output` (
    id BIGINT,
    actor VARCHAR,
    alias VARCHAR,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:postgresql://postgresql.example:50060/myDatabase?currentSchema=mySchema&reWriteBatchedInserts=true', -- 请替换为您的实际 PostgreSQL 连接参数
    'table-name' = 'MyTable',   -- 需要写入的数据表
    'username' = 'user',        -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'helloworld'   -- 数据库访问的密码
);

INSERT INTO `Data_Output` SELECT * FROM `Data_Input`;
复制

GenericRowData before = (GenericRowData) payload.getField(0);  // 更新前的数据GenericRowData after = (GenericRowData) payload.getField(1);   更新后的数据String op = payload.getField(2).toString();             获取 "op" 字段的类型if (OP_CREATE.equals(op) || OP_READ.equals(op)) {   如果是创建 (c) 或快照读取 (r) 消息    after.setRowKind(RowKind.INSERT);        // 设置消息类型为新建 (+I)    out.collect(after);                // 发送给下游} else if (OP_UPDATE.equals(op)) {        // 如果是更新 (u) 消息    before.setRowKind(RowKind.UPDATE_BEFORE);  // 把更新前的数据类型设置为撤回 (-U)    after.setRowKind(RowKind.UPDATE_AFTER);    // 把更新后的数据类型设置为更新 (+U)    out.collect(before);            // 发送两条数据给下游    out.collect(after);} else if (OP_DELETE.equals(op)) {    // 如果是删除 (d) 消息    before.setRowKind(RowKind.DELETE);  // 将消息类型设置为删除 (-D)    out.collect(before);        // 发送给下游} else {    ...  // 异常处理逻辑}
复制

this.engine = DebeziumEngine.create(Connect.class)    .using(properties) // 初始化 Debezium 所需的参数    .notifying(debeziumConsumer) // 收到批量的变更消息, 则 Debezium 会回调 DebeziumChangeConsumer 来反序列化并向下游输出数据    .using(OffsetCommitPolicy.always())    .using(            (success, message, error) -> {                if (!success && error != null) {                    this.reportError(error);                }            })    .build();...     executor.execute(engine);  // 向 Executor 提交 Debezium 线程以启动运行
复制

for (ChangeEvent<SourceRecord, SourceRecord> event : changeEvents) {  // 轮询各个事件
    SourceRecord record = event.value();
    if (isHeartbeatEvent(record)) {    // 如果时心跳包
        // 只更新当前 offset 信息, 然后继续（不进行实际处理）
        synchronized (checkpointLock) {
            debeziumOffset.setSourcePartition(record.sourcePartition());
            debeziumOffset.setSourceOffset(record.sourceOffset());
        }
        continue;
    }


    deserialization.deserialize(record, debeziumCollector);    // 反序列化这条消息


    if (isInDbSnapshotPhase) {  // 如果处于数据库快照期, 需要阻止 Flink 检查点（Checkpoint）生成
        if (!lockHold) {
            MemoryUtils.UNSAFE.monitorEnter(checkpointLock);
            lockHold = true;
            ...
        }
        if (!isSnapshotRecord(record)) {  // 如果已经不在数据库快照期了, 就释放锁, 允许 Flink 正常生成检查点（Checkpoint）
            MemoryUtils.UNSAFE.monitorExit(checkpointLock);
            isInDbSnapshotPhase = false;
            ...
        }
    }


  // 更新当前 offset 信息, 并向下游 Flink 算子发送数据
    emitRecordsUnderCheckpointLock(
        debeziumCollector.records, record.sourcePartition(), record.sourceOffset());
}
复制

SET GLOBAL binlog_format = 'ROW';
SET GLOBAL binlog_row_image = 'FULL';
复制

GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '用户名' IDENTIFIED BY '密码';
FLUSH PRIVILEGES;
复制