FLink-Canal

浩客4个月前技术文章178

canal format

Canal 是一个 CDC(ChangeLog Data Capture,变更日志数据捕获)工具,可以实时地将 MySQL 变更传输到其他系统。Canal 为变更日志提供了统一的数据格式,并支持使用 JSON 或 protobuf 序列化消息(Canal 默认使用 protobuf)。

Flink 支持将 Canal 的 JSON 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中。在很多情况下,利用这个特性非常的有用,例如

  • 将增量数据从数据库同步到其他系统

  • 日志审计

  • 数据库的实时物化视图

  • 关联维度数据库的变更历史,等等。

Flink 还支持将 Flink SQL 中的 INSERT / UPDATE / DELETE 消息编码为 Canal 格式的 JSON 消息,输出到 Kafka 等存储中。 但需要注意的是,目前 Flink 还不支持将 UPDATE_BEFORE 和 UPDATE_AFTER 合并为一条 UPDATE 消息。因此,Flink 将 UPDATE_BEFORE 和 UPDATE_AFTER 分别编码为 DELETE 和 INSERT 类型的 Canal 消息。

示例:

1、在mysql中创建学生表,插入几条数据

CREATE TABLE `student` (
 `id` varchar(20) NOT NULL,
 `name` varchar(255) DEFAULT NULL,
 `age` bigint(20) DEFAULT NULL,
 `gender` varchar(255) DEFAULT NULL,
 `clazz` varchar(255) DEFAULT NULL,
 PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

2、在flink中创建kafka soure表指定数据的格式为canl-json

canal采集的数据包含三种类型,INSERT, UPDATE,DELETE

FLink 会自动将三种类型转换成变更日志流。同时会自动解析数据

CREATE TABLE student_kafka (
 id STRING,
 name STRING,
 age bigint,
 gender STRING,
 clazz STRING
) WITH (
'connector' = 'kafka',
'topic' = 'bigdata.student',
'properties.bootstrap.servers' = 'master:9092',
'properties.group.id' = 'testGroup',
'scan.startup.mode' = 'earliest-offset',
'format' = 'canal-json'  -- 使用 canal-json 格式
);


3、统计班级的人数

select clazz,count(1)  as c 
from student_kafka
group by clazz

4、将统计的结果保存到数据库中

CREATE TABLE clazz_num (
 clazz STRING,
 c BIGINT,
 PRIMARY KEY (clazz) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8',
  'table-name' = 'clazz_num',
  'username' ='root',
  'password' = '123456'
);
insert into clazz_num
select clazz,count(1)  as c
from student_kafka
group by clazz

相关文章

Python 序列化与反序列化

1、为什么要序列化内存中的字典、列表、集合以及各种对象,如何保存到一个文件中?如果是自己定义的类的实例,如何保存到一个文件中?如何从文件中读取数据,并让它们在内存中再次恢复成自己对应的类的实例?要设计...

数据湖技术之iceberg(五)Hive与Iceberg整合

数据湖技术之iceberg(五)Hive与Iceberg整合

1.  版本支持约束条件Iceberg就是一种表格式,支持使用Hive对Iceberg进行读写操作,但是对Hive的版本有要求,如下:Iceberg 与 Hive 2.x 和 Hive 3....

大数据集群部署规划(三)节点选配方案

节点部署原则适用场景组网规则管理节点、控制节点和数据节点分开部署(此方案至少需要8个节点,manager为部署商业化大数据集群时所需例如:hdp,cdh等)core × 11 + worker × n...

开源大数据集群部署(六)Keytab文件生成

开源大数据集群部署(六)Keytab文件生成

1、 创建keytab文件除了使用明文密码登录之外,Kerberos还可以使用keytab密码文件登陆,现在为testcuser创建它的keytab文件ipa-getkeytab -s ipa.hdp...

CDH实操--HDFS高可用设置

CDH实操--HDFS高可用设置

1 概述        在HDFS集群中NameNode存在单点故障(SPOF),对于只有一个NameNode的集群,如果NameNode机...

Trino配置yanagishima-23.0(包含编译)

Trino配置yanagishima-23.0(包含编译)

1 环境介绍1.1 本文采用trino 359yanagishima v23.02 编译yanagishima2.1 安装编译yanagishima需要的工具安装编译yanagishima需要的工具w...

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。