使用Flink SQL CDC读取动态配置表

Love The Way You Lie 2022-11-09 14:29 780阅读 0赞

## Flink SQL CDC 介绍 ##

CDC 全称是`Change Data Capture` ，它是一个比较广义的概念，只要能`捕获变更的数据`，我们都可以称为 CDC 。业界主要有`基于查询的 CDC 和基于日志的 CDC` ，可以从下面表格对比他们功能和差异点。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

**传统的数据同步场景** :  
缺点: 采集端组件过多导致维护繁杂  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

**改进后的架构**:  
![在这里插入图片描述][20210315203822269.png]

Flink社区开发了 flink-cdc-connectors 组件，这是一个`可以直接从 MySQL、PostgreSQL 等数据库直接读取全量数据和增量变更数据的 source 组件`。目前也已开源，开源地址：[https://github.com/ververica/flink-cdc-connectors  
][https_github.com_ververica_flink-cdc-connectors]  
![在这里插入图片描述][2021031520391253.png]  
![在这里插入图片描述][20210315203918235.png]

## 具体使用 ##

#### 修改mysql配置 ####

增加对数据库gmall2021\_realtime监控

[mysqld]
    server-id= 1
    log-bin=mysql-bin
    binlog_format=row
    binlog-do-db=gmall2021_realtime

注意:

1.  需要重启mysql数据库

sudo systemctl restart mysqld

1.  确认msyql有没有启动成功

ps -ef | grep mysqld

1.  注意maxwell不要再采集这个数据库的数据,在maxwell的配置中添加如下配置

filter=exclude:gmall2021_realtime.*

#### 导入CDC依赖 ####

<dependency>
        <groupId>com.alibaba.ververica</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>1.1.1</version>
    </dependency>

#### 具体实现代码 ####

/** * 读取的配置表的数据 * * @param env * @return */
    private SingleOutputStreamOperator<TableProcess> readProcessTable(StreamExecutionEnvironment env) { 
        // 配置表在jdbc, 使用flink-sql-cdc来完成
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv
            .executeSql("CREATE TABLE `table_process`( " +
                            " `source_table` string, " +
                            " `operate_type` string, " +
                            " `sink_type` string, " +
                            " `sink_table` string, " +
                            " `sink_columns` string, " +
                            " `sink_pk` string, " +
                            " `sink_extend` string, " +
                            " PRIMARY KEY (`source_table`,`operate_type`) NOT ENFORCED" +
                            ")with(" +
                            " 'connector' = 'mysql-cdc', " +
                            " 'hostname' = 'hadoop162', " +
                            " 'port' = '3306', " +
                            " 'username' = 'root', " +
                            " 'password' = 'aaaaaa', " +
                            " 'database-name' = 'gmall2021_realtime', " +
                            " 'table-name' = 'table_process'," +
                            " 'debezium.snapshot.mode' = 'initial' " +  // 读取mysql的全量,增量以及更新数据
                            ")");
        final Table table = tenv.sqlQuery("select " +
                                              " source_table sourceTable, " +
                                              " sink_type sinkType, " +
                                              " operate_type operateType, " +
                                              " sink_table sinkTable, " +
                                              " sink_columns sinkColumns, " +
                                              " sink_pk sinkPk, " +
                                              " sink_extend sinkExtend " +
                                              "from table_process ");
        return tenv
            .toRetractStream(table, TableProcess.class)
            .filter(t -> t.f0)
            .map(t -> t.f1);
    }

最新版1.2.0已经更新，但是测试发现还有很多问题，待解决后再更新。。。。。。。。。。。。。。。。。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221023/4614435b0bde4e84b13086d14e364766.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221023/f29fd3030f86443f921dc7f5fcd08150.png
[20210315203822269.png]: /images/20221023/78cffc902f354abb943bc530af98a596.png
[https_github.com_ververica_flink-cdc-connectors]: https://github.com/ververica/flink-cdc-connectors
[2021031520391253.png]: /images/20221023/9c35972fa9874eb19d089d90ffd5cb67.png
[20210315203918235.png]: /images/20221023/c4cad8e635564515931df47aedbdc884.png