Flink CDC 原理及生产实践

Stella981
• 阅读 832

点击上方蓝色字体,选择“设为星标

回复”资源“获取更多资源

Flink CDC 原理及生产实践

Flink CDC 原理及生产实践

大数据技术与架构

点击右侧关注,大数据开发领域最强公众号!

Flink CDC 原理及生产实践

Flink CDC 原理及生产实践

大数据真好玩

点击右侧关注,大数据真好玩!

Flink CDC 原理及生产实践

Flink CDC 原理及生产实践 MySQL CDC连接器允许从MySQL数据库读取快照数据和增量数据。本文档根据官网翻译了如何设置MySQL CDC连接器以对MySQL数据库运行SQL查询。

依赖关系

为了设置MySQL CDC连接器,下表提供了使用构建自动化工具(例如Maven或SBT)和带有SQL JAR捆绑包的SQL Client的两个项目的依赖项信息。

1、Maven依赖

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

2、SQL客户端JAR

 下载flink-sql-connector-mysql-cdc-1.1.0.jar并将其放在下 <FLINK_HOME> /lib/。

设置MySQL服务器

您必须定义一个对Debezium MySQL连接器监视的所有数据库具有适当权限的MySQL用户。

1、创建MySQL用户

mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';

2、向用户授予所需的权限

mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';

3、最终确定用户的权限

mysql> FLUSH PRIVILEGES;

注意

1、MySQL CDC源代码如何工作

启动MySQL CDC源时,它将获取一个全局读取锁(FLUSH TABLES WITH READ LOCK),该锁将阻止其他数据库的写入。然后,它读取当前binlog位置以及数据库和表的schema。之后,将释放 全局读取锁。然后,它扫描数据库表并从先前记录的位置读取binlog。Flink将定期执行checkpoints以记录binlog位置。如果发生故障,作业将重新启动并从checkpoint完成的binlog位置恢复。因此,它保证了仅一次的语义。

2、向MySQL用户授予RELOAD权限

如果未授予MySQL用户RELOAD权限,则MySQL CDC源将改为使用表级锁,并使用此方法执行快照。这会阻止写入更长的时间。

3、全局读取锁(FLUSH TABLES WITH READ LOCK)

全局读取锁 在读取binlog位置和schema期间保持。这可能需要几秒钟,具体取决于表的数量。全局读取锁定会阻止写入,因此它仍然可能影响在线业务。如果要跳过读取锁,并且可以容忍至少一次语义,则可以添加'debezium.snapshot.locking.mode' = 'none'选项以跳过锁。

4、为每个作业设置一个differnet SERVER ID 每个用于读取binlog的MySQL数据库客户端都应具有唯一的ID,称为server id。MySQL服务器将使用此ID维护网络连接和binlog位置。如果不同的作业共享相同的server id,则可能导致从错误的binlog位置进行读取。提示:默认情况下,启动TaskManager时,server id是随机的。如果TaskManager失败,则再次启动时,它可能具有不同的server id。但这不应该经常发生(作业异常不会重新启动TaskManager),也不会对MySQL服务器造成太大影响。

因此,建议为每个作业设置不同的server id ,例如:

通过SQL Hints:SELECT * FROM source_table /+ OPTIONS('server-id'='123456') / ;

通过Stream ApI的 创建source时设置:MySQLSource.builder().xxxxxx.serverId(123456);

重点:Mysq的binlog 可以说是针对库级别,所以相同的server id去拉一个库里的不同表或者相同表可能会造成数据丢失。所以建议设置server id。

5、扫描数据库表期间无法执行检查点

在扫描表期间,由于没有可恢复的位置,因此我们无法执行checkpoints。为了不执行检查点,MySQL CDC源将保持检查点等待超时。超时检查点将被识别为失败的检查点,默认情况下,这将触发Flink作业的故障转移。因此,如果数据库表很大,则建议添加以下Flink配置,以避免由于超时检查点而导致故障转移:

execution.checkpointing.interval: 10minexecution.checkpointing.tolerable-failed-checkpoints: 100restart-strategy: fixed-delayrestart-strategy.fixed-delay.attempts: 2147483647

6、设置MySQL会话超时

为大型数据库创建初始一致的快照时,在读取表时,您建立的连接可能会超时。您可以通过在MySQL配置文件中配置Interactive_timeout和wait_timeout来防止此行为。

  • interactive_timeout:服务器在关闭交互式连接之前等待活动的秒数。

  • wait_timeout:服务器在关闭非交互式连接之前等待其活动的秒数。

如何创建MySQL CDC表

1、Sql的方式:(1)定义表如下:

-- register a MySQL table 'orders' in Flink SQLCREATE TABLE orders (  order_id INT,  order_date TIMESTAMP(0),  customer_name STRING,  price DECIMAL(10, 5),  product_id INT,  order_status BOOLEAN) WITH (  'connector' = 'mysql-cdc',  'hostname' = 'localhost',  'port' = '3306',  'username' = 'root',  'password' = '123456',  'database-name' = 'mydb',  'table-name' = 'orders');-- read snapshot and binlogs from orders tableSELECT * FROM orders;

