【大数据】Flink CDC 实时同步mysql数据

一、前言

在微服务系统架构中，经常会涉及到跨系统的数据同步，比如需要从A系统的mysql数据库同步到B系统的oracle数据库，再比如说需要将mysql中的数据同步到es中，更有一些场景下，涉及到多个系统的异构数据源，需要加工处理后将数据推送到kafka中。在这些场景下，需求各异，需要满足的场景也不尽相同。很难有某一种开源工具可以满足所有的场景，接下来本文将介绍使用flinkcdc的方式同步mysql的数据。

二、Flink CDC介绍

2.1 什么是Flink CDC

Flink CDC（Change Data Capture）是一个用于实时捕获数据库变更事件的工具，它是基于 Apache Flink 构建的。Flink CDC 可以从关系型数据库中实时捕获表的数据变更事件，并将这些事件转化为流式数据，以便进行实时处理和分析。这对于实时数据仓库、实时数据分析以及数据同步等场景非常有用。

Flink-CDC 开源地址： Apache/Flink-CDC

Flink-CDC 中文文档：Apache Flink CDC | Apache Flink CDC

Flink CDC connector 可以捕获在一个或多个表中发生的所有变更。该模式通常有一个前记录和一个后记录。Flink CDC connector 可以直接在Flink中以非约束模式（流）使用，而不需要使用类似 kafka 之类的中间件中转数据。

Flink SQL CDC 内置了 Debezium 引擎，利用其抽取日志获取变更的能力，将 changelog 转换为 Flink SQL 认识的 RowData 数据。

2.2 Flink CDC 特点

Flink CDC 具备如下特点

实时性：Flink CDC 支持实时捕获数据库中的数据变更事件，这意味着它可以立即响应数据库中的任何更改。
支持多种数据库：Flink CDC 支持多种关系型数据库，如 MySQL、PostgreSQL、Oracle 等。
易于集成：Flink CDC 可以轻松集成到现有的 Flink 应用程序中，利用 Flink 强大的流处理能力来处理捕获的数据。
高可用性：Flink CDC 设计为高可用的，可以在分布式环境中运行，并且支持故障恢复和容错机制。
灵活的数据处理：捕获的数据可以被转换、过滤、聚合等，支持复杂的数据处理逻辑。

2.3 Flink CDC 核心工作原理

Flink CDC 的工作原理主要包括以下几个步骤：

连接数据库：Flink CDC 通过 JDBC 连接到目标数据库，并监听数据库的变更事件。
捕获变更事件：通过监听数据库的日志（如 MySQL 的 Binlog、PostgreSQL 的 WAL 日志等），捕获数据的插入、更新、删除等变更事件。
转换为流数据：将捕获的变更事件转换为 Flink 可处理的流数据格式。
处理数据：将转换后的流数据送入 Flink 的流处理管道，进行进一步的数据处理，如清洗、聚合、分析等。
输出结果：处理后的数据可以被输出到其他系统，如写入另一个数据库、发送到消息队列、写入文件系统等。

2.4 Flink CDC 使用场景

Flink CDC（Change Data Capture）是一种用于实时捕获数据库变更事件的技术，它能够从关系型数据库中捕获表的数据变更，并将这些变更事件转化为流式数据，以便进行实时处理。基于这个特性，在下面这些场景下，可以考虑使用Flink CDC处理：

