Flink cdc 介绍及使用 FlinkCDC读取mysql 及 jdbc 连接参数配置

Flink cdc 介绍及使用 FlinkCDC读取mysql 及 jdbc 连接参数配置、官方案例

1. Flink cdc 介绍

CDC 的全称是 Change Data Capture ，在广义的概念上，只要是能捕获数据变更的技术，我们都可以称之为 CDC。

目前通常描述的 CDC 技术主要面向数据库的变更，是一种用于捕获数据库中数据变更的技术。CDC 技术的应用场景非常广泛；

Flink 的 cdc 是基于日志的：实时消费日志，流处理，例如 MySQL 的 binlog 日志完整记录了数据库中的变更，可以把 binlog文件当作流的数据源；保障数据一致性，因为 binlog 文件包含了所有历史变更明细；保障实时性，因为类似 binlog的日志文件是可以流式消费的，提供的是实时数据。

2. 常见cdc开源方案

3. Flink cdc 使用案例

3.1 Mysql开启binlog

修改配置文件

vi /etc/my.cnf

my.cnf文件内容

# 第一个参数是打开binlog日志
log_bin=ON

# 第二个参数是binlog日志的基本文件名，后面会追加标识来表示每一个文件
log_bin_basename=/usr/local/mysql/log-bin/mysql-bin

# 第三个参数指定的是binlog文件的索引文件，这个文件管理了所有的binlog文件的目录
log_bin_index=/usr/local/mysql/log-bin/mysql-bin.index

修改完成后 查看 binlog 开启状态

show variables like '%log_bin%';

如下图所示 ON 为开启状态

3.2 Flink cdc读取mysql 及 jdbc 连接参数配置

flink-connector-mysql-cdc 2.2 版本之前没有找到关于 jdbc 连接参数的配置，此处以 2.2 为主

3.2.1 Maven POM 文件

<dependencies><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.21</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.21</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.11</artifactId><version>1.13.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.13.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.11</artifactId><version>1.13.0</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.1.3</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.49</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner-blink_2.11</artifactId><version>1.13.0</version></dependency><!--mysql cdc --><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>2.2.0</version><!--<scope>provided</scope>--></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.75</version></dependency><!--kafka--><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.1.1</version></dependency><!--本地调试flink ui--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime-web_2.11</artifactId><version>1.13.0</version><scope>compile</scope></dependency></dependencies>

3.2.2 Flink CDC 代码

Flink cdc

package flink_cdc;import com.ververica.cdc.connectors.mysql.source.MySqlSource;import com.ververica.cdc.connectors.mysql.table.StartupOptions;import org.apache.flink.api.common.eventtime.WatermarkStrategy;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Properties;publicclassFlinkCDC_Mysql{publicstaticvoidmain(String[] args) throws Exception {//创建执行环境
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        Properties prop =newProperties();
        prop.setProperty("autoReconnect","true");//创建 Flink-MySQL-CDC 的 Source//initial (default): 在第一次启动时对被监视的数据库表执行初始快照，并继续读取最新的binlog (开启断点续传后从上次消费offset继续消费)//latest-offset: 永远不要在第一次启动时对被监视的数据库表执行快照，只从binlog的末尾读取，这意味着只有自连接器启动以来的更改//timestamp: 永远不要在第一次启动时对监视的数据库表执行快照，直接从指定的时间戳读取binlog。使用者将从头遍历binlog，并忽略时间戳小于指定时间戳的更改事件//specific-offset: 不允许在第一次启动时对监视的数据库表进行快照，直接从指定的偏移量读取binlog。
        MySqlSource<String> build = MySqlSource.<String>builder().serverTimeZone("UTC").hostname("localhost").port(3306).username("root").password("123456").databaseList("test")//tableList为可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式.tableList("test.test").startupOptions(StartupOptions.latest())//自定义反序列化器.deserializer(newFlinkCdcDataDeserializationSchema())//jdbc连接参数配置.jdbcProperties(prop).build();//使用 CDC Source 从 MySQL 读取数据
        DataStreamSource<String> mysqlDS = env.fromSource(build, WatermarkStrategy.noWatermarks(),"MysqlSource");//打印数据
        mysqlDS.printToErr("------>").setParallelism(1);

        mysqlDS.addSink(newMysqlSink());//6.执行任务
        env.execute("FlinkCDC_mysql");}}

自定义反序列化类
自定义反序列化类解析读入mysql的binlog为指定的json格式（实现接口DebeziumDeserializationSchema重写deserialize、getProducedType方法）

