【Iceberg分析】Spark与Iceberg集成之常用存储过程

Spark与Iceberg集成之常用存储过程

spark与iceberg集成后，可以通过内置的存储过程来进行表的管理。使用CALL来调用存储过程。所有的存储过程在system的命名空间中。

官网地址

由于表迁移功能的风险较大，所以不去进行表的迁移，使用重建Iceberg表，重写数据的方式进行切换。

调用语法

catalog_name代表catalog的名称，procedure_name代表存储过程的名称，参数可以通过指定参数名的方式入参，也可以使用位移的方式入参。

CALL catalog_name.system.procedure_name(arg_name_2 => arg_2, arg_name_1 => arg_1);
CALL catalog_name.system.procedure_name(arg_1, arg_2, ... arg_n);

调用样例

 SparkSession spark = SparkSession.builder().master("local").appName("Iceberg spark example").config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions").config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog").config("spark.sql.catalog.local.type", "hadoop") //指定catalog 类型.config("spark.sql.catalog.local.warehouse", "iceberg_warehouse").getOrCreate();spark.sql("CALL local.system.rollback_to_snapshot('iceberg_db.table2', 3285133177610707025) ");

表快照管理

快照回滚

根据snapshotid进行回滚

存储过程名 rollback_to_snapshot

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
snapshot_id	✔️	long	metadata文件中snapshots中的快照子项snapshot-id的值

根据timestamp进行回滚

存储过程名 rollback_to_timestamp

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
timestamp	✔️	timestamp	metadata文件中snapshots中的快照子项timestamp-ms的值

设置表当前生效的快照

存储过程名 set_current_snapshot

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
snapshot_id		long	取metadata文件中snapshots中的快照子项snapshot-id的值
ref		string	快照引用，取metadata文件中refs中的分支名称或者标记名称

❗️ snapshot_id与ref只能取其一。

表元数据管理

设置快照过期时间

Iceberg中的每次write/update/delete/upsert/compaction都会生成一个新快照，同时保留旧数据和元数据，以便进行快照隔离和时间旅行。expire_snapshots过程可用于删除不再需要的旧快照及其文件。

这个过程将删除旧快照和那些旧快照唯一需要的数据文件。这意味着expire_snapshots过程永远不会删除未过期快照仍然需要的文件。

存储过程名 expire_snapshots

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
older_than	️	timestamp	超过该时限的快照将被删除 ，默认是: 5天前的被删除
retain_last		int	忽略 older_than作用， 而保留的快照数量 (defaults to 1)
max_concurrent_deletes		int	用于删除文件操作的线程池大小（默认情况下不使用线程池）
stream_results		boolean	为true时，删除文件将按 RDD 分区发送给 Spark 驱动程序（默认情况下，所有文件都将发送给 Spark 驱动程序）。建议将该选项设置为 true，以防止 Spark 驱动程序因文件大小而发生 OOM。
snapshot_ids		array of long	要过期的快照 ID 数组

如果省略 older_than 和 retain_last，则将使用表的expiration properties。仍被分支或标记引用的快照不会被删除。默认情况下，分支和标记永不过期，但可以使用表属性 history.expire.max-ref-age-ms 更改其保留策略。main分支永不过期。

❗️ 使用此存储过程时，必须增加stream_results且值为true。

清除孤岛文件

用于删除未在iceberg表的任何元数据文件中引用的文件，因此可视为 “孤岛”。

存储过程名 remove_orphan_files

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
older_than	️	timestamp	删除在此时间戳之前创建的孤岛文件 (Defaults to 3 days ago)
location		string	查找文件的目录 (defaults to the table’s location)
dry_run		boolean	预执行，若值为true，实际并未删除文件 (defaults to false)
max_concurrent_deletes		int	用于删除文件操作的线程池大小（默认情况下不使用线程池）
file_list_view		string	查找文件的数据集（跳过目录列表）
equal_schemes		map	被视为相同的文件系统方案的映射。键是一个以逗号分隔的方案列表，值是一个方案 (defaults to map('s3a,s3n','s3')).
equal_authorities		map	被视为等同的文件系统权限映射。键是以逗号分隔的权限列表，值是权限。
prefix_mismatch_mode		string	位置前缀（方案/授权）不匹配时的操作行为: ERROR- 抛出异常. (default) IGNORE - 啥也不干. DELETE - 删除文件.

重写数据文件

Iceberg 在一个表格中跟踪每个数据文件。数据文件越多，存储在清单文件中的元数据也就越多，而数据文件过小则会导致不必要的元数据量和文件打开成本，从而降低查询效率。

Iceberg 可以使用 Spark 的 rewriteDataFiles 操作并行压缩数据文件。这将把小文件合并为大文件，以减少元数据开销和运行时文件打开成本。

