使用Flink SQL实时入湖Hudi/Hive

文章目录

  • 1 Hudi 简介
  • 2 COW和MOR
  • 3 接入COW模式Hudi表
  • 4 使用Flink SQL查看新接表
  • 5 使用Hive查看新接表
  • 6 总结

1 Hudi 简介

Hudi是一个流式数据湖平台,使用Hudi可以直接打通数据库与数据仓库,连通大数据平台,支持对数据增删改查。Hudi还支持同步数据入库,提供了事务保证、索引优化,是打造实时数仓、实时湖仓一体的新一代技术。下面以我实际工作中遇到的问题,聊下湖仓一体的好处,如有不对,敬请指正。
在这里插入图片描述
像传统关系型数据库,MySQL/Oracle等大多支持OLTP,但不支持OLAP。如果写很复杂的SQL,传统关系型数据库根本跑不动,尤其是需要跨系统/跨数据库联合查询分析,传统关系型数据库并不支持(这个可以使用Presto解决)。

在这里插入图片描述
而离线数仓无法支持实时/准实时需求,无法记录级更新,当业务表数据量很大时,无论使用增量还是全量接入Hive,对业务库都有很大压力(使用从库可缓解)。Hudi能很好解决这个问题,通过配置可以准实时的写入Hudi,并同步到Hive,相当于业务表数据准实时的同步到Hive,这时取快照或者直接当作ODS层都可,再也不用担心ODS接入延迟了。
在这里插入图片描述

2 COW和MOR

Hudi有两种表类型,COW和MOR,如果接入表读多写少可选择COW,如字典表,读少写多使用MOR。
Copy on write:写时复制,使用列式文件格式(如 parquet)存储数据。不同进程在访问同一资源的时候,只有更新操作,才会去复制一份新的数据并更新替换,否则都是访问同一个资源。
Merge on read:读时合并,使用列式+基于行的(例如avro)文件格式的组合存储数据。更新被记录到增量文件中,然后被压缩以同步或异步地生成新版本的列式文件。
在这里插入图片描述
如果Hudi表是COPY_ON_WRITE类型,那么映射的Hive表对应是指定的Hive表名,此表中存储着Hudi所有数据。

如果Hudi表类型是MERGE_ON_READ模式,那么映射的Hive表将会有2张,一张后缀为rt ,另一张表后缀为ro。后缀rt对应的Hive表中存储的是Base文件Parquet格式数据+log Avro格式数据,也就是全量数据。后缀为ro Hive表中存储的是存储的是Base文件对应的数据。

3 接入COW模式Hudi表

开发测试时,可在客户端调试

./bin/sql-client.sh embedded -s yarn-session

调试没问题后,在DolphinScheduler配置上线
在这里插入图片描述
选择FLINK_STREAM
在这里插入图片描述
根据集群类型,选择部署方式

初始化脚本
初始化脚本配置一些参数和建表

