FLink-Canal

浩客2年前技术文章1114

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

相关文章

数仓主流架构简介之三

数仓主流架构简介之三

一、数仓架构经历过程随着数据量的暴增和数据实时性要求越来越高,以及大数据技术的发展驱动企业不断升级迭代,数据仓库架构方面也在不断演进,分别经历了以下过程:早期经典数仓架构 > 离线大数据架构 &...

数据湖技术之iceberg(八)Spark与Iceberg整合DDL操作

数据湖技术之iceberg(八)Spark与Iceberg整合DDL操作

1.CREATE TABLE 创建表Create table 创建Iceberg表,创建表不仅可以创建普通表还可以创建分区表,再向分区表中插入一批数据时,必须对数据中分区列进行排序,否则会出现文件关闭...

EMR-java配置国密SM4加密

EMR-java配置国密SM4加密

首先找到bcprov-jdk15on-1.56.jar这个包<dependency>    <groupId>org.bouncycastle</groupId> ...

grafana版本升级

grafana版本升级

      因Grafana需接入腾讯云监控数据,安装腾讯云监控插件。腾讯云监控应用插件需运行在 Grafana ≥ 7.3且 < 8.0 的版本上。当前使用版本为...

触发kube-apiserver流控导致ACK集群中节点状态异常

触发kube-apiserver流控导致ACK集群中节点状态异常

问题现象ack集群中一个节点状态为notready,但是排查节点上的核心组件:kubelet、kube-proxy、flannel、etcd等状态均为正常。排查步骤1、查日志分别排查核心组件以及系统日...

单节点Kafka部署并开启Kerberos配置

安装Kerberosserver节点安装kerberos相关软件yum install -y krb5-server krb5-workstation krb5-libs修改配置文件krb5.conf...

发表评论    

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