阿里云全托管flink-vvp平台hudi connector实践(基于emr集群oss-hdfs存储)

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
对象存储 OSS,20GB 3个月
简介: 阿里云全托管flink-vvp平台hudi sink connector实践,本文数据湖hudi基于阿里云E-MapReduce产品,以云对象存储oss-hdfs作为存储

1. 上游数据准备

上游数据以mysql为例作为hudi入湖的上游应用表,详细的建表、插入及更新语句如下

droptable if exists `sunyf_db`.`flink_test_02_hudi`;createtable if not EXISTS `sunyf_db`.`flink_test_02_hudi`(  id BIGINTnotnull PRIMARY KEY 
,`name` varchar(20),datime TIMESTAMP,price FLOAT);INSERTINTO `sunyf_db`.`flink_test_02_hudi` values(1,'zhao', CURRENT_TIMESTAMP,20);INSERTINTO `sunyf_db`.`flink_test_02_hudi` values(2,'qian', CURRENT_TIMESTAMP,30);INSERTINTO `sunyf_db`.`flink_test_02_hudi` values(3,'sun', CURRENT_TIMESTAMP,40);INSERTINTO `sunyf_db`.`flink_test_02_hudi` values(4,'li', CURRENT_TIMESTAMP,50);INSERTINTO `sunyf_db`.`flink_test_02_hudi` values(5,'zhou', CURRENT_TIMESTAMP,60);select*from `sunyf_db`.`flink_test_02_hudi`
update `sunyf_db`.`flink_test_02_hudi` set `name` ='sunyf35'where id =3;


2. Flink-SQL任务

flink-hudi-connector参数参考:https://hudi.apache.org/docs/basic_configurations#Flink-Options

--********************************************************************---- Author:         sunyf-- Created Time:   2023-07-03 17:52:47-- Description:    Write your description here-- Hints:          You can use SET statements to modify the configuration--********************************************************************--CREATE TEMPORARY TABLE mysqlcdc_source
(    id            bigint,`name`      STRING
,datime       TIMESTAMP,price      float,PRIMARY KEY (id)NOT ENFORCED
)WITH ('connector'='mysql','hostname'='rm-xxx.mysql.rds.aliyuncs.com','port'='3306','username'='用户名','password'='密码','database-name'='sunyf_db','table-name'='flink_test_02_hudi');CREATE TEMPORARY TABLE hudi_sink
(    id       bigintNOTNULL,`name`  STRING
,`price` float,PRIMARY KEY (id)NOT ENFORCED
-- 或者直接在id字段后面写 id int not null primary key 也可以-- 或者 指定如下参数 hoodie.datasource.write.recordkey.field)WITH ('connector'='hudi','oss.endpoint'='oss-cn-xx-internal.aliyuncs.com','accessKeyId'='用户名','accessKeySecret'='密码','path'='oss://sunyf-oss-emr02-hudi/emr_hudi_cluster0630.db/flink_test_03_hudi_cow','table.type'='COPY_ON_WRITE'--MERGE_ON_READ,'write.operation'='insert'-- ,'precombine.field' = 'ts'-- ***************-- 下面的元数据相关参数如果不指定的话--hive_sync.enable=false,仅落地文件,需要参考(标题3)中的两种方案建表,'hive_sync.db'='emr_hudi_cluster0630','hive_sync.table'='flink_test_03_hudi_cow','hive_sync.enable'='true','hive_sync.mode'='hms','dlf.catalog.endpoint'='dlf-vpc.cn-beijing.aliyuncs.com','dlf.catalog.region'='cn-beijing'-- ***************-- 在线压缩参数-- ,'compaction.async.enabled' = 'false'-- 指定分区字段,'hoodie.datasource.write.partitionpath.field'='name'-- 指定使用hive的分区表达方式来展示-- 文件夹名为:(day=20230101),不然的话分在oss上展示为20230101,'hoodie.datasource.write.hive_style_partitioning'='true');CREATE TEMPORARY TABLE print_sink
(    id       bigintNOTNULL,`name`  STRING
,datime       TIMESTAMP,`price` float,PRIMARY KEY (id)NOT ENFORCED
)WITH ('connector'='print','logger'='true','print-identifier'='sunyfOutputFormat');BEGIN STATEMENT SET;INSERTINTO hudi_sink
SELECTid
,`name`
,price
FROM mysqlcdc_source
;INSERTINTO print_sink
SELECT*FROM mysqlcdc_source
;end;