SET 'yarn.application.queue' = 'root.etl';
set execution.checkpointing.interval='300s';
SET execution.checkpointing.mode = AT_LEAST_ONCE;
-- 保存checkpoint文件的目录
set state.checkpoints.dir='hdfs://cluster/tmp/flink/checkpoints/h_account_holiday';
-- 恢复时需设置检查点 set execution.savepoint.path='hdfs://cluster/tmp/flink/checkpoints/h_account_holiday/077107d6530a1c63cb9126258cfe2546/chk-72';set taskmanager.network.memory.buffer-debloat.enabled=true;SET state.checkpoints.num-retained= 3; 
SET execution.checkpointing.externalized-checkpoint-retention = RETAIN_ON_CANCELLATION;set execution.checkpointing.min-pause = '180000';
set 'table.exec.sink.upsert-materialize' = 'NONE';
set execution.checkpointing.max-concurrent-checkpoints=1;set akka.ask.timeout = '1200s';
set web.timeout = '500000';
set heartbeat.timeout=500000;SET 'connector.mysql-cdc.max-connection-attempts' = '5';
SET 'connector.mysql-cdc.connection-attempts-timeout' = '1200s';SET restart-strategy='fixed-delay';
SET restart-strategy.fixed-delay.attempts='50';
SET restart-strategy.fixed-delay.delay='1min';
SET execution.checkpointing.timeout='40min';SET state.backend='rocksdb';
SET state.backend.incremental=true;set high-availability='zookeeper';
set high-availability.storageDir='hdfs://cluster/tmp/flink/ha-yarn';
set high-availability.zookeeper.quorum='bigdata-093:2181,bigdata-094:2181,bigdata-ds-12-195:2181,bigdata-ds-12-198:2181,bigdata-ds-12-199:2181';
set high-availability.zookeeper.path.root='/flink_yarn';
set yarn.application-attempts='10';CREATE CATALOG cdc_catalog WITH (
'type' = 'hive',
'default-database' = 'flink_cdc',
'hive-conf-dir' = '/opt/apps/apache-hive-2.1.1-bin/conf'
);
-- 使用刚创建的catalog
use catalog cdc_catalog;
-- 选择flink_cdc库
use flink_cdc;drop table if exists source_account_holiday;
create table if not exists source_account_holiday(
`id` int primary key not enforced
,workday date
,week int
,next_workday date
,create_time timestamp
,update_time timestamp
) with (
'connector'='mysql-cdc',
'hostname'='10.100.xx.xx',
'port'='3306',
'server-time-zone'='Asia/Shanghai',
'server-id'='6066-6070', -- 注意同一个实例,id不要重复,数字范围要大于并行度
'username'='xxx',
'password'='xxx',
'debezium.snapshot.mode'='initial',
'database-name'='xd_account',
'table-name'='account_holiday',
'connect.timeout'='1000000'
);drop table if exists sink_account_holiday;
create table if not exists sink_account_holiday(
`id` int primary key not enforced
,workday date
,week int
,next_workday date
,create_time string -- 注意timestamp需转成string
,update_time string -- 注意timestamp需转成string
) with (
'connector' = 'hudi',
'path' = 'hdfs://cluster/tmp/flink/hudi/sink_account_holiday',
'hoodie.datasource.write.recordkey.field'='id', -- 设置主键
'table.type'='COPY_ON_WRITE',
'write.timezone'='Asia/Shanghai',
'hive_sync.enabled'='true',
'hive_sync.mode'='hms',
'hive_sync.metastore.uris'='thrift://bigdata-003:9083,thrift://bigdata-004:9083,thrift://bigdata-009:9083,thrift://bigdata-012:9083,thrift://bigdata-008:9083,thrift://bigdata-007:9083',
'hive_sync.db'='hudi', -- 同步到hive hudi库h_account_holiday,自动建表
'hive_sync.table'='h_account_holiday',
'hive_sync.username'='hive',
'hoodie.datasource.hive_sync.omit_metadata_fields'='true'
);

脚本
从source表写入sink表

insert into sink_account_holiday
select id
,workday 
,week 
,next_workday 
,date_format(create_time, 'yyyy-MM-dd HH:mm:ss') -- 注意timestamp需转成string
,date_format(update_time, 'yyyy-MM-dd HH:mm:ss') -- 注意timestamp需转成string
from source_account_holiday;

在这里插入图片描述
执行后注意看日志,成功会有Application ID 和 Job ID
在这里插入图片描述
可通过Application ID 和 Job ID查看任务运行情况
在这里插入图片描述

4 使用Flink SQL查看新接表

使用Flink SQL,可以实时看到数据更新

cd /opt/apps/flink-1.14.4/
./bin/sql-client.sh embedded -s yarn-session

embedded 内嵌模式

Flink SQL> CREATE CATALOG cdc_catalog WITH (
> 'type' = 'hive',
> 'default-database' = 'flink_cdc',
> 'hive-conf-dir' = '/opt/apps/apache-hive-2.1.1-bin/conf'
> );
log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
[INFO] Execute statement succeed.Flink SQL> use catalog cdc_catalog;
[INFO] Execute statement succeed.Flink SQL> show databases;Flink SQL> use hudi;
[INFO] Execute statement succeed.
Flink SQL> select * from h_account_holiday limit 10;

在这里插入图片描述

5 使用Hive查看新接表

前面初始化脚本必须配置同步到hive,hive查不了source和sink表,只能查同步到hive的表

