【Flink SQL API体验数据湖格式之paimon】

发布时间:2023年12月25日

前言

随着大数据技术的普及,数据仓库的部署方式也在发生着改变,之前在部署数据仓库项目时,首先想到的是选择国外哪家公司的产品,比如:数据存储会从Oracle、SqlServer中或者Mysql中选择,ETL工具会从Informatica、DataStage或者Kettle中选择,BI报表工具会从IBM cognos、Sap Bo或者帆软中选择,基本上使用的产品组合都类似,但随着数据量的激增,之前的部署方式已经越来越不能满足业务场景,例如:不同格式的数据存储,传出的数据库无法存储,而且随着数量的增多,数据库的响应速度就会下降,并且数据大都是T+1的,往往从业务需求的提交到BI报表开发都需要一段时间,等BI报表开发后,数据的时效性大大降低,无法为业务的决策及时性提供帮助,后来随着hadoop的流行,数据仓库慢慢的就演变为以hadoop为基础存储的大数据仓库,并解决了传统数仓无法承载激增数据量的问题,并且随着计算引擎的迭代更新,现在也能实现数据的实时性和事务性,本篇就以新起之秀的数据存储方式来展开介绍。


提示:以下案例仅供参考

一、paimon是什么?

paimon是一种基于LSM形式的数据湖存储格式,与hudi、iceberg定位相同,都是一种基于对hdfs文件存储管理的技术,flink与hudi和iceberg都有做过集成,但hudi和iceberg相当于spark的功能更为完善,这些数据湖格式也都更偏向于批处理,而相对于flink来说,提供的功能相较于spark来说,没有那么完善,虽然flink针对这些方面有做过努力尝试,但结果都不太理想,于是,flink基于前者的有点,自己创造一种数据湖存储格式,其基于flink table store的基础,在结合其他开源数据湖格式的特点加以改进,于是一种新的数据湖格式paimon就诞生了,本人也是最近才开始尝试这种新的数据湖格式的一些功能,下面是基于sql api编写的一个简单的例子。

二、Fink SQL API方式编程

1.创建kafka流标

我这边是以yarn-session的方式执行的,所以首先启动的session,cd $FLINK_HOME,执行bin/yarn-session -d -nm test创建一个名称为test的session会话,随后执行bin/sql-client -s yarn-session进入sql客户端,直接使用默认的catalog和database,执行下面的DDL语句,就会在default_catalog.default_database下创建一个kafka_table表

create temporary table `kafka_table`(
`distinct_id` string,
`login_id` string,
`anonymous_id` string,
`type` string,
`event` string,
`_track_id` string,
`time` string,
`_flush_time` string,
`device_id` string,
`project_id` string,
`map_id` string,
`user_id` string,
`recv_time` string
 ) with(
 'connector'='kafka',
 'topic'='event_topic',
 'properties.group.id'='testgroup',
 'properties.bootstrap.servers'='cdp1:9092',
 'scan.startup.mode'='latest-offset',
 'format'='json'
 );

2.创建paimon append表

接着执行如下DDL语句

CREATE TABLE paimon_append (
`distinct_id` string,
`login_id` string,
`anonymous_id` string,
`type` string,
`event` string,
`_track_id` string,
`time` string,
`_flush_time` string,
`device_id` string,
`project_id` string,
`map_id` string,
`user_id` string,
`recv_time` string
) PARTITIONED BY (`distinct_id`)
WITH (
'bucket' = '-1'
);

3.数据导入

SET ‘execution.checkpointing.interval’ = ‘1 min’;
INSERT INTO paimon_append SELECT * FROM kafka_table;


总结

以上就是一个消费kafka主题数据,并每隔一定的间接直接,写入到paimon表中,paimon会对小文件数据量达到一定程度后,对文件进行压缩合并,并且paimon也支持merge into、update、以及schema evolution等功能,由于时间有限,这里就不仔细展开了,有兴趣的朋友,可以亲自尝试下,版本目标已经更新到0.7,为flink的生态状态又增加了一环,目前flink cdc 、paimon的加持、能很好的解决lamda架构数据不一致,以及kappa架构数据追溯的问题,相信随着后续版本的迭代更多强大的功能也会推出。

文章来源:https://blog.csdn.net/weixin_43153588/article/details/135187113
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。