3. 建表方案

3.1. 创建hive外表读hudi文件

-- 使用hive外表的时候可以不指定flink hudi sink表中hive_sync 相关参数-- 直接采用外表的属性即可,这个外表的数据没有hudi的元数据字段createtable flink_sink_hudi_externel
(    id    bigintnotnull primary key
,`name` string
,price  float)ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'STORED AS INPUTFORMAT 'org.apache.hudi.hadoop.HoodieParquetInputFormat'OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'LOCATION 'oss://sunyf-oss-emr02-hudi/emr_hudi_cluster0630.db/flink_sink_hudi';

3.2. 创建spark hudi外表

# /bin/bash on spark 3.xspark-sql \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'# sql on spark
use emr_hudi_cluster0630;# 指定location在oss而非默认的oss-hdfs((标题4)中有相关说明)createtable emr_hudi_cluster0630.spark_hudi_location_cow(  id bigint,  name string,  price float)using hudi options(type ='cow',primaryKey='id')location 'oss://sunyf-oss-emr02-hudi/emr_hudi_cluster0630.db/spark_hudi_location_cow';-- 为了避免字段类型不一致,导致的spark报错,这里伪造的原始数据都进行了一下cast-- 有些值cast过后会有精度的损失,hiveinsertinto emr_hudi_cluster0630.spark_hudi_location_cowvalues(cast(1asbigint),'sun',       cast(999.2asfloat)),(cast(2asbigint),'li',        cast(12.1asfloat));

3.3. 外表的删除

# 任一客户端执行:droptable emr_hudi_cluster0630.flink_test_02_hudi_cow;# 外表要手动删除oss上的文件,drop表仅drop元数据,不会删除数据文件
hdfs dfs -rm -f -r oss://sunyf-oss-emr02-hudi/emr_hudi_cluster0630.db/flink_test_02_hudi_cow

4. hudi表更新

通过修改参数实现cow表的实时更新

'write.operation' = 'insert'

image.png

'write.operation' = 'upsert'

image.png

5. 踩坑问题

5.1. spark创建hudi表

创建的时候指定location,desc的时候就会看到表是EXTERNAL 而不是 MANAGED,要注意表删除时相关数据是否留存或者删表重建,数据异常的问题

5.2. spark与hive获取元数据差异

在oss或者通过hdfs dfs命令对hudi表的文件进行删除或者更改后,维持当前的spark-sql-session是不能获取到这一元数据更改的,select查询该表会报文件不存在的异常,需要重启spark-sql客户端或者

refresh table[table_name]

进行元数据更新,hive侧正常执行。

5.3. 字段类型报错

hudi表字段类型与flink的表结构中字段类型要对应,否则在某些增量(手动写入+flink写入)场景下,可能导致数据读取异常(spark),hive兼容性较好,会进行字段类型的隐式转换,但是会有精度损失,如double->float

image.pngimage.png

5.4. flink hudi con对oss-hdfs支持问题

  1. 根据dlf中表存储的path(图3)直接添加到with参数(图1)中,会报非法参数异常(图2)
  2. 尝试:将oss-hdfs的路径后面的endpoint更改到endpoint参数中,jobmanager可以正常启动,但是写下的文件还是不在该内表指定的oss-hdfs存储路径下(日志中可以观察到,图4),oss上验证文件写入情况,如(图5,6)
  3. 查了下oss-hdfs的文档,应该是目前还不支持这个connector直接写默认存储在oss-hdfs的表。
  1. 参考文档:https://help.aliyun.com/document_detail/419069.html?spm=a2c4g.427753.0.i0
  2. 参考文档:https://help.aliyun.com/document_detail/471050.html?spm=a2c4g.141562.0.i6
  1. 绕行方案:在hudi建表的时候指定location在oss正常的bucket路径中,而不是oss-hdfs,可以正常使用image.pngimage.pngimage.pngimage.pngimage.pngimage.png

5.5. oss目录未删除问题

oss-hdfs 删除的db未删除(图1),dlf中已删除(图2 show databases 与dlf结果一致)

hdfs dfs -rm -f -r oss://sunyf-oss-emr02-hudi/emr_hudi_cluster0630.db/flink_sink_hudi_location

image.png

image.png