实时数据仓库- 实时更新数据仓库：Flink CDC 可以实时捕获数据库中的变更事件，并将这些事件实时地加载到数据仓库中，从而实现实时的数据更新和查询。- 实时分析：通过捕获和处理实时数据，企业可以进行实时的业务分析，比如实时监控销售数据、实时统计用户行为等。
实时数据同步- 数据库间同步：Flink CDC 可以将一个数据库中的变更事件实时同步到另一个数据库，实现数据的实时复制。这对于需要跨数据中心或多云环境的数据同步场景非常有用。- 异构系统间同步：可以将传统的关系型数据库的数据实时同步到 NoSQL 数据库、搜索引擎或其他存储系统中，实现数据的实时一致性和可用性。
实时数据处理- 实时ETL：Flink CDC 可以作为 ETL（Extract, Transform, Load）流程的一部分，用于实时提取、转换和加载数据。例如，可以从数据库中实时提取数据，进行清洗和聚合处理后，再加载到数据仓库或其他系统中。- 实时统计分析：通过对捕获的变更事件进行实时处理，可以实现各种实时统计分析，如实时流量统计、实时交易统计等。
实时告警和监控- 实时监控：Flink CDC 可以实时捕获数据库中的变更事件，并对这些事件进行实时分析和处理，用于实时监控关键指标的变化。- 实时告警：当检测到异常数据变更时，可以实时触发告警通知，帮助企业及时响应和处理问题。
实时推荐系统- 实时用户行为追踪：通过对用户行为数据的实时捕获和处理，可以实现实时的个性化推荐。- 实时更新推荐模型：Flink CDC 可以实时捕获用户的最新行为数据，并用于实时更新推荐模型，从而提高推荐的准确性和时效性。
实时交易处理- 金融交易：在金融行业中，Flink CDC 可以用于实时捕获和处理交易数据，实现高速的交易处理和结算。- 欺诈检测：通过对实时交易数据的分析，可以快速检测出潜在的欺诈行为，并及时采取措施。
物联网（IoT）应用- 实时数据采集：在 IoT 场景中，Flink CDC 可以用于实时采集设备数据，并进行实时处理和分析。- 实时决策支持：通过对 IoT 数据的实时分析，可以支持实时的决策制定，如实时调整设备的工作状态、优化资源配置等。
实时日志处理- 实时日志分析：Flink CDC 可以用于实时捕获数据库的日志数据，并进行实时处理和分析，用于性能监控、错误诊断等。- 审计与合规：通过实时捕获数据库的操作记录，可以用于审计和合规性的检查。

三、常用的数据同步方案对比

3.1 数据同步概述

当系统发展到一定阶段，尤其是系统的规模越来越大，业务体量也不断扩大的时候，一个系统可能会用到多种数据存储中间件，而不再是单纯的mysql，pgsql等，甚至一个系统中一个或多个微服务无法再满足业务的需求，而需要单独做数据存储的服务，在类似的场景下，很难避免新的服务需要从原有的微服务中抽取数据，或定期做数据的同步处理，诸如此类的场景还有很多，如何处理数据同步的需求呢？这就是下面要讨论的问题。

3.1.1 数据同步来源

系统数据同步的需求通常源于多种业务和技术上的考量。以下是一些主要的原因，解释为什么需要进行系统数据同步：

高可用性和灾难恢复- 数据冗余：通过将数据同步到多个位置，可以增加数据的冗余度，减少单点故障的风险。- 灾难恢复：在发生硬件故障、自然灾害或其他不可预见事件时，可以迅速切换到备用数据副本，确保业务连续性
负载均衡和性能优化- 读写分离：将读操作和写操作分开处理，通过将读操作分散到多个数据库实例上来降低单一数据库的负载。- 性能优化：通过将热点数据同步到性能更高的存储设备或系统，提升数据访问速度和系统整体性能。
地理分布- 全球分布：在全球范围内分布数据，以减少网络延迟，提高用户体验。- 多地备份：在不同地理位置建立数据备份中心，以应对局部故障或自然灾害的影响。
数据整合- 数据仓库：将来自不同源的数据整合到一个统一的数据仓库中，以便进行集中管理和分析。- 实时分析：通过实时同步数据，可以快速获取最新的业务数据，进行实时分析和决策支持。
业务扩展- 系统迁移：当公司需要更换数据库系统或升级现有系统时，需要将数据从旧系统同步到新系统。- 新系统上线：在部署新的应用程序或服务时，需要将现有数据同步到新的系统中。
数据共享- 多部门协作：不同部门或团队需要访问相同的数据集，以促进信息共享和协同工作。- 合作伙伴集成：与外部合作伙伴或第三方系统共享数据，实现数据互通和业务合作。
数据一致性- 分布式事务：在分布式系统中，需要保证数据的一致性，通过数据同步来协调不同节点之间的状态。- 分布式缓存：在使用分布式缓存（如 Redis Cluster）时，需要将数据同步到不同的缓存节点，保持数据的一致性。
数据迁移- 数据转移：在进行数据迁移时，需要将数据从一个系统或数据库转移到另一个系统或数据库，以适应业务发展和技术进步的需要。

