【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.0shuffle 分区的数量以及 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

2147483647ZOrder 算法中的字节交错量

重写清单文件

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

清单中的数据文件按分区规范中的字段排序。该程序使用 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)单个文件组中应重写的最大数据量。整个重写操作会根据分区和分区内文件组的大小分成若干块。这有助于分解超大分区的重写，否则由于群集的资源限制，这些分区可能无法重写。