devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

什么是Flink CDC，以及如何使用_flinkcdc(1)，一文全懂

作者：Monodyee | 2024-04-23 08:47:38

赞

踩

什么是Flink CDC，以及如何使用_flinkcdc(1)，一文全懂

先自我介绍一下，小编浙江大学毕业，去过华为、字节跳动等大厂，目前阿里P7

深知大多数程序员，想要提升技能，往往是自己摸索成长，但自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年最新大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。
img
img
img
img
img

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！

由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新

如果你需要这些资料，可以添加V获取：vip204888 （备注大数据）
img

正文

开启mysql数据库bin-log日志
1.如果是服务器

在my.cnf中添加binlog配置,并重启mysql数据库



    server-id = 123
    log_bin = mysql-bin
    binlog_format = row
    binlog_row_image = full
    expire_logs_days = 10
    gtid_mode = on
    enforce_gtid_consistency = on
   


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

已经为 MySQL 设置了一些配置参数。下面是对这些参数的解释：

  1. server-id = 123：指定服务器的唯一标识符，通常用于区分不同的数据库服务器。
  2. log_bin = mysql-bin：启用二进制日志记录，以便在数据库出现故障时可以恢复数据。
  3. binlog_format = row：指定二进制日志的记录格式。row 格式会记录每个更改行的详细信息，这对于需要事务完整性的应用程序非常有用。
  4. binlog_row_image = full：设置 row 格式的二进制日志记录行的完整信息，包括列值、注释等。这有助于提高应用程序的可恢复性。
  5. expire_logs_days = 10：设置自动清理过期二进制日志文件的天数。在这个例子中，设置为 10 天。
  6. gtid_mode = on：启用全局事务 ID 模式，这使得基于 GTID 的复制成为可能。
  7. enforce_gtid_consistency = on：强制执行 GTID 一致性，确保事务在不同的 MySQL 实例之间保持一致。
2.如果在Windows使用小皮

在小皮面板里设置，如图：

在这里插入图片描述

打开bin日志开关

搭建Flink CDC java环境

添加maven相关pom

在pom里添加相关Flink CDC依赖

<!-- flink connector 基础包-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>1.14.4</version>
        </dependency>
<!-- CDC mysql 源-->
        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-sql-connector-mysql-cdc</artifactId>
            <version>2.3.0</version>
        </dependency>
<!-- Flink Steam流处理-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.14.4</version>
        </dependency>
<!-- flink java客户端-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.14.4</version>
        </dependency>
<!-- 开启webui支持，默认是8081，默认没有开启-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web_2.12</artifactId>
            <version>1.14.4</version>
        </dependency>
<!-- Flink Table API和SQL API使得在Flink中进行数据处理变得更加简单和高效
通过使用Table API和SQL API，可以像使用传统的关系型数据库一样，通过编写SQL语句或者使用类似于
Java的API进行数据处理和分析-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime_2.11</artifactId>
            <version>1.14.4</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.11</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>2.0.6</version>
        </dependency>


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49

这是一段 Maven 依赖配置，它引入了 Flink Connector Base、CDC MySQL Source、Flink Streaming Java、Flink Java Client、Flink Runtime Web、Flink Table Runtime 和 Logback Classic。

这些依赖库提供了以下功能：

  • Flink Connector Base:Flink 的连接器基础包，用于将 Flink 与其他系统进行集成。
  • CDC MySQL Source:Flink 的 MySQL CDC 源，用于从 MySQL 数据库中读取数据流。
  • Flink Streaming Java:Flink 的 Java 流处理 API,用于编写并发程序以处理数据流。
  • Flink Java Client:Flink 的 Java API,用于在 Java 应用程序中使用 Flink。
  • Flink Runtime Web:Flink 的 Web UI,用于监控和管理 Flink 集群。
  • Flink Table Runtime:Flink 的 Table API,使在 Flink 中进行数据处理变得更加简单和高效。
  • Logback Classic:日志记录库，用于记录应用程序的日志信息。
构建Sink

Flink CDC(Change Data Capture)中的Sink用于将CDC接收到的数据写入外部系统(如数据库或文件系统),以实现数据同步和数据备份等功能，并将其转换为DataStream流。然后，Sink将这个DataStream流写入到外部系统中，以便进行后续的数据处理和分析。

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
public class CustomSink extends RichSinkFunction<String> {

    @Override
    public void invoke(String value, Context context) throws Exception {
        System.out.println("json->: "+value);
    }
}



  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

这段代码定义了一个名为CustomSink的类，它继承自RichSinkFunction类。RichSinkFunction是Flink CDC中用于将数据写入外部系统(如数据库或文件系统)的函数接口。CustomSink的作用是将CDC接收到的数据写入外部系统中。具体实现方式由子类CustomSink来定义。由于这个类继承了RichSinkFunction,因此可以使用Flink中的其他RichSink函数特性，例如设置日志级别、配置连接等,invoke则是处理函数。

main配置运行

如下面的代码，构建Flink CDC连接

  public static void main(String[] args) throws Exception {
        MySqlSourceBuilder<String> builder = MySqlSource.builder();
        MySqlSource<String> source = builder.hostname("192.168.2.6")
                .port(3306)
                .databaseList("mydb")
                .tableList("mydb.user")
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .includeSchemaChanges(true)
                .build();
// 启动webui，绑定本地web-ui端口号
        Configuration configuration=new Configuration();
        configuration.setInteger(RestOptions.PORT,8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

        env.enableCheckpointing(5000);
        env.fromSource(source, WatermarkStrategy.noWatermarks(),"MYSQL Source")
                .addSink(new CustomSink());
        env.execute();
    }


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

这段代码是使用Flink构建一个数据流处理任务，从MySQL数据库中读取数据并进行处理。

首先，使用MySqlSourceBuilder创建一个MySqlSource对象，并设置连接参数(hostname、port、databaseList、tableList、username和password)以及反序列化器(JsonDebeziumDeserializationSchema)。然后，创建Configuration对象并设置WebUI端口号(RestOptions.PORT),接着使用StreamExecutionEnvironment创建一个执行环境，启用检查点(checkpointing)并将MySqlSource和自定义的Sink添加到执行环境中。最后，执行整个任务。

操作数据库查看结果

如图所示：

{
  "before": null,
  "after": {
    "id": "1661935564737286146",
    "qu\_type": 4,
    "level": 1,
    "image": "",
    "content": "dos查看日期、时间",
    "create\_time": 1685100091000,
    "update\_time": 1685100091000,
    "remark": "",
    "analysis": ""
  },
  "source": {
    "version": "1.6.4.Final",
    "connector": "mysql",
    "name": "mysql\_binlog\_source",
    "ts\_ms": 0,
    "snapshot": "false",
    "db": "mydb",
    "sequence": null,
    "table": "t\_user",
    "server\_id": 0,
    "gtid": null,
    "file": "",


**网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。**

**需要这份系统化的资料的朋友，可以添加V获取：vip204888 （备注大数据）**
![img](https://img-blog.csdnimg.cn/img_convert/11b15b9a5e9ee848de1ccae2f10038ae.png)

**一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**

大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。**

**需要这份系统化的资料的朋友，可以添加V获取：vip204888 （备注大数据）**
[外链图片转存中...(img-AgbMrrP4-1713171905829)]

**一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/473070
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号