3.2 常用的数据同步方案汇总

下图列举了几种主流的用于解决数据同步场景的方案

关于图中几种技术，做如下简单的介绍，便于做技术选型作为对比

Debezium- Debezium是国外⽤户常⽤的CDC组件，单机对于分布式来说，在数据读取能力的拓展上，没有分布式的更具有优势，在大数据众多的分布式框架中（Hive、Hudi等）Flink CDC 的架构能够很好地接入这些框架。
DataX- DataX无法支持增量同步。如果一张Mysql表每天增量的数据是不同天的数据，并且没有办法确定它的产生时间，那么如何将数据同步到数仓是一个值得考虑的问题。DataX支持全表同步，也支持sql查询的方式导入导出，全量同步一定是不可取的，sql查询的方式没有可以确定增量数据的字段的话也不是一个好的增量数据同步方案。
Canal- Canal是用java开发的基于数据库增量日志解析，提供增量数据订阅&消费的中间件。Canal主要支持了MySQL的Binlog解析，将增量数据写入中间件中（例如kafka,Rocket MQ等），但是无法同步历史数据，因为无法获取到binlog的变更。
Sqoop- Sqoop主要用于在Hadoop(Hive)与传统的数据库(mysql、postgresql...)间进行数据的传递。Sqoop将导入或导出命令翻译成mapreduce程序来实现，这样的弊端就是Sqoop只能做批量导入，遵循事务的一致性，Mapreduce任务成功则同步成功，失败则全部同步失败。
Kettle- Kettle是一款开源的数据集成工具，主要用于数据抽取、转换和加载（ETL）。它提供了图形化的界面，使得用户可以通过拖拽组件的方式来设计和执行数据集成任务。Kettle 被广泛应用于数据仓库构建、数据迁移、数据清洗等多种场景。- 虽然 Kettle 在处理中小型数据集时表现良好，但在处理大规模数据集时可能会遇到性能瓶颈，尤其是在没有进行优化的情况下。- Kettle 在运行时可能会消耗较多的内存和 CPU 资源，特别是当处理复杂的转换任务时。- 虽然 Kettle 的基本操作相对简单，但对于高级功能和复杂任务的设计，用户仍需投入一定的时间和精力来学习和掌握。
Flink CDC- Flink CDC 基本都弥补了以上框架的不足，将数据库的全量和增量数据一体化地同步到消息队列和数据仓库中；也可以用于实时数据集成，将数据库数据实时入湖入仓；无需像其他的CDC工具一样需要在服务器上进行部署，减少了维护成本，链路更少；完美套接Flink程序，CDC获取到的数据流直接对接Flink进行数据加工处理，一套代码即可完成对数据的抽取转换和写出，既可以使用flink的DataStream API完成编码，也可以使用较为上层的FlinkSQL API进行操作。

3.3 为什么推荐Flink CDC

在java生态中，以一个使用较多的场景为例说明，如下，是一个业务系统与大数据系统进行交互时的数据处理流

在上面这个处理流程下，业务系统将数据正常写入mysql数据库，数据库需要开启binlog，然后借助canal监听binlog把日志写入到kafka中，而Flink实时消费Kakfa的数据实现mysql数据的同步或其他内容等，所以整个过程分为下面几个步骤：

mysql开启binlog
canal同步binlog数据写入到kafka
flink（或其他组件）读取kakfa中的binlog数据进行相关的业务处理。

不难发现，上面的处理流程整体的链路较长，需要用到的组件也比较多，一旦中间某个组件运行过程中发生问题，下游的其他业务均会中断，系统的脆弱性风险就增加了，而使用了 Flink CDC之后，则会变成下面这样了，也就是说，只需要数据库开启binlog即可，而后Flink CDC就可以自动进行数据同步了，整个链路就大大缩减了

3.4 Flink CDC 适用范围