package flink_cdc;import com.alibaba.fastjson.JSONObject;import com.ververica.cdc.debezium.DebeziumDeserializationSchema;import io.debezium.data.Envelope;import org.apache.flink.api.common.typeinfo.TypeInformation;import org.apache.flink.util.Collector;import org.apache.kafka.connect.data.Field;import org.apache.kafka.connect.data.Struct;import org.apache.kafka.connect.source.SourceRecord;publicclassFlinkCdcDataDeserializationSchemaimplementsDebeziumDeserializationSchema<String>{

    @Override
    publicvoiddeserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {

        Struct valueStruct =(Struct)sourceRecord.value();
        Struct sourceStruct = valueStruct.getStruct("source");//获取数据库名称,表名,操作类型
        String database = sourceStruct.getString("db");
        String table = sourceStruct.getString("table");
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();if(type.equals("create")) type="insert";
        JSONObject jsonObject =newJSONObject();
        jsonObject.put("database",database);
        jsonObject.put("table",table);
        jsonObject.put("type",type);//格式转换
        Struct beforeStruct = valueStruct.getStruct("before");
        JSONObject beforeDataJson =newJSONObject();if(beforeStruct !=null){for(Field field : beforeStruct.schema().fields()){
                beforeDataJson.put(field.name(),beforeStruct.get(field));}}

        Struct afterStruct = valueStruct.getStruct("after");
        JSONObject afterDataJson =newJSONObject();if(afterStruct !=null){for(Field field : afterStruct.schema().fields()){
                afterDataJson.put(field.name(),afterStruct.get(field));}}

        jsonObject.put("beforeData",beforeDataJson);
        jsonObject.put("afterData",afterDataJson);//向下游传递数据
        collector.collect(jsonObject.toJSONString());}

    @Override
    public TypeInformation<String>getProducedType(){return TypeInformation.of(String.class);}}

Mysql Sink
不需要可忽略

package flink_cdc;import com.alibaba.fastjson.JSONObject;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;import java.sql.Connection;import java.sql.DriverManager;import java.sql.PreparedStatement;publicclassMysqlSinkextendsRichSinkFunction<String>{

    Connection connection =null;
    PreparedStatement insertSmt =null;

    @Override
    publicvoidopen(Configuration parameters) throws Exception {
        String url ="jdbc:mysql://localhost:3306/test?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false";
        connection = DriverManager.getConnection(url,"root","123456");
        insertSmt = connection.prepareStatement("REPLACE into test2(id,name) values (?,?)");}

    @Override
    publicvoidinvoke(String value, Context context) throws Exception {

        System.err.println(value);
        JSONObject jsonObject = JSONObject.parseObject(value);
        System.out.println(jsonObject.get("afterData"));

        TestBean afterData = JSONObject.parseObject(JSONObject.toJSONString(jsonObject.get("afterData")), TestBean.class);//直接执行更新语句
        insertSmt.setInt(1,afterData.getId());
        insertSmt.setString(2,afterData.getName());
        insertSmt.execute();}

    @Override
    publicvoidclose() throws Exception {
        insertSmt.close();
        connection.close();}}

sink测试实体

package flink_cdc;publicclassTestBean{private int id;private String name;public int getId(){return id;}publicvoidsetId(int id){this.id = id;}public String getName(){return name;}publicvoidsetName(String name){this.name = name;}}

4. 补充

基于 Flink CDC 构建 MySQL 和 Postgres 的 Streaming ETL
假设我们正在经营电子商务业务，商品和订单的数据存储在 MySQL 中，订单对应的物流信息存储在 Postgres 中。 对于订单表，为了方便进行分析，我们希望让它关联上其对应的商品和物流信息，构成一张宽表，并且实时把它写到 ElasticSearch 中。

通过Flink Mysql/Postgres CDC 来实现这个需求，系统的整体架构如下图所示：

基于 Flink CDC 同步 MySQL 分库分表构建实时数据湖
在 OLTP 系统中，为了解决单表数据量大的问题，通常采用分库分表的方式将单个大表进行拆分以提高系统的吞吐量。 但是为了方便数据分析，通常需要将分库分表拆分出的表在同步到数据仓库、数据湖时，再合并成一个大表。

将数据从 MySQL 同步到 Iceberg 为例的整个流程，架构图如下所示：

也可以使用不同的 source 比如 Oracle/Postgres 和 sink 比如 Hudi 来构建自己的 ETL 流程。

Flink cdc 官方文档地址：https://ververica.github.io/flink-cdc-connectors/master/index.html