2、Stream API

MySQL CDC连接器也可以是DataStream源。您可以创建SourceFunction,如下所示:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.source.SourceFunction;import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;public class MySqlBinlogSourceExample {  public static void main(String[] args) throws Exception {    SourceFunction<String> sourceFunction = MySQLSource.<String>builder()      .hostname("localhost")      .port(3306)      .databaseList("inventory") // monitor all tables under inventory database      .username("flinkuser")      .password("flinkpw")      .deserializer(new StringDebeziumDeserializationSchema()) // converts SourceRecord to String      .build();    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();    env      .addSource(sourceFunction)      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering    env.execute();  }}

特征

1、Exactly-Once Processing 一次处理 MySQL CDC连接器是Flink Source连接器,它将首先读取数据库快照,然后即使发生故障,也将以完全一次的处理继续读取二进制日志。请阅读连接器如何执行数据库快照。

2、Single Thread Reading 单线程阅读 MySQL CDC源无法并行读取,因为只有一个任务可以接收Binlog事件。

常见问题

1、如何跳过快照并仅从binlog中读取?可以通过选项进行控制debezium.snapshot.mode,您可以将其设置为:

  • never:指定连接永远不要使用快照,并且在第一次使用逻辑服务器名称启动时,连接器应该从binlog的开头读取;请谨慎使用,因为只有在binlog保证包含数据库的整个历史记录时才有效。

  • schema_only:如果自连接器启动以来不需要数据的连续快照,而只需要它们进行更改,则可以使用该schema_only选项,其中连接器仅对模式(而不是数据)进行快照。

2、如何读取包含多个表(例如user_00,user_01,...,user99)的共享数据库?该table-name选项支持正则表达式以监视多个与正则表达式匹配的表。因此,您可以设置table-name为user.*监视所有user_前缀表。database-name选项相同。请注意,共享表应该在相同的架构中。

3、ConnectException:收到用于处理的DML'...',binlog可能包含使用语句或基于混合的复制格式生成的事件 如果有上述异常,请检查是否binlog_format为ROW,您可以通过show variables like '%binlog_format%'在MySQL客户端中运行来进行检查。请注意,即使binlog_format您的数据库配置为ROW,也可以通过其他会话更改此配置,例如SET SESSION binlog_format='MIXED'; SET SESSION tx_isolation='REPEATABLE-READ'; COMMIT;。还请确保没有其他会话正在更改此配置

实践中遇到的问题

1、不同的kafka版本依赖冲突会造成cdc报错:http://apache-flink.147419.n8.nabble.com/cdc-td8357.html#a8393

Flink CDC 原理及生产实践

2、超时问题:根据上面提到设置wait_timeout解决。

Flink CDC 原理及生产实践

Flink CDC 原理及生产实践

Flink CDC 原理及生产实践

版权声明:

本文为大数据技术与架构整理,原作者独家授权。未经原作者允许转载追究侵权责任。

微信公众号|import_bigdata

编辑 《大数据技术与架构》

插画 《大数据技术与架构》

文章链接 https://www.jianshu.com/p/439b1d1247b2

欢迎点赞+收藏+转发朋友圈素质三连

Flink CDC 原理及生产实践

文章不错?点个【在看】吧!** 👇**

本文分享自微信公众号 - 大数据技术与架构(import_bigdata)。
如有侵权,请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
待兔 待兔
5个月前
手写Java HashMap源码
HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Stella981 Stella981
3年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Wesley13 Wesley13
3年前
mysql设置时区
mysql设置时区mysql\_query("SETtime\_zone'8:00'")ordie('时区设置失败,请联系管理员!');中国在东8区所以加8方法二:selectcount(user\_id)asdevice,CONVERT\_TZ(FROM\_UNIXTIME(reg\_time),'08:00','0
Wesley13 Wesley13
3年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Stella981 Stella981
3年前
Docker 部署SpringBoot项目不香吗?
  公众号改版后文章乱序推荐,希望你可以点击上方“Java进阶架构师”,点击右上角,将我们设为★“星标”!这样才不会错过每日进阶架构文章呀。  !(http://dingyue.ws.126.net/2020/0920/b00fbfc7j00qgy5xy002kd200qo00hsg00it00cj.jpg)  2
Stella981 Stella981
3年前
ClickHouse大数据领域企业级应用实践和探索总结
点击上方蓝色字体,选择“设为星标”回复”资源“获取更多资源!(https://oscimg.oschina.net/oscnet/bb00e5f54a164cb9827f1dbccdf87443.jpg)!(https://oscimg.oschina.net/oscnet/dc8da835ff1b4
Wesley13 Wesley13
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
10个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这