Flink CDC（Change Data Capture）连接器是 Apache Flink 社区为 Flink 提供的一种用于捕获数据库变更事件的工具。它允许用户从关系型数据库中实时捕获表的数据变更，并将这些变更事件转化为流式数据，以便进行实时处理，比如你需要将mysql的数据同步到另一个mysql数据库，就需要使用mysql连接器，如果需要同步mongodb的数据，则需要使用mongodb的连接器。截止到Flink CDC 2.2 为止，支持的连接器：

支Flink CDC 持的Flink版本，在实际使用的时候需要根据版本的对照进行选择：

3.5 Flink CDC不同版本对比

Flink CDC技术发展到今天，历经了多个版本，所以底层在实现数据同步的时候技术也发生了变化，在实际开发使用中需要搞清楚这一点，为了加深对此处的理解，下面列举了几个版本的实现技术对比

3.5.1 Flink CDC 1.x

在 Flink cdc 1.x 版本中，底层选用 debezium 作为采集工具，Debezium 为保证数据一致性，通过对读取的数据库或者表进行加锁，加锁是在全量的时候加锁。

下图是开发者社区的一张全局锁和表锁的过程图

FlinkCDC全量同步时会获取全局读锁，或者表锁。所谓加锁，目的是为了确认Mysql binlog 的起始位置和Mysql 表的Schema，获取到锁后，Mysql所有的ddl，dml操作都会处于wait read lock阶段，如果锁获取时间超时，程序还会抛出异常；而增量同步时，因为是监控binlog的方式，所以对mysql没有影响。

什么是表锁，表锁会自动加锁。查询操作（SELECT），会自动给涉及的所有表加读锁，更新操作（UPDATE、DELETE、INSERT）。如果启动一个CDC任务，而另一个CDC程序也处于初始化阶段，获取不到全局锁，那么那么此程序就会去获取表级锁，表及锁锁的时间会更长，一般是全局读锁的几十倍时长。

Flink CDC 1.x 可以不加锁，能够满足大部分场景，但牺牲了一定的数据准确性。Flink CDC 1.x 默认加全局锁，虽然能保证数据一致性，但存在上述 hang 住数据的风险。由此可以看来FlinkCDC 1.x 存在着一些不足：

由于其锁机制，全量同步阶段之有一个任务在进行同步，不支持并发同步，数据传输会比较慢。
锁表时会阻止其他事务提交。
不支持断点续传，如果在同步过程中，出现mysql连接超时，或者flink程序快照中断，那么我们无法从断开点开始续传，因为目前暂不支持checkpoint。

3.5.2 Flink CDC 2.x

Flink 2.x 时代就是说既然大家都觉得这个锁没有必要了，那就把Debezium的锁踢开，踢开之后我们就引用了Netfix 的DBlog paper 的思想；

DBlog paper的思想说白了也是分而治之的思想，既然我对全局表加锁会有比较大的影响，那么我把数据做切分，每一段数据做大数据一致，那是不是就可以保证数据的整体一致呢？

