flinkCDC+iceberg(hadoop catalog) 同步mysql数据库数据

flink可以实现hadoop catalog 和hive catalog 。

前者映射hdfs地址，实现数据存储

后者映射hive表地址，用于有关hive的数据处理及其他项目实操。

1. 环境准备

官网：https://iceberg.apache.org/

1. hadoop 一定要 分布式集群

2.需要flink-connector-mysql-cdc-1.4.0.jar 即mysql-cdc连接器放到flink 的 lib下，为了连接数据库

3. 需要将iceberg 0.13.1的jar包放在${FLINK_HOME}/lib下

4. 开启mysql的binlog日志，mysql需要开启binlog日志,需要修改/etc/my.cnf文件，加上如下两段代码。

server-id=1
log-bin=mysql-bin 

本文版本：版本 flink 1.13.6+iceberg 0.13.1

2.启动flink环境

#启动flink环境
${FLINK_HOME}/bin/start-cluster.sh


#启动flinksql
${FLINK_HOME}/bin/sql-client.sh

3.放纵时刻

#流操作
SET execution.runtime-mode=streaming;

##手动设置checkpoint时间 
set execution.checkpointing.interval=5sec;

####数据库连接
CREATE TABLE mysql_binlog(
    name STRING ,
    cust_id INT,
    age INT,
    PRIMARY KEY (cust_id) NOT ENFORCED
)WITH (
'connector' = 'mysql-cdc',
'hostname' = 'hadoop01',
'port' = '3306',
'username' = 'root',
'password' = 'pwd',
'server-time-zone' = 'Asia/Shanghai',
'debezium.snapshot.mode'='initial',
'database-name' = 'FlinkCDC',
'table-name' = 'person'
);


#检查数据库连接数据是否正常
select * from mysql_binlog;


#创建iceberg表
CREATE TABLE sample(
    name STRING ,
    cust_id INT,
    age INT,
    PRIMARY KEY (cust_id) NOT ENFORCED
)WITH (
'connector'='iceberg',
'catalog-name'='iceberg_catalog',  --catalog名称
'catalog-type'='hadoop',  --创建的为hadoop的catalog
'warehouse'='hdfs://hadoop01:8020/iceberg/test2',  --指定数据位置
'format-version'='2'  --版本 2 支持带主键的表upsert但是仍然无法流式查询。版本 1 不支持带主键的表upsert
);

##导入数据到iceberg
insert into sample select * from mysql_binlog;

##检查数据是否导入成功
select * from sample;

丰富知识：
iceberg使用代码合并小文件后如果不开启此配置，历史文件同样不会删除，开启后就会实现合并后清除历史文件
    启用提交后写入元数据删除
    write.metadata.delete-after-commit.enabled=true
    配置保留历史数量(比如配置为5，则元数据和数据都保留5份历史数据和1份最新数据)
    write.metadata.previous-versions-max=5    

历史文件保留最大值为5，metadata和data里面文件数则始终保持为6个