hive> use hudi;
OK
Time taken: 2.406 seconds
hive> set role admin;
OK
Time taken: 0.093 seconds
hive> select * from h_account_holiday limit 10;
OK
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
44      2024-05-12      7       2024-05-13      2024-01-20 15:17:59     2024-01-20 15:17:59
45      2024-05-18      6       2024-05-20      2024-01-20 15:17:59     2024-01-20 15:17:59
89      2024-10-04      5       2024-10-08      2024-01-20 15:17:59     2024-01-20 15:17:59
110     2024-12-14      6       2024-12-16      2024-01-20 15:17:59     2024-01-20 15:17:59
112     2024-12-21      6       2024-12-23      2024-01-20 15:17:59     2024-01-20 15:17:59
115     2024-12-29      7       2024-12-30      2024-01-20 15:17:59     2024-01-20 15:17:59
91      2024-10-06      7       2024-10-08      2024-01-20 15:17:59     2024-01-20 15:17:59
93      2024-10-13      7       2024-10-14      2024-01-20 15:17:59     2024-01-20 15:17:59
50      2024-06-02      7       2024-06-03      2024-01-20 15:17:59     2024-01-20 15:17:59
95      2024-10-20      7       2024-10-21      2024-01-20 15:17:59     2024-01-20 15:17:59
Time taken: 0.147 seconds, Fetched: 10 row(s)

在这里插入图片描述

6 总结

使用这种方案,真正实现了湖仓一体,基本满足了实时和离线需求,且主要使用SQL,开发和维护成本较低。不过,该方案也有个问题,flink cdc 会挂,导致数据没更新,还是要多关注下。

参考链接:
https://blog.csdn.net/qq_32727095/article/details/123863620
https://zhuanlan.zhihu.com/p/471842018
https://zhuanlan.zhihu.com/p/526372429
https://blog.csdn.net/JH_Zhai/article/details/136042662
https://www.jianshu.com/p/0837ada9de76

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/832822.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

如何应对访问国外服务器缓慢的问题?SDWAN组网是性价比之选

访问国外服务器缓慢通常由以下原因造成: 1、政策限制:我国管理互联网,限制部分国外网站和服务器,以维护国家安全稳定。 2、技术障碍:国内与国际互联网的网络架构和协议存在差异,可能导致数据传输不兼容。 …

数据结构——图的基础知识与其表示

一:定义 由顶点的集合和边的集合组成;常以 G(V,E) 表示,G 代表图,V代表 顶点的集合,E代表边的集合; 如图: 在G1图中,有 0~4 五个顶点,有 0-1,0-2&…

Tomcat同时支持http及https

Tomcat同时支持http及https 1、Tomcat同时支持http及https2、Springboot同时支持http及https 1、Tomcat同时支持http及https <Connector port"8443" protocol"org.apache.coyote.http11.Http11NioProtocol" maxThreads"150" SSLEnabled"…

V23 中的新功能:LEADTOOLS 展示了它的 EXCEL-lence

LEADTOOLS (Lead Technology)由Moe Daher and Rich Little创建于1990年&#xff0c;其总部设在北卡罗来纳州夏洛特。LEAD的建立是为了使Daher先生在数码图象与压缩技术领域的发明面向市场。在过去超过30年的发展历程中&#xff0c;LEAD以其在全世界主要国家中占有的市场领导地位…

苹果自研大语言模型“Ajax“ 助力iOS 18升级;Stack Overflow与OpenAI建立API合作伙伴关系

&#x1f989; AI新闻 &#x1f680; 苹果自研大语言模型"Ajax" 助力iOS 18升级 摘要&#xff1a;苹果公司预计通过自研大语言模型Ajax来为iOS 18和Siri带来重大升级&#xff0c;但不计划推出类似ChatGPT的AI聊天机器人。Ajax模型基于Google的Jax框架&#xff0c;并…

深度学习之基于YOLOv5目标检测可视化系统

欢迎大家点赞、收藏、关注、评论啦 &#xff0c;由于篇幅有限&#xff0c;只展示了部分核心代码。 文章目录 一项目简介 二、功能三、系统四. 总结 一项目简介 一、项目背景与意义 随着深度学习技术的快速发展&#xff0c;目标检测在多个领域中的应用日益广泛&#xff0c;包括…

filezilla 上传文件到服务器

1. 下载windows ftp客户端 https://filezilla-project.org/index.php 点击左侧的 Download FileZilla Client All platforms 默认安装。 2. 双击打开 filezilla 按图中填写&#xff0c;并上传。 主机名: 10.20.89.* 用户名: zhuzh 密码: *** 端口: 22 点击 快速连接&#…

中电金信:金Gien乐道 | 4月要闻速览,精彩再回顾

中国电子党组副书记、总经理李立功一行调研中电金信 4月10日&#xff0c;中国电子党组副书记、总经理李立功一行赴中电金信进行调研&#xff0c;深入听取了中电金信经营发展情况、研发工作及“源启”行业数字底座平台的汇报&#xff0c;并参观了公司展厅和科技研发场所&#xf…

