数据湖技术之iceberg(六)Iceberg表数据组织与查询

南墨2年前技术文章1216

1     Iceberg表数据组织与查询

1) 下载avro-tools jar包

由于后期需要查看avro文件内容,我们可以通过avro-tool.jar来查看avro数据内容。可以在以下网站中下载avro-tools对应的jar包,下载之后上传到节点上:

https://mvnrepository.com/artifact/org.apache.avro/avro-tools

查看avro文件信息可以直接执行如下命令,可以将avro中的数据转换成对应的json数据。

[root@node5 ~]# java -jar /software/avro-tools-1.8.1.jar tojson snap-*-wqer.avro

2) 在Hive中创建Iceberg表并插入数据

在Hive中创建Iceberg格式表,并插入如下数据:

#在Hive中创建iceberg格式表

create table test_iceberg_tbl1(

id int ,

name string,

age int)

partitioned by (dt string)

stored by 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler';

 

#插入如下数据

insert into test_iceberg_tbl1 values (1,"zs",21,"20211212");

insert into test_iceberg_tbl1 values (2,"ls",22,"20211212");

insert into test_iceberg_tbl1 values (3,"ww",23,"20211213");

insert into test_iceberg_tbl1 values (4,"ml",24,"20211213");

insert into test_iceberg_tbl1 values (5,"tq",25,"20211213");

3) 查看Iceberg底层数据存储

下图为Iceberg表“test_iceberg_tbl1”在HDFS中存储的数据组织图:

通过上图我们可以看到有5个Snapshot快照,以上5个Snapshot实际上就是对应了5个Manifest list清单列表。

查询最新快照数据

为了了解Iceberg如何查询最新数据,可以参照下面这张图来详细了解底层实现。

查询Iceberg表数据时,首先获取最新的metadata信息,这里先获取到“00000-*ec504.metadata.json”元数据信息,解析当前元数据文件可以拿到当前表的快照id:“949358624197301886”以及这张表的所有快照信息,也就是json信息中snapshots数组对应的值。根据当前表的快照id值可以获取对应的snapshot对应的avro文件信息:“snap-*-32800.avro”,我们可以找到当前快照对应的路径,看到其包含的Manifest 清单文件有5个:"*32800-m0.avro"、"*2abba-m0.avro"、"*d33de-m0.avro"、"*748bf-m0.avro"、"*b946e-m0.avro",读取该Iceberg格式表最新数据就是读取这几个文件中描述对应的parquet数据文件即可。

我们可以看到“snap-*-32800.avro”快照文件中不仅有包含的manifest路径信息,还有“added_data_files_count”、“existing_data_files_count”、“deleted_data_files_count”三个属性,Iceberg 根据 deleted_data_files_count 大于 0 来判断对应的manifest清单文件里面是不是被删除的数据,如果一个manifest清单文件该值大于0代表数据删除,读数据时就无需读这个manifest清单文件对应的数据文件。

根据Manifest list找到了各个对应的manifest 清单文件,每个文件中描述了对应parquet文件存储的位置信息,可以看到在对应的avro文件中有“status”属性,该属性为1代表对应的parquet文件为新增文件,需要读取,为2代表parquet文件被删除。

查询某个快照的数据

Apache Iceberg支持查询历史上任何时刻的快照,在查询时需要指定snapshot-id属性即可,这个只能通过Spark/Flink来查询实现,例如在Spark中查询某个快照数据如下:

spark.read.option("snapshot-id",6155408340798912701L).format("iceberg").load("path")

查询某个快照数据的原理如下图所示(以查询快照id为“6155408340798912701”的数据为例):

通过上图可以看出,实际上读取历史快照数据和读取最新数据不同之处就是找到的snapshot-id不同而已,原理都是一样。

根据时间戳查看某个快照的数据

Apache iceberg还支持通过as-of-timestamp参数执行时间戳来读取某个快照的数据,同样也是通过Spark/Flink来读取,Spark读取代码如下:

spark.read.option("as-of-timestamp","时间戳").format("iceberg").load("path")

实际上通过时间戳找到对应数据文件的原理与通过snapshot-id找到数据文件原理一样,在*.metadata.json文件中,除了有“current-snapshot-id”、“snapshots”属性外还有“snapshot-log”属性,该属性对应的值如下:

我们可以看到其中有个 timestamp-ms 属性和 snapshot-id 属性,并且是按照 timestamp-ms 升序的。在 Iceberg 内部实现中,它会将 as-of-timestamp 指定的时间和 snapshot-log 数组里面每个元素的 timestamp-ms 进行比较,找出最后一个满足 timestamp-ms <= as-of-timestamp 对应的 snapshot-id,原理同上,通过snapshot-id再找到要读取的数据文件。


相关文章

keycloak部署和使用

keycloak部署和使用

简介Keycloak是一个开源软件产品,旨在为现代的应用程序和服务,提供包含身份管理和访问管理功能的单点登录工具。截至2018年3月,红帽公司负责管理这一JBoss社区项目,并将其作为他们RH-SSO...

trino容器对接hudi(五)

trino容器对接hudi(五)

前提:本文是基于已经部署了trino容器的基础上进行的。冒烟测试是在trino对接ldap后并且ranger已经对接了metastore权限后,并且spark组件已经对接hudi,并且成功创建hudi...

kafka启动失败

kafka启动失败

问题现象客户重启kafka服务,页面显示run fail        问题报错问题原因  &nbs...

MS SQL Server 实战 统计与汇总重复记录

MS SQL Server 实战 统计与汇总重复记录

需求在日常的数据管理应用中,统计和汇总重复记录的情况是经常遇到的一个问题,然后我们会根据统计结果进一步对数据进行合理化处理。比如我们有一组题库数据,主要包括题目和选项字段(如单选题目、多选题目和判断题...

基于Jenkins和Kubernetes的CI/CD

基于Jenkins和Kubernetes的CI/CD

jenkins安装安装helm CRD编辑好yaml文件后直接安装即可k3s kubectl apply -f jenkins-crd.yaml查看我们安装好的helmchart crdk3s kub...

SQL Server优化入门系列(一)——快速定位阻塞SQL

SQL Server优化入门系列(一)——快速定位阻塞SQL

引言我们在运维数据库的时候,经常会面对这样的问题:数据库现在运行得怎么样有哪些会话在执行,当前状态是什么,在执行什么SQL哪些会话被阻塞,阻塞原因是什么会话是从哪些IP连接过来的,使用了什么账号通过这...

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。