相关实践学习
如何快速连接云数据库RDS MySQL
本场景介绍如何通过阿里云数据管理服务DMS快速连接云数据库RDS MySQL,然后进行数据表的CRUD操作。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
打赏
0
0
0
0
26
分享
相关文章
Flink 2.0 存算分离状态存储 — ForSt DB 
本文整理自阿里云技术专家兰兆千在Flink Forward Asia 2024上的分享,主要介绍Flink 2.0的存算分离架构、全新状态存储内核ForSt DB及工作进展与未来展望。Flink 2.0通过存算分离解决了本地磁盘瓶颈、检查点资源尖峰和作业恢复速度慢等问题,提升了云原生部署能力。ForSt DB作为嵌入式Key-value存储内核,支持远端读写、批量并发优化和快速检查点等功能。性能测试表明,ForSt在异步访问和本地缓存支持下表现卓越。未来,Flink将继续完善SQL Operator的异步优化,并引入更多流特性支持。
359 88
Flink 2.0 存算分离状态存储 — ForSt DB 
大数据实时计算产品的对比测评:实时计算Flink版 VS 自建Flink集群
本文介绍了实时计算Flink版与自建Flink集群的对比,涵盖部署成本、性能表现、易用性和企业级能力等方面。实时计算Flink版作为全托管服务,显著降低了运维成本,提供了强大的集成能力和弹性扩展,特别适合中小型团队和业务波动大的场景。文中还提出了改进建议,并探讨了与其他产品的联动可能性。总结指出,实时计算Flink版在简化运维、降低成本和提升易用性方面表现出色,是大数据实时计算的优选方案。
zdl
212 56
基于OpenLake的Flink+Paimon+EMR StarRocks流式湖仓分析
阿里云OpenLake解决方案建立在开放可控的OpenLake湖仓之上,提供大数据搜索与AI一体化服务。通过元数据管理平台DLF管理结构化、半结构化和非结构化数据,提供湖仓数据表和文件的安全访问及IO加速,并支持大数据、搜索和AI多引擎对接。本文为您介绍以Flink作为Openlake方案的核心计算引擎,通过流式数据湖仓Paimon(使用DLF 2.0存储)和EMR StarRocks搭建流式湖仓。
641 5
基于OpenLake的Flink+Paimon+EMR StarRocks流式湖仓分析
Flink-12 Flink Java 3分钟上手 Kubernetes云原生下的Flink集群 Rancher Stateful Set yaml详细 扩容缩容部署 Docker容器编排
Flink-12 Flink Java 3分钟上手 Kubernetes云原生下的Flink集群 Rancher Stateful Set yaml详细 扩容缩容部署 Docker容器编排
134 3
|
4月前
|
实时计算Flink版在稳定性、性能、开发运维、安全能力等等跟其他引擎及自建Flink集群比较。
实时计算Flink版在稳定性、性能、开发运维和安全能力等方面表现出色。其自研的高性能状态存储引擎GeminiStateBackend显著提升了作业稳定性,状态管理优化使性能提升40%以上。核心性能较开源Flink提升2-3倍,资源利用率提高100%。提供一站式开发管理、自动化运维和丰富的监控告警功能,支持多语言开发和智能调优。安全方面,具备访问控制、高可用保障和全链路容错能力,确保企业级应用的安全与稳定。
75 0
大数据-127 - Flink State 04篇 状态原理和原理剖析:状态存储 Part2
大数据-127 - Flink State 04篇 状态原理和原理剖析:状态存储 Part2
40 0
大数据-126 - Flink State 03篇 状态原理和原理剖析:状态存储 Part1
大数据-126 - Flink State 03篇 状态原理和原理剖析:状态存储 Part1
105 0
实时计算 Flink版产品使用问题之如何高效地将各分片存储并跟踪每个分片的消费位置
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
大数据-82 Spark 集群模式启动、集群架构、集群管理器 Spark的HelloWorld + Hadoop + HDFS
大数据-82 Spark 集群模式启动、集群架构、集群管理器 Spark的HelloWorld + Hadoop + HDFS
244 6
Hadoop-20 Flume 采集数据双写至本地+HDFS中 监控目录变化 3个Agent MemoryChannel Source对比
Hadoop-20 Flume 采集数据双写至本地+HDFS中 监控目录变化 3个Agent MemoryChannel Source对比
93 3

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等