存储过程名 rewrite_data_files

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
strategy		string	策略名称 - 二进制包或排序。默认为 binpack 策略
sort_order		string	对于 Zorder，请在 zorder() 中使用逗号分隔的列列表。例如：zorder(c1,c2,c3)。否则，以逗号分隔的排序顺序格式为（ColumnName SortDirection NullOrder）。其中，SortDirection 可以是 ASC 或 DESC。NullOrder 可以是 NULLS FIRST 或 NULLS LAST。默认为表格的排序顺序
options	️	map	支持一些配置项
where	️	string	用于筛选文件的字符串。请注意，所有可能包含匹配过滤器数据的文件都会被选中进行重写

运用参数示例

spark.sql("CALL catalog_name.system.rewrite_data_files(table => 'db.sample', options => map('min-input-files','2'))");

options

General Options

属性名称	Default Value	备注
max-concurrent-file-group-rewrites	5	同时重写的最大文件组数
partial-progress.enabled	false	是否“分步提交”，允许在完成整个重写之前提交文件组
partial-progress.max-commits	10	如果启用了部分进度，允许此重写产生的最大提交次数
use-starting-sequence-number	true	使用压缩开始时快照的序列号，而不是新生成快照的序列号
rewrite-job-order	none
target-file-size-bytes	536870912 (512 MB, default value of write.target-file-size-bytes from table properties)	重写输出文件大小
min-file-size-bytes	75% of target file size	低于此阈值的文件将被考虑重写，而不考虑任何其他标准
max-file-size-bytes	180% of target file size	文件大小超过此阈值时，将考虑重写，而不考虑任何其他标准
min-input-files	5	任何文件组的文件数超过这个数量，都将被重写，而不考虑其他标准
rewrite-all	false	强制重写所有提供的文件，优先于其他选项
max-file-group-size-bytes	107374182400 (100GB)	单个文件组中应重写的最大数据量。整个重写操作会根据分区和分区内文件组的大小分成若干块。这有助于分解超大分区的重写，否则由于群集的资源限制，这些分区可能无法重写。
delete-file-threshold	2147483647	考虑重写数据文件所需的最少删除次数

Options for sort strategy

属性名称	Default Value	备注
compression-factor	1.0	shuffle 分区的数量以及 Spark 排序创建的输出文件数量取决于文件重写器使用的输入数据文件的大小。由于压缩，磁盘文件大小可能无法准确代表输出文件的大小。该参数允许用户调整用于估算实际输出数据大小的文件大小。系数大于 1.0 会生成比我们根据磁盘文件大小所预期的更多的文件。如果数值小于 1.0，根据磁盘上的文件大小，生成的文件将比我们预期的少
shuffle-partitions-per-file	1	每个输出文件要使用的分区数量。iceberg会使用自定义的合并操作，将这些已排序的分区拼接成一个已排序的文件。

Options for sort strategy with zorder sort_order

属性名称	Default Value	备注
var-length-contribution	8	从长度可变的输入列（字符串、二进制）中考虑的字节数
max-output-size	2147483647	ZOrder 算法中的字节交错量

重写清单文件

重写表的清单，优化扫描规划。

清单中的数据文件按分区规范中的字段排序。该程序使用 Spark 作业并行运行。

存储过程名 rewrite_manifests

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
use_caching	️	boolean	在运行过程中使用 Spark 缓存（默认为 true）
spec_id	️	int	要重写的清单的规格 ID（默认为当前规格 ID）

重写位置删除文件

Iceberg 可以重写位置删除文件，这样做有两个目的：

• 小型压缩： 将小的位置删除文件压缩成大文件。这样可以减少存储在清单文件中的元数据大小，并减少打开小的删除文件的开销。
• 删除悬而未决的删除记录： 过滤掉引用不再有效的数据文件的位置删除记录。重写数据文件后，指向重写数据文件的位置删除记录并不总是被标记为删除，而是会继续被表的实时快照元数据跟踪。这就是所谓的 “悬空删除 ”问题。

存储过程名 rewrite_position_delete_files

参数名称	是否必填	参数类型	备注
table	✔️	string	表名
options	️	map	支持一些配置项

在重写过程中，悬挂删除总是会被过滤掉。

Options

属性名称	Default Value	备注
max-concurrent-file-group-rewrites	5	同时重写的最大文件组数
partial-progress.enabled	false	是否“分步提交”，允许在完成整个重写之前提交文件组
partial-progress.max-commits	10	如果启用了部分进度，允许此重写产生的最大提交次数
rewrite-job-order	none
target-file-size-bytes	67108864 (64MB, default value of write.delete.target-file-size-bytes from table properties)	重写输出文件大小
min-file-size-bytes	75% of target file size	低于此阈值的文件将被考虑重写，而不考虑任何其他标准
max-file-size-bytes	180% of target file size	文件大小超过此阈值时，将考虑重写，而不考虑任何其他标准
min-input-files	5	任何文件组的文件数超过这个数量，都将被重写，而不考虑其他标准
rewrite-all	false	强制重写所有提供的文件，优先于其他选项
max-file-group-size-bytes	107374182400 (100GB)	单个文件组中应重写的最大数据量。整个重写操作会根据分区和分区内文件组的大小分成若干块。这有助于分解超大分区的重写，否则由于群集的资源限制，这些分区可能无法重写。