Flink CDC

1、Flink CDC

1.1、什么是 Flink CDC？

        CDC（Change Data Capture）也就是变更数据采集的意思，我们之前学的 Maxwell 就是一种 CDC 工具。今天要学的 Flink CDC 则是 Flink 自己开发的一款与 Flink 自身无缝集成的 CDC 工具，它可以将源数据库（Source）的增量变动记录，同步到一个或多个数据目的（Sink）。在同步过程中，还可以对数据进行一定的处理，例如过滤、关联、分组、统计等。

        目前专业做数据库事件接受和解析的中间件是Debezium，如果是捕获Mysql，还有Canal。而Flink 正是集成了 Debezium 作为捕获数据变更的引擎，所以它可以充分发挥 Debezium 的能力。Flink 和 Debezium 最大的不同之处在于，Flink 并不强制依赖于 Kafka。

1.2、为什么要使用 Flink CDC？

        之前我们监听 MySQL 的业务数据变更是通过 Maxwell 或者可以通过 Cannal ，它们都是通过监听 binlog 日志来实现数据同步的。

        在实时数仓中，如果我们要使用上面的 Maxwell 或 Cannal 做业务数据同步，那必须通过下面这几步：

• mysql 开启 binlog
• Maxwell/Canal 同步 binlog 数据写入到 Kafka
• Flink 读取 Kakfa 中的 binlog 数据进行相关的业务处理

整体来看步骤很长，而且用到的组件也比较多，那能不能直接 Flink 一步到位，直接自己通过 binlog 来实现日志同步呢？所以 Flink CDC 这就出现了。

Flink CDC 数据格式

{
    "before": {
        "id": "PF1784570096901248",
        "pay_order_no": null,
        "out_no": "J1784570080435328",
        "title": "充值办卡",
        "from_user_id": "PG11111",
        "from_account_id": "1286009802396288",
        "user_id": "BO1707796995184000",
        "account_id": "1707895210106496",
        "amount": 13400,
        "profit_state": 1,
        "profit_time": 1686758315000,
        "refund_state": 0,
        "refund_time": null,
        "add_time": 1686758315000,
        "remark": "充值办卡",
        "acct_circle": "PG11111",
        "user_type": 92,
        "from_user_type": 90,
        "company_id": "PG11111",
        "profit_mode": 1,
        "type": 2,
        "parent_id": null,
        "oc_profit_id": "1784570096901248",
        "keep_account_from_user_id": null,
        "keep_account_from_bm_user_id": null,
        "keep_account_user_id": null,
        "keep_account_bm_user_id": null,
        "biz_company_id": "PG11111"
    },
    "after": {
        "id": "PF1784570096901248",
        "pay_order_no": null,
        "out_no": "J1784570080435328",
        "title": "充值办卡",
        "from_user_id": "PG11111",
        "from_account_id": "1286009802396288",
        "user_id": "BO1707796995184000",
        "account_id": "1707895210106496",
        "amount": 13400,
        "profit_state": 1,
        "profit_time": 1686758315000,
        "refund_state": 0,
        "refund_time": null,
        "add_time": 1686758315000,
        "remark": "充值办卡1",
        "acct_circle": "PG11111",
        "user_type": 92,
        "from_user_type": 90,
        "company_id": "PG11111",
        "profit_mode": 1,
        "type": 2,
        "parent_id": null,
        "oc_profit_id": "1784570096901248",
        "keep_account_from_user_id": null,
        "keep_account_from_bm_user_id": null,
        "keep_account_user_id": null,
        "keep_account_bm_user_id": null,
        "biz_company_id": "PG11111"
    },
    "source": {
        "version": "1.6.4.Final",
        "connector": "mysql",
        "name": "mysql_binlog_source",
        "ts_ms": 1686734882000,
        "snapshot": "false",
        "db": "cloud_test",
        "sequence": null,
        "table": "acct_profit",
        "server_id": 1,
        "gtid": null,
        "file": "mysql-bin.000514",
        "pos": 650576218,
        "row": 0,
        "thread": null,
        "query": null
    },
    "op": "u",
    "ts_ms": 1686734882689,
    "transaction": null
}

1.3、使用 Flink CDC 

MySQL 首先得开启 binlog 功能，这里需要特别注意 MySQL 时区的设置（我们需要保证 Flink 的时间和数据库时间保持一致）：

show variables like '%time_zone%';
# 如果失去是UTC，则需要时间校正
set time_zone='+8:00';

1.3.1、变更数据处理逻辑

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

public class CustomSink extends RichSinkFunction<String> {

    // 自定义向下游写入
    @Override
    public void invoke(String json, Context context) throws Exception {
        // 每次产生变更操作后对数据进行什么操作，比如写入到ES、Redis、MongoDB等
        System.out.println(">>>" + json);
    }

    // 写入其它数据源需要创建连接
    @Override
    public void open(Configuration parameters) throws Exception {

    }
    // 关闭外部数据库连接
    @Override
    public void close() throws Exception {

    }
}

1.3.2、构建 Flink CDC 监听程序

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MySqlSourceExample {
    public static void main(String[] args) throws Exception {
        MySqlSource<String> source = MySqlSource.builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("数据库名")
                .tableList("表名")    // 多个表之间用逗号切割
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema())  // 反序列化器
                .includeSchemaChanges(true) // 是否监听表结构的变化
                .build();

        // 启动本地 WEB-UI
        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT,8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        // 检查点时间间隔
        env.enableCheckpointing(5000);
        DataStreamSink<String> sink = env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .addSink(new CustomSink());

        env.execute();
    }
}