CDC概念

CDC全称是Change data Cpature，即变更数据捕获，主要面向数据库的变更，是数据库领域非常常见的技术，主要用于捕获数据库的一些变更，然后可以把变更数据发送到下游。

CDC类型

1.基于查询的，客户端会通过SQL方式查询源库表变更数据，然后对外发送。

2.基于日志的，这也是业界广泛使用的一种方式，一般是通过binlog方式，变更的记录会写入binlog，解析binlog后会写入消息系统，或直接基于Flink CDC进行处理。

CDC数据入湖

基于CDC数据的入湖，架构：上游各种各样的数据源，比如DB的变更数据、事件流，以及各种外部数据源，都可以通过变更流的方式写入表中，再进行外部的查询分析

典型CDC入湖的链路：

链路1是大部分公司采取的链路，前面CDC的数据先通过CDC工具导入kafka或者Pulsar，再通过Flink或者是spark流式消费写到Hudi里

链路2是通过Flink CDC直联到MySQL上游数据源，直接写到下游hudi表。

Flink CDC Hudi概述

基于Flink CDC技术，实时采集MySQL数据库表数据，进行ETL转换处理，最终存储Hudi表

实践

MySQL数据库创建表，实时添加数据，通过Flink CDC将数据写入Hudi表，并且Hudi与Hive集成，自动在hive中创建表与添加分区信息，最后hive终端beeline查询分析数据。

hudi表与hive表自动关联集成，需要重新编译hudi源码，指定hive版本及编译时包含hive依赖jar包

1.MySQL数据库，创建表及开启binlog

2.创建flink CDC表，关联到MySQL数据库表

3.创建视图，数据来源输入表，字段与输出表相同

4.创建输出表，关联到hudi表，自动同步到hive中，字段与hudi表相同

5.查询视图数据，插入到输出表(hudi表)

6.查询hive表数据，ro类型(读优化查询)和rt类型(快照查询)

准备工作

1.编译hudi源码

修改hudi集成flink和hive编译依赖版本配置

原因：现在版本hudi，在编译的时候后本身默认已经集成了flink-SQL-connector-hive的包，会和flink lib包下的flink-SQL-connector-hive冲突。所以，编译的过程中只修改hive编译版本

文件: hudi-0.10.1/packaging/hudi-flink-bundle/pom.xml，将hive.version改为2.3.9

<include>org.apache.flink:flink-sql-connector-hive-2.3.9_${scala.binary.version}</include>

该问题从 0.10 版本已经解决

mvn clean install -DskipTests -Drat.skip=true -Dscala-2.12 -Dspark3 -Dhive.version=2.3.9 -Pflink-bundle-shade-hive2

编译完成后，有2个jar包，非常重要

hudi-flink-bundle_2.12-0.10.1.jar位于hudi-0.10.1/packaging/hudi-flink-bundle/target，flink用来写入和读取数据，将其拷贝至$FLINK_HOME/lib目录中，如果以前有同名jar包，先删除再拷贝。

hudi-hadoop-mr-bundle-0.10.1.jar位于hudi-0.10.1/packaging/hudi-hadoop-mr-bundle/target，hive需要用来读hudi数据，将其拷贝至$HIVE_HOME/lib目录中。

2.将Flink CDC MySQL对应jar包，放到$FLINK_HOME/lib目录中

建议用新版的CDC2，因为功能差了好多，CDC1的时候是把数据先全都读取到内存，再执行后续操作，CDC2是边读边执行后续操作，这么一比，CDC1被秒杀。

CDC1：

https://mvnrepository.com/artifact/com.alibaba.ververica/flink-connector-mysql-cdc/1.4.0

flink-sql-connector-mysql-cdc-1.4.0.jar

CDC2[升级版]:

flink-connector-mysql-cdc-2.0.2.jar

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

实现

零、组件版本

hudi：0.10.1
flink：1.13.1
hive：2.3.9
Hadoop：2.8.0
Scala：2.12.4

一、MySQL配置

