等待下一个秋

  • Spark
  • Flink
  • Hive
  • 数据仓库
  • ClickHouse
  • 收徒弟
  • Java
    • Spring
    • Mybatis
    • SpringBoot
    • 面试题
  • Python
    • Python基础
    • 爬虫
    • Numpy
    • matplotlib
    • Flask
  • 技术杂谈
    • Linux知识
    • Docker
    • Git教程
    • Redis教程
    • mysql
    • 前端
    • R语言
    • 机器学习
  • 关于我
  • 其它
    • 副业挣钱
    • 资料下载
    • 资料文档
专注于Hadoop/Spark/Flink/Hive/数据仓库等
关注公众号:大数据技术派,获取更多学习资料。
  1. 首页
  2. Flink
  3. 正文

flink-cdc同步mysql数据到hive

2022年9月16日 1774点热度 0人点赞 0条评论

本文首发于我的个人博客网站 等待下一个秋-Flink

什么是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目录下面。

  1. flink-sql-connector-hive-2.2.0_2.11-1.13.5.jar

如果你的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

官方文档地址在这里,可以自行查看。

3. 启动flink-sql client

1) 先在yarn上面启动一个application,进入flink13.5目录,执行:

bin/yarn-session.sh -d -s 2 -jm 1024 -tm 2048 -qu root.sparkstreaming -nm flink-cdc-hive

2) 进入flink sql命令行

bin/sql-client.sh embedded -s flink-cdc-hive

img

4. 操作Hive

1) 首选创建一个catalog

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操作,验证配置是否有问题:

use catalog hive_catalog;
show databases;

随便查询一张表

use test
show tables;
select * from people;

可能会报错:

image-20220915183211513

把hadoop-mapreduce-client-core-3.0.0.jar放到flink的Lib目录下,这是我的,实际要根据你的hadoop版本对应选择。

注意:很关键,把这个jar包放到Lib下面后,需要重启application,然后重新用yarn-session启动一个application,因为我发现好像有缓存,把这个application kill 掉,重启才行:

image-20220915183454691

然后,数据可以查询了,查询结果:

image-20220915183102548

5. mysql数据同步到hive

mysql数据无法直接在flink sql导入hive,需要分成两步:

  1. mysql数据同步kafka;
  2. kafka数据同步hive;

至于mysql数据增量同步到kafka,前面有文章分析,这里不在概述;重点介绍kafka数据同步到hive。

1) 建表跟kafka关联绑定:

前面mysql同步到kafka,在flink sql里面建表,connector='upsert-kafka',这里有区别:

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表:

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'
);

然后做数据同步:

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:

img

网上看到一篇写的实时数仓架构方案,觉得还可以:

image-20220916134859155

参考资料

https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/table/hive/hive_dialect/

标签: Flink
最后更新:2022年9月16日

等待下一个秋

待我代码写成,便娶你为妻!专注于Hadoop/Spark/Flink/Hive/数据仓库等,关注公众号:大数据技术派,获取更多学习资料。

打赏 点赞
< 上一篇
下一篇 >

文章评论

取消回复

等待下一个秋

待我代码写成,便娶你为妻!专注于Hadoop/Spark/Flink/Hive/数据仓库等,关注公众号:大数据技术派,获取更多学习资料。

搜一搜
微信
最新 热点 随机
最新 热点 随机
ClickHouse 自定义分区键 ClickHouse数据副本引擎 ClickHouse ReplacingMergeTree引擎 ClickHouse MergeTree引擎 clickhouse简介 Flink SQL管理平台flink-streaming-platform-web安装搭建
Matplotlib 图例指南 Python作图模块之turtle Redis 简介 数仓建模—ID Mapping 第05讲:Flink SQL & Table 编程和案例 2021,我这一年
标签聚合
mysql 挣钱 Hive 书籍 算法 大数据 Python Flink Java Redis 数据仓库 R语言
文章归档
  • 2022年12月
  • 2022年11月
  • 2022年9月
  • 2022年7月
  • 2022年6月
  • 2022年5月
  • 2022年4月
  • 2022年3月
  • 2022年2月
  • 2022年1月
  • 2021年12月
  • 2021年11月
  • 2021年10月
  • 2021年9月
  • 2021年8月
  • 2021年6月
  • 2021年5月
  • 2021年4月
  • 2021年3月
  • 2021年2月
  • 2021年1月
  • 2020年12月
  • 2020年11月
  • 2020年10月
  • 2020年9月
  • 2020年8月
  • 2020年7月
  • 2020年5月
  • 2020年4月
  • 2020年1月
  • 2019年9月
  • 2019年8月
  • 2019年7月
  • 2019年6月
  • 2019年5月
  • 2019年4月
  • 2019年3月
  • 2019年1月
  • 2018年12月
  • 2017年5月

©2022 ikeguang.com. 保留所有权利。

鄂ICP备2020019097号-1

鄂公网安备 42032202000160号