举例来说：假如mysql表里有100万条数据，那么我把这些数据切分成1万个chunk(chunk 本意是树干，在这里就是一段数据集合的意思）；每个chunk 里面有100条数据，先记录这个chunk里面的最低位和最高位；比如是101和200；读完之后把这个数据存入一个缓冲区，我们叫做buffer；如果后期有个binlog的chunk 过来，比如是105，150和160的数据发生了变化，那么我们就会针对相应的值改动，并不会改动全部的值，改动效率就比较高了，这就是无锁的大致思想了；

水平扩展说白了就是把原来的这个单线程换成多线程的处理嘛，由Flink的source 去实现的；做个checkpoint 也是对最新的那些buffer和chunk 做个checkpoint ，实现了容错；

Flink 2.x不仅引入了增量快照读取机制，还带来了一些其他功能的改进。以下是对Flink 2.x的主要功能的介绍：

增量快照读取：Flink 2.x引入了增量快照读取机制，这是一种全新的数据读取方式。该机制支持并发读取和以chunk为粒度进行checkpoint。在增量快照读取过程中，Flink首先根据表的主键将其划分为多个块（chunk），然后将这些块分配给多个读取器并行读取数据。这一机制极大地提高了数据读取的效率。
精确一次性处理：Flink 2.x引入了Exactly-Once语义，确保数据处理结果的精确一次性。MySQL CDC 连接器是Flink的Source连接器，可以利用Flink的checkpoint机制来确保精确一次性处理。
动态加表：Flink 2.x支持动态加表，通过使用savepoint来复用之前作业的状态，解决了动态加表的问题。
无主键表的处理：Flink 2.x对无主键表的读取和处理进行了优化。在无主键表中，Flink可以通过一些额外的字段来识别数据记录的唯一性，从而实现准确的数据读取和处理。

3.5.3 Flink CDC 3.x

虽然 Flink CDC 有很多技术优势，社区用户增长很快，但随着 Flink CDC 项目用户基数的日益增长，以及应用场景的不断扩大，很多问题随着用户在社区的反馈也不断冒出，因此Flink CDC技术团队也适时推出Flink CDC 3.x。

Flink CDC 3.0 的整体架构自顶而下分为 4 层：

Flink CDC API：面向终端用户的 API 层，用户使用 YAML 格式配置数据同步流水线，使用 Flink CDC CLI 提交任务
Flink CDC Connect：对接外部系统的连接器层，通过对 Flink 与现有 Flink CDC source 进行封装实现对外部系统同步数据的读取和写入
Flink CDC Composer：同步任务的构建层，将用户的同步任务翻译为 Flink DataStream 作业
Flink CDC Runtime：运行时层，根据数据同步场景高度定制 Flink 算子，实现 schema 变更、路由、变换等高级功能

四、Java使用Flink CDC同步mysql数据

4.1 环境准备

根据flink cdc的版本不同，在java或springboot集成的时候选择的jdk版本也不尽相同，需要注意这一点，否则会在运行过程中出现奇怪的问题。

4.1.1 组件版本说明

本文演示中使用的相关组件版本如下：

jdk17；
springboot版本，2.5.5；
mysql，8.0.23；
Flink CDC 相关核心依赖版本，1.13.0；

4.1.2 数据库准备

案例要演示的需求场景为：使用Flink CDC 监听某个表的数据，然后同步到另一张表中，因此需要提前准备两张表，源表和目标表，为了简单起见，两个表除了表名不一样，其他的均相同，建表sql如下：

CREATE TABLE `tb_role` (
  `id` varchar(32) NOT NULL COMMENT '主键',
  `role_code` varchar(32) NOT NULL COMMENT '版本号',
  `role_name` varchar(32) DEFAULT NULL COMMENT '角色名称',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COMMENT='tb角色表';

4.1.3 导入相关的依赖

主要是flink cdc相关的依赖包

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.13.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.13.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_2.12</artifactId>
        <version>1.13.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>1.13.0</version>
    </dependency>

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

补充说明，单独使用java运行时，还需要添加下面两个日志组件依赖

<dependency>
     <groupId>org.slf4j</groupId>
     <artifactId>slf4j-log4j12</artifactId>
     <version>1.7.30</version>
 </dependency>

 <dependency>
     <groupId>org.apache.logging.log4j</groupId>
     <artifactId>log4j-to-slf4j</artifactId>
     <version>2.14.0</version>
 </dependency>

4.2 使用Flink CDC动态监听mysql数据变化

需求场景：通过Flink CDC ，监听tb_role表数据变化，写入tb_role_copy

4.2.1 自定义反序列化器

反序列化器的目的是为了解析flink cdc监听到mysql表数据变化的日志，以json的形式进行解析，方便对日志中的关键参数进行处理

package com.congge.flink.blog;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
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.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

public class CustomerSchema implements DebeziumDeserializationSchema<String> {

    /**
     * 封装的数据格式
     * {
     * "database":"",
     * "tableName":"",
     * "before":{"id":"","tm_name":""....},
     * "after":{"id":"","tm_name":""....},
     * "type":"c u d"
     * //"ts":156456135615
     * }
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //1.创建JSON对象用于存储最终数据
        JSONObject result = new JSONObject();
        //2.获取库名&表名
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        String database = fields[1];
        String tableName = fields[2];

        Struct value = (Struct) sourceRecord.value();
        //3.获取"before"数据
        Struct before = value.getStruct("before");
        JSONObject beforeJson = new JSONObject();
        if (before != null) {
            Schema beforeSchema = before.schema();
            List<Field> beforeFields = beforeSchema.fields();
            for (Field field : beforeFields) {
                Object beforeValue = before.get(field);
                beforeJson.put(field.name(), beforeValue);
            }
        }

        //4.获取"after"数据
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            Schema afterSchema = after.schema();
            List<Field> afterFields = afterSchema.fields();
            for (Field field : afterFields) {
                Object afterValue = after.get(field);
                afterJson.put(field.name(), afterValue);
            }
        }
        //5.获取操作类型  CREATE UPDATE DELETE
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
        if ("create".equals(type)) {
            type = "insert";
        }
        //6.将字段写入JSON对象
        result.put("database", database);
        result.put("tableName", tableName);
        result.put("before", beforeJson);
        result.put("after", afterJson);
        result.put("type", type);
        //7.输出数据
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}

4.2.2 自定义Sink输出

Sink即为Flink CDC的输出连接器，即监听到源表数据变化并经过处理后最终写到哪里，以mysql为例，我们在监听到tb_role表数据变化后，同步到tb_role_copy中去

package com.congge.flink.blog;

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;

public class MyJdbcSink extends RichSinkFunction<String> {

    // 提前声明连接和预编译语句
    private Connection connection = null;
    private PreparedStatement insertStmt = null;
    private PreparedStatement updateStmt = null;
    private PreparedStatement preparedStatement = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        if (connection == null) {
            Class.forName("com.mysql.cj.jdbc.Driver");//加载数据库驱动
            connection = DriverManager.getConnection("jdbc:mysql://IP:3306/db", "root", "root");
            connection.setAutoCommit(false);//关闭自动提交
        }
//        connection = DriverManager.getConnection("jdbc:mysql://IP:3306/db", "root", "root");
    }

    // 每来一条数据，调用连接，执行sql
    @Override
    public void invoke(String value, Context context) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(value);
        String type = jsonObject.getString("type");
        String tableName = "tb_role_copy";
        String database = jsonObject.getString("database");
        if(type.equals("insert")){
            JSONObject after = (JSONObject)jsonObject.get("after");
            Integer id = after.getInteger("id");
            String roleCode = after.getString("role_code");
            String roleName = after.getString("role_name");
            String sql = String.format("insert into %s.%s values (?,?,?)", database, tableName);
            insertStmt = connection.prepareStatement(sql);
            insertStmt.setInt(1, Integer.valueOf(id));
            insertStmt.setString(2, roleCode);
            insertStmt.setString(3, roleName);
            insertStmt.execute();
            connection.commit();
        } else if(type.equals("update")){
            JSONObject after = jsonObject.getJSONObject("after");
            Integer id = after.getInteger("id");
            String roleCode = after.getString("role_code");
            String roleName = after.getString("role_name");
            String sql = String.format("update %s.%s set role_code = ?, role_name = ? where id = ?", database, tableName);
            updateStmt = connection.prepareStatement(sql);
            updateStmt.setString(1, roleCode);
            updateStmt.setString(2, roleName);
            updateStmt.setInt(3, id);
            updateStmt.execute();
            connection.commit();
        } else if(type.equals("delete")){
            JSONObject after = jsonObject.getJSONObject("before");
            Integer id = after.getInteger("id");
            String sql = String.format("delete from %s.%s where id = ?", database, tableName);
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.setInt(1, id);
            preparedStatement.execute();
            connection.commit();
        }
    }

    @Override
    public void close() throws Exception {
        if(insertStmt != null){
            insertStmt.close();
        }
        if(updateStmt != null){
            updateStmt.close();
        }
        if(preparedStatement != null){
            preparedStatement.close();
        }
        if(connection != null){
            connection.close();
        }
    }
}

4.2.3 启动任务类

本例先以main程序运行，在实际进行线上部署使用时，可以打成jar包或整合springboot进行启动即可

package com.congge.flink.blog;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDC_CustomerSchema {

    public static void main(String[] args) throws Exception {
        //1、获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(3000);

        //2、通过FlinkCDC构建SourceFunction并读取数据
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("IP")
                .port(3306)
                .username("root")
                .password("3306")
                .databaseList("db")
                .tableList("db.tb_role")//若不添加该数据，则消费指定数据库中所有表的数据，如果指定，指定方式为db.table
                .deserializer(new CustomerSchema())
                //.deserializer(new StringDebeziumDeserializationSchema())
                //.startupOptions(StartupOptions.initial())
                .startupOptions(StartupOptions.latest())
                .build();
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);

        //3、打印数据
        streamSource.print();

        streamSource.addSink(new MyJdbcSink());
        //4、启动任务
        env.execute("FlinkCDC_CustomerSchema");
    }
}

4.2.4 效果测试

运行上面的代码，通过控制台可以看到任务已经运行起来了，监听并等待数据源数据变更

测试之前，确保两张表数据是一致的

此时为tb_role表增加一条数据，很快控制台可以监听并输出相关的日志

而后，tb_role_copy表同步新增了一条数据

4.3 与springboot整合实现过程

4.3.1 补充依赖

Flink cdc的依赖保持不变，需要单独引入logback的依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-logging</artifactId>
</dependency>

同时，在工程的resources目录下添加logback-spring.xml文件

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <!-- 定义日志的根级别和输出方式 -->
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- 日志文件保留策略 -->
    <property name="FILE_LOG_POLICY" value="10 MB" />

    <!-- 日志文件最大历史数 -->
    <property name="MAX_HISTORY" value="30" />

    <logger name="org.apache.flink.connector.mysql.cdc" level="error" />

    <!-- 应用程序的根日志级别 -->
    <root level="info">
        <appender-ref ref="STDOUT" />
    </root>
</configuration>

4.3.2 启动类改造

可以直接基于启动类改造，也可以新增一个类，实现ApplicationRunner接口，重写里面的run方法

不难发现，run方法里面的代码逻辑即是从上述main方法运行任务里面拷贝过来的；

package com.congge;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.congge.flink.blog.CustomerSchema;
import com.congge.flink.blog.MyJdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.servlet.ServletComponentScan;

@SpringBootApplication
@MapperScan(basePackages = {"com.congge.dao"})
@ServletComponentScan("com.congge.filter")
public class SeekOrderApp implements ApplicationRunner {

    public static void main(String[] args) {
        SpringApplication.run(SeekOrderApp.class,args);
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        //1、获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //env.enableCheckpointing(3000);

        //2、通过FlinkCDC构建SourceFunction并读取数据
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("192.168.1.196")
                .port(13306)
                .username("root")
                .password("Ghtms@123")
                .databaseList("erp_cloud")
                .tableList("erp_cloud.tb_role")//若不添加该数据，则消费指定数据库中所有表的数据，如果指定，指定方式为db.table
                .deserializer(new CustomerSchema())
                //.startupOptions(StartupOptions.initial())
                .startupOptions(StartupOptions.latest())
                .build();
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);

        //3、打印数据
        //streamSource.print();

        streamSource.addSink(new MyJdbcSink());
        //4、启动任务
        env.execute("FlinkCDC_CustomerSchema");
    }
}

4.3.3 效果测试

启动工程之后，相当于是通过flink cdc启动了一个用于监听数据变更的后台进程

然后我们再在数据库tb_role表增加一条数据，控制台可以看到输出了相关的日志

此时再检查数据表，可以发现tb_role_copy表新增了一条一样的数据

五、写在文末

本文通过较大的篇幅详细介绍了Flink CDC相关的技术，最后通过一个实际案例演示了使用Flink CDC同步mysql表数据的示例，希望对看到的同学有用，本篇到此结束感谢观看。

标签： flink mysql 大数据

本文转载自: https://blog.csdn.net/zhangcongyi420/article/details/142730700
版权归原作者 小码农叔叔 所有，如有侵权，请联系我们删除。