1.MySQL数据库，创建表及开启binlog

vim /etc/my.cnf

在[mysqld]下面添加内容

server-id=2
log-bin=mysql-bin
binlog_format=rot
expire_logs_day=15
binlog_row_image=full

2.重启MySQL

service mysqld restart

查看是否生效

show master logs

3.建表

create database test_hudi;
create table test_hudi.tbl_users(id bigint auto_increment primary key,name varchar(20) null,birthday timestamp default CURRENT_TIMESTAMP NOT NULL,ts timestamp default CURRENT_TIMESTAMP NOT NULL
);INSERT INTO test_hudi.tbl_users(name) VALUES(“测试”);
INSERT INTO test_hudi.tbl_users(name) VALUES(“张三”);
INSERT INTO test_hudi.tbl_users(name) VALUES(“李四”);
INSERT INTO test_hudi.tbl_users(name) VALUES(“王五”);
INSERT INTO test_hudi.tbl_users(name) VALUES(“赵六");

二、创建Flink CDC表,关联到MySQL数据库表

1.启动相关服务

HDFShadoop-daemon.sh start namenodehadoop-daemon.sh start datanode
Hive./hive --service metastore &./hive --service hiveserver2  &
Flink standalonestart-cluster.sh
启动Flink SQL Client客户端sql-client.sh embedded -j /Users/FengZhen/Desktop/Hadoop/flink/flink-1.13.1/lib/flink-connector-mysql-cdc-1.4.0.jar shell
-j /Users/FengZhen/Desktop/Hadoop/flink/flink-1.13.1/lib/hudi-hive-sync-bundle-0.10.1.jar设置属性set  sql-client.execution.result-mode=tableau;set execution.checkpointing.interval=3sec;

2.创建输入表，关联MySQL表，采用MySQL CDC关联

CREATE TABLE users_source_mysql(id BIGINT PRIMARY KEY NOT ENFORCED,name STRING,birthday TIMESTAMP(3),ts TIMESTAMP(3)
)WITH('connector' = 'mysql-cdc','hostname' = 'localhost','port' = '3306','username' = 'root','password' = '1234qwer','server-time-zone'= 'Asia/Shanghai','debezium.snapshot.mode' = 'initial','database-name' = 'test_hudi','table-name' = 'tbl_users'
);
Flink SQL> select * from users_source_mysql;
+----+----------------------+--------------------------------+-------------------------+-------------------------+
| op |                   id |                           name |                birthday |                      ts |
+----+----------------------+--------------------------------+-------------------------+-------------------------+
| +I |                    1 |                           测试 | 2022-03-15 22:09:00.000 | 2022-03-15 22:09:00.000 |

使用CDC2必须配置一下主键，因为参数【scan.incremental.snapshot.enabled】默认为true，增量读取就必须配置PK，如果不做增量读取，直接改为false即可。

debezium.snapshot.mode枚举值：

参数值	描述
initial(默认)	连接器执行数据库的初始一致性快照，快照完成后，连接器开始为后续数据库更改流式传输事件记录。
initial_only	连接器只执行数据库的初始一致性快照，不允许捕获任何后续更改的事件。
schema_only	连接器只捕获所有相关表的表结构，不捕获初始数据，但是会同步后续数据库的更改记录
schema_only_recovery	设置此选项可恢复丢失或损坏的数据库历史主题(database.history.kafka.topic)。

三、创建视图，查询输入表，字段与输出表相同

创建视图，增加分区列part，方便后续同步hive分区表

CREATE VIEW view_users_cdc AS
SELECT *, DATE_FORMAT(birthday, 'yyyyMMdd') AS part FROM users_source_mysql;
Flink SQL> select * From view_users_cdc;
+----+----------------------+--------------------------------+-------------------------+-------------------------+--------------------------------+
| op |                   id |                           name |                birthday |                      ts |                           part |
+----+----------------------+--------------------------------+-------------------------+-------------------------+--------------------------------+
| +I |                    1 |                           测试 | 2022-03-15 22:09:00.000 | 2022-03-15 22:09:00.000 |                       20220315 |
| +I |                    2 |                           张三 | 2022-03-17 13:26:05.000 | 2022-03-17 13:26:05.000 |                       20220317 |
| +I |                    3 |                           李四 | 2022-03-17 13:26:05.000 | 2022-03-17 13:26:05.000 |                       20220317 |
| +I |                    4 |                           王五 | 2022-03-17 13:26:05.000 | 2022-03-17 13:26:05.000 |                       20220317 |
| +I |                    5 |                           赵六 | 2022-03-17 13:26:05.000 | 2022-03-17 13:26:05.000 |                       20220317 |

四、创建CDC Hudi Sink 表，并自动同步hive分区

CREATE TABLE users_sink_hudi_hive(id bigint ,name string,birthday TIMESTAMP(3),ts TIMESTAMP(3),part VARCHAR(20),primary key(id) not enforced
)
PARTITIONED BY (part)
with('connector'='hudi','path'= 'hdfs://localhost:9000/hudi-warehouse/users_sink_hudi_hive', 'hoodie.datasource.write.recordkey.field'= 'id'-- 主键, 'write.precombine.field'= 'ts'-- 自动precombine的字段, 'write.tasks'= '1', 'compaction.tasks'= '1', 'write.rate.limit'= '2000'-- 限速, 'table.type'= 'MERGE_ON_READ'-- 默认COPY_ON_WRITE,可选MERGE_ON_READ, 'compaction.async.enabled'= 'true'-- 是否开启异步压缩, 'compaction.trigger.strategy'= 'num_commits'-- 按次数压缩, 'compaction.delta_commits'= '1'-- 默认为5, 'changelog.enabled'= 'true'-- 开启changelog变更, 'read.streaming.enabled'= 'true'-- 开启流读, 'read.streaming.check-interval'= '3'-- 检查间隔，默认60s, 'hive_sync.enable'= 'true'-- 开启自动同步hive, 'hive_sync.mode'= 'hms'-- 自动同步hive模式，默认jdbc模式, hms:hive metastore, 'hive_sync.metastore.uris'= 'thrift://localhost:9083'-- hive metastore地址-- , 'hive_sync.jdbc_url'= 'jdbc:hive2://localhost:10000'-- hiveServer地址, 'hive_sync.table'= 'users_sink_hudi_hive_sync'-- hive 新建表名, 'hive_sync.db'= 'db_hudi'-- hive 新建数据库名, 'hive_sync.username'= ''-- HMS 用户名, 'hive_sync.password'= ''-- HMS 密码, 'hive_sync.support_timestamp'= 'true'-- 兼容hive timestamp类型
);

五、视图数据写入hudi表

INSERT INTO users_sink_hudi_hive
SELECT id,name,birthday,ts,part FROM view_users_cdc;

六、Hive表查询

需要将hudi-hadoop-mr-bundle-0.10.1.jar包，放到$HIVE_HOME/lib下

启动beeline客户端，连接hiveserver2

beeline -u jdbc:hive2://localhost:10000 -n root -p 123456

我吐了，我的hive没生成下边这两张表，我各种操作都TM不行，这个情况先保留，以后再看

会自动生成hudi MOR模式的两张表

users_sink_hudi_hive_ro：ro表全称read optimized table，对于MOR表同步的xxx_ro表，只暴露压缩后的parquet。其查询方式和COW表类似。设置完hiveInputformat之后和普通的hive表一样查询即可。

users_sink_hudi_rt：rt表示增量视图，主要针对增量查询的rt表；ro表只能查parquet文件数据，rt表parquet文件数据和log文件数据都可查。

查看hive表数据

set hive.exec.mode.local.auto = true;-- 不添加的话会导致count数据和select * 数据量不一致
set hive.input.format = org.apache.hudi.hadoop.hive.HoodieCombineHiveInputFormat;
set hive.mapred.mode = nonstrict;