什么是CDC？
CDC是（Change Data Capture 变更数据获取）的简称。核心思想是，监测并捕获数据库的变动（包括数据 或 数据表的插入INSERT、更新UPDATE、删除DELETE等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。
Flink_CDC
1. 环境准备
- mysql
- Hive
- flink 1.13.5 on yarn
说明：如果没有安装hadoop，那么可以不用yarn，直接用flink standalone环境吧。
2. 下载下列依赖包
下面两个地址下载flink的依赖包，放在lib目录下面。
- 【flink-sql-connector-hive-2.2.0_2.11-1.13.5.jar】https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-hive-2.2.0_2.11/1.13.5/flink-sql-connector-hive-2.2.0_2.11-1.13.5.jar
如果你的Flink是其它版本，可以来【这里】https://repo.maven.apache.org/maven2/org/apache/flink下载。
说明：我hive版本是2.1.1，为啥这里我选择版本号是2.2.0呢，这是官方文档给出的版本对应关系：
Metastore version | Maven dependency | SQL Client JAR |
|---|---|---|
1.0.0 - 1.2.2 | flink-sql-connector-hive-1.2.2 | Download |
2.0.0 - 2.2.0 | flink-sql-connector-hive-2.2.0 | Download |
2.3.0 - 2.3.6 | flink-sql-connector-hive-2.3.6 | Download |
3.0.0 - 3.1.2 | flink-sql-connector-hive-3.1.2 | Download |
官方文档地址在：https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/table/hive/overview/，可以自行查看。
3. 启动flink-sql client
- 先在yarn上面启动一个application，进入flink13.5目录，执行：
bin/yarn-session.sh -d -s 2 -jm 1024 -tm 2048 -qu root.sparkstreaming -nm flink-cdc-hive
- 进入flink sql命令行
bin/sql-client.sh embedded -s flink-cdc-hive
img
4. 操作Hive
1） 首选创建一个catalog
代码语言：javascript复制CREATE CATALOG hive_catalog WITH (
'type' = 'hive',
'hive-conf-dir' = '/etc/hive/conf.cloudera.hive'
);
这里需要注意：hive-conf-dir是你的hive配置文件地址，里面需要有hive-site.xml这个主要的配置文件，你可以从hive节点复制那几个配置文件到本台机器上面。
2） 查询
此时我们应该做一些常规DDL操作，验证配置是否有问题：
代码语言：javascript复制use catalog hive_catalog;
show databases;
随便查询一张表
代码语言：javascript复制use test
show tables;
select * from people;
可能会报错：
把hadoop-mapreduce-client-core-3.0.0.jar放到flink的Lib目录下，这是我的，实际要根据你的hadoop版本对应选择。
注意：很关键，把这个jar包放到Lib下面后，需要重启application，然后重新用yarn-session启动一个application，因为我发现好像有缓存，把这个application kill 掉，重启才行：
然后，数据可以查询了，查询结果：
5. mysql数据同步到hive
mysql数据无法直接在flink sql导入hive，需要分成两步：
- mysql数据同步kafka；
- kafka数据同步hive；
至于mysql数据增量同步到kafka，前面有文章分析，这里不在概述；重点介绍kafka数据同步到hive。
1） 建表跟kafka关联绑定：
前面mysql同步到kafka，在flink sql里面建表，connector='upsert-kafka'，这里有区别：
代码语言：javascript复制CREATE TABLE product_view_mysql_kafka_parser(
`id` int,
`user_id` int,
`product_id` int,
`server_id` int,
`duration` int,
`times` string,
`time` timestamp
) WITH (
'connector' = 'kafka',
'topic' = 'flink-cdc-kafka',
'properties.bootstrap.servers' = 'kafka-001:9092',
'scan.startup.mode' = 'earliest-offset',
'format' = 'json'
);
2） 建一张hive表
创建hive需要指定SET table.sql-dialect=hive;，否则flink sql 命令行无法识别这个建表语法。为什么需要这样，可以看看这个文档Hive 方言。
-- 创建一个catalag用户hive操作
CREATE CATALOG hive_catalog WITH (
'type' = 'hive',
'hive-conf-dir' = '/etc/hive/conf.cloudera.hive'
);
use catalog hive_catalog;
-- 可以看到我们的hive里面有哪些数据库
show databases;
use test;
show tables;
上面我们可以现在看看hive里面有哪些数据库，有哪些表；接下来创建一张hive表：
代码语言：javascript复制CREATE TABLE product_view_kafka_hive_cdc (
`id` int,
`user_id` int,
`product_id` int,
`server_id` int,
`duration` int,
`times` string,
`time` timestamp
) STORED AS parquet TBLPROPERTIES (
'sink.partition-commit.trigger'='partition-time',
'sink.partition-commit.delay'='0S',
'sink.partition-commit.policy.kind'='metastore,success-file',
'auto-compaction'='true',
'compaction.file-size'='128MB'
);
然后做数据同步：
代码语言：javascript复制insert into hive_catalog.test.product_view_kafka_hive_cdc
select *
from
default_catalog.default_database.product_view_mysql_kafka_parser;
注意：这里指定表名，我用的是catalog.database.table，这种格式，因为这是两个不同的库，需要明确指定catalog - database - table。
网上还有其它方案，关于mysql实时增量同步到hive：
网上看到一篇写的实时数仓架构方案，觉得还可以：
参考资料
https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/table/hive/hive_dialect/
免责声明：本公众号所发布的文章为本公众号原创，或者是在网络搜索到的优秀文章进行的编辑整理，文章版权归原作者所有，仅供读者朋友们学习、参考。对于分享的非原创文章，有些因为无法找到真正来源，如果标错来源或者对于文章中所使用的图片、连接等所包含但不限于软件、资料等，如有侵权，请直接联系后台，说明具体的文章，后台会尽快删除。给您带来的不便，深表歉意。