定性分析和定量分析

目录 一、什么是定性分析和定量分析 二、定性分析和定量分析的区别 三、常用分析方法 3.1 常用定性分析方法 3.2 常用定量分析方法 一、什么是定性分析和定量分析 定性研究方法是根据社会现象或事物所具有的属性和在运动中的矛盾变化&#xff0c;从事物的内在规定性来研究…

hadoop---基于Hive的数据仓库相关函数机制及其优化方案

Hive相关函数&#xff08;部分&#xff09;&#xff1a; if函数: 作用: 用于进行逻辑判断操作 语法: if(条件, true返回信息,false返回信息) 注意: if函数支持嵌套使用 select if(aa,’bbbb’,111) fromlxw_dual; bbbb select if(1<2,100,200) fromlxw_dual; 200nvl函数:…

矿山机械自动化中的激光雷达技术探索

在矿山机械自动化技术的快速发展中&#xff0c;激光雷达技术作为其关键组成部分&#xff0c;正发挥着越来越重要的作用。本文将深入探讨激光雷达在矿山机械自动化中的应用&#xff0c;以及其所面临的挑战与未来发展趋势。 一、激光雷达在矿山机械自动化中的应用 激光雷达技术…

Java中的maven的安装和配置

maven的作用 依赖管理 方便快捷的管理项目依赖的资源&#xff0c;避免版本冲突问题 统一项目管理 提供标准&#xff0c;统一的项目结构 项目构建 标准跨平台&#xff08;Linux、windows、MacOS&#xff09;的自动化项目构建方式 maven的安装和配置 在maven官网下载maven Ma…

SAP-ABAP-搜索帮助03

一、搜索帮助有两种&#xff1a; 1、外键/检查表&#xff1a;基于个表的&#xff0c;数据有效性检测。 2、SE11手工创建搜索帮助&#xff1a;基于数据元素的&#xff0c;纯帮助&#xff0c;无校验。 二、创建搜索帮助 1、外键/检查表 例如学员考勤表&#xff0c;包括学院的…

2015-2016 ACM-ICPC Southwestern Europe Regional Contest (SWERC 15) A.Promotions

Promotions 题目描述 The Fair Inc. administration decided to promote the best employees and limited the number of promotions to a fixed interval [A, B]. The directors compared the employees’ performance and their evaluations resulted in a consistent prec…

拯救者杯OPENAIGC开发者大赛城市巡回沙龙,苏州站报名开启!

由联想拯救者、AIGC开放社区、英特尔联合主办的“AI生成未来第二届拯救者杯OPENAIGC开发者大赛”自上线以来&#xff0c;吸引了广大开发者的热情参与。 为了向技术开发者、业务人员、高校学生、以及个体创业人员等参赛者们提供更充分的帮助与支持&#xff0c;AIGC开放社区特别…

vue 输入框@人功能组件,支持复制粘贴

文章目录 概要整体架构流程技术细节小结自定义输入框代码外部调用示例 概要 开发任务系统中&#xff0c;业务需求&#xff1a;需要在任务描述、评论等地方支持人员功能&#xff0c;可以将任务外部人员添加至当前任务中。 功能&#xff1a;1、支持输入展开下拉&#xff0c;可通…

Linux:进程通信(二)信号的保存

目录 一、信号的处理是否是立即处理的&#xff1f; 二、信号如何保存 1、阻塞、未决、递达 2、信号集 3、信号集操作函数 4、sigprocmask函数 5、sigpending 函数 上篇文章我们讲解了信号的产生&#xff1a;Linux&#xff1a;进程信号&#xff08;一&#xff09;信号的产…

ResponseHttp

文章目录 HTTP响应详解使用抓包查看响应报文协议内容 Response对象Response继承体系Response设置响应数据功能介绍Response请求重定向概述实现方式重定向特点 请求重定向和请求转发比较路径问题Response响应字符数据步骤实现 Response响应字节数据步骤实现 HTTP响应详解 使用抓…

「AIGC」深度学习

深度学习是机器学习的一个子领域&#xff0c;它基于人工神经网络的学习算法。深度学习在图像和语音识别、自然语言处理、医学图像分析、药物发现、自动驾驶汽车等领域取得了显著的进展。以下是围绕深度学习的几个关键主题的阐述。 学习路线 基础数学&#xff1a; 了解线性代数…