Iceberg Spark存储过程-表治理工具

发布时间 2023-12-08 16:12:35作者: 黑水滴

一、简介

存储过程(Procedure)是数据库领域的概念,类似于编程语言中的方法或函数,是对实现特定操作的封装,原生的 Spark SQL 中是不支持存储过程的,Iceberg 0.11.0版本之后对其进行了扩展,并提供了部分存储过程的实现。Iceberg 中提供的所有存储过程都在system namespace 下,分快照管理、元数据管理、表迁移三种。如果需要使用则spark新增如下配置项

spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

二、存储过程使用

1、语法简介

-- 基于参数位置
call catalog_name.system.procedure_name(arg_1, arg_2, ... arg_n);
-- 基于参数名称
call catalog_name.system.procedure_name(arg_name_2 => arg_2, arg_name_1 => arg_1);

三、Iceberg快照管理

1、回滚表到特定的快照ID(rollback_to_snapshot)

(1)参数

table(必须):string,表名

snapshot_id(必须):long,快照 ID

(2)示例

call catalog_name.system.rollback_to_snapshot('db.sample', 1);

2、回滚表到特定时间(rollback_to_timestamp)

(1)参数

table(必须):string,表名

timestamp(必须):long,回滚的时间戳

(2)示例

call catalog_name.system.rollback_to_timestamp('db.sample', TIMESTAMP '2023-12-08 00:00:00.000');

3、设置表的当前快照ID(set_current_snapshot)

与回滚不同,该参数可以在各个快照之间任意穿梭

(1)参数

table(必须):string,表名

snapshot_id(必须):long,快照 ID

(2)示例

call catalog_name.system.set_current_snapshot('db.sample', 1);

4、从现有快照创建新快照(cherrypick_snapshot)

(1)参数

table(必须):string,表名

snapshot_id(必须):long,快照 ID

四、元数据管理

1、删除过期快照和相关数据文件(expire_snapshots)

(1)参数

table(必须):string,表名
older_than:timestamp,该时间戳之前的快照将被删除,默认为 5 天前
retain_last:int,和 older_than 同时存在的时候,要保留的快照数(默认为 1)
max_concurrent_deletes:int,用于删除文件操作的线程池大小(默认不使用线程池)

(2)示例

CALL spark_catalog.system.expire_snapshots(table => 'db.sample', older_than => TIMESTAMP '2023-12-07 10:40:00.000');

2、删除Iceberg元数据中未被引用的文件(remove_orphan_files)

(1)参数

table(必须):string,表名。
older_than:timestamp,删除在此时间戳之前创建的孤立文件(默认为 3 天前)。
location:string,查找文件的目录(默认为表的位置)。
dry_run:boolean,当为 true 时,实际上不删除文件(默认为 false)。
max_concurrent_deletes:int,用于删除文件操作的线程池大小(默认不使用线程池)。

(2)示例