Iceberg表存储过程管理

使用方法

可通过CALL语句从任何已配置的Iceberg Catalog中调用存储过程，所有存储过程均位于system命名空间下。

CALL支持按名称传递参数（推荐方式）或按位置传递参数，且不支持混合使用位置参数和命名参数。

• 命名参数

  所有存储过程参数都有名称。按名称传递参数时，参数顺序可以任意，且任何可选参数都可省略。

  CALL spark_catalog.system.{存储名} (arg_name_2 => 'arg_2_value', arg_name_1 => 'arg_1_value');

• 位置参数
  按位置传递参数时，仅当末尾的参数为可选参数时，才可省略。

  CALL spark_catalog.system.{存储名} (arg_1, arg_2, ... arg_n);

快照管理

• rollback_to_snapshot

  可使用rollback_to_snapshot将表回滚到特定的快照ID。以下命令为将表db.sample回滚到快照ID为1的版本：

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

  其中，表名字段的类型为string，快照ID字段的类型为long。

  输出结果介绍请参见表1。

  表1 回滚输出结果

  参数	类型	说明
  previous_snapshot_id	long	回滚前的当前快照ID。
  current_snapshot_id	long	回滚后的新当前快照 ID。

• rollback_to_timestamp

  可使用rollback_to_timestamp将表回滚到特定时间点对应的活跃快照。以下命令为将db.sample表回滚到特定日期和时间：

  CALL spark_catalog.system.rollback_to_timestamp('db.sample', TIMESTAMP '2021-06-30 00:00:00.000');

  其中，表名字段的类型为string，timestamp的类型为timestamp，表示回滚目标时间戳。

  输出结果介绍请参见表1。

• set_current_snapshot

  可使用set_current_snapshot设置表的当前快照ID。与回滚不同，该操作不要求目标快照必须是表当前状态的祖先快照。例如：

  将db.sample表的当前快照设置为ID为1的快照：

  CALL spark_catalog.system.set_current_snapshot('db.sample', 1);

  将db.sample表的当前快照设置为标签s1对应的快照：

  CALL spark_catalog.system.set_current_snapshot(table => 'db.sample', ref => 's1');

  其中，表名字段的类型为string，快照ID字段的类型为long，“ref”为设置当前快照的快照引用（分支或标签），类型为string。且必须设置快照ID或ref，二者不可同时提供，也不可均不提供。

  输出结果介绍请参见表2。

  表2 快照设置输出结果

  参数	类型	说明
  previous_snapshot_id	long	设置前的当前快照ID。
  current_snapshot_id	long	设置后新的当前快照ID。

• cherrypick_snapshot

  可使用cherrypick_snapshot从指定快照中挑选变更并应用到表的当前状态。挑选操作会基于现有快照创建一个新快照，且不会修改或删除原始快照。仅支持对追加和动态覆盖类型的快照执行挑选操作。例如：

  挑选快照1的变更：

  CALL spark_catalog.system.cherrypick_snapshot('my_table', 1);

  使用命名参数挑选快照1的变更：

  CALL spark_catalog.system.cherrypick_snapshot(snapshot_id => 1, table => 'my_table' );

  其中，“table”表示待更新的表名，类型为string；“snapshot_id”为要设为当前快照的快照ID，类型为long。

  输出结果介绍请参见表3。

  表3 快照变更输出结果

  参数	类型	说明
  source_snapshot_id	long	设置前的当前快照ID。
  current_snapshot_id	long	设置后新的当前快照ID。

• publish_changes

  可使用publish_changes将暂存的WAP ID中的变更发布到表的当前状态。变更发布会基于现有快照创建一个新快照，且不会修改或删除原始快照。仅追加和动态覆盖类型的快照可成功发布。例如：

  使用WAP ID“wap_id_1”发布变更：

  CALL spark_catalog.system.publish_changes('my_table', 'wap_id_1');

  使用命名参数发布变更：

  CALL spark_catalog.system.publish_changes(wap_id => 'wap_id_2', table => 'my_table');

  其中，“table”表示待更新的表名，类型为string；“wap_id”为要从暂存环境发布的WAP ID，类型为long。

  输出结果介绍请参见表4。

  表4 快照变更输出结果

  参数	类型	说明
  source_snapshot_id	long	设置前的当前快照ID。
  current_snapshot_id	long	设置后新的当前快照ID。

• fast_forward

  可使用fast_forward将一个分支的当前快照快速更新到另一个分支的最新快照。例如：

  将main分支的当前快照快速更新到audit-branch分支的最新快照中：

  CALL spark_catalog.system.fast_forward('my_table', 'main', 'audit-branch');

  其中，表字段的类型为string，分支字段的类型为string。

  输出结果介绍请参见表5。

  表5 快照更新输出结果

  参数	类型	说明
  branch_updated	string	已完成快照快速更新的分支名称。
  previous_ref	long	应用快速更新前的快照ID。
  updated_ref	long	应用快速更新后的当前快照ID。

元数据管理

expire_snapshots

在Iceberg中，每次写入、更新、删除、upsert、压缩操作都会生成新快照，同时保留旧数据和元数据，以支持快照隔离和时间旅行功能。expire_snapshots可用于移除不再需要的旧快照及其文件。

expire_snapshots会移除旧快照以及这些旧快照唯一需要的数据文件，意味着该操作绝不会移除仍被非过期快照所需要的文件。例如：

• 移除特定日期和时间之前的快照，但保留最后100个快照：

  CALL spark_catalog.system.expire_snapshots('db.sample', TIMESTAMP '2021-06-30 00:00:00.000', 100);

• 移除快照ID为123的快照，且此快照ID不应是当前快照：

  CALL spark_catalog.system.expire_snapshots('db.sample', snapshot_ids => array(123));

相关参数介绍请参见表6。

如果不设置older_than和retain_last，将使用表的过期属性，仍被分支或标签引用的快照不会被移除。默认情况下，分支和标签永不过期，但可通过表属性history.expire.max-ref-age-ms更改其保留策略，且主分支永不过期。

输出结果介绍请参见表7。

remove_orphan_files

用于移除未在Iceberg表的任何元数据文件中引用的文件，这些文件可被视为“孤立文件”。

输出结果

示例

对表执行remove_orphan_files命令的试运行，列出所有候选删除文件但不实际删除它们：

CALL spark_catalog.system.remove_orphan_files(table => 'db.sample', dry_run => true);

移除表db.sample的tablelocation/data文件夹中所有不被该表识别的文件：

CALL spark_catalog.system.remove_orphan_files(table => 'db.sample', location => 'tablelocation/data');

移除files_view视图中所有不被表db.sample识别的文件：

Dataset<Row> compareToFileList = spark.createDataFrame(allFiles, FilePathLastModifiedRecord.class).withColumnRenamed("filePath", "file_path").withColumnRenamed("lastModified", "last_modified");
String fileListViewName = "files_view";
compareToFileList.createOrReplaceTempView(fileListViewName);

CALL spark_catalog.system.remove_orphan_files(table => 'db.sample', file_list_view => 'files_view');

当文件除位置前缀（方案/权限）外与元数据文件中的引用匹配时，默认会抛出错误。通过将prefix_mismatch_mode设置为IGNORE，可忽略该错误并跳过文件：

CALL spark_catalog.system.remove_orphan_files(table => 'db.sample', prefix_mismatch_mode => 'IGNORE');

通过将prefix_mismatch_mode设置为DELETE，仍可删除该文件：

CALL spark_catalog.system.remove_orphan_files(table => 'db.sample', prefix_mismatch_mode => 'DELETE');

也可通过将不匹配的前缀视为相等来删除文件：

CALL spark_catalog.system.remove_orphan_files(table => 'db.sample', equal_schemes => map('file', 'file1'));

CALL spark_catalog.system.remove_orphan_files(table => 'db.sample', equal_authorities => map('ns1', 'ns2'));

列出所有符合删除条件的候选文件，并使用基于前缀的列表查询（prefix listing）方式进行扫描。

CALL catalog_name.system.remove_orphan_files(table => 'db.sample', prefix_listing => true);

rewrite_data_files

Iceberg会跟踪表中的每个数据文件。数据文件越多，清单文件中存储的元数据就越多；而小数据文件会产生不必要的元数据开销，并因文件打开成本导致查询效率降低。

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

使用方法

配置选项

排序策略选项

ZORDER排序顺序的排序策略选项

输出结果

示例

使用默认的bin-packing重写算法重写表db.sample中的数据文件，合并小文件并根据表的默认写入大小拆分大文件：

CALL spark_catalog.system.rewrite_data_files('db.sample');

通过按id和name对所有数据进行排序来重写表db.sample中的数据文件，使用与bin-pack相同的默认值来确定需要重写的文件：

CALL spark_catalog.system.rewrite_data_files(table => 'db.sample', strategy => 'sort', sort_order => 'id DESC NULLS LAST,name ASC NULLS FIRST');

通过对列c1和c2执行ZOrder排序来重写表db.sample中的数据文件，使用与bin-pack相同的默认值来确定需要重写的文件：

CALL spark_catalog.system.rewrite_data_files(table => 'db.sample', strategy => 'sort', sort_order => 'zorder(c1,c2)');

使用bin-pack策略重写表db.sample中任何至少有两个文件需要重写的分区中的数据文件，然后移除所有悬空的删除文件：

CALL spark_catalog.system.rewrite_data_files(table => 'db.sample', options => map('min-input-files', '2', 'remove-dangling-deletes', 'true'));

重写表db.sample中的数据文件，并选择可能包含与筛选条件（id=3且name="foo"）匹配的数据的文件进行重写：

CALL spark_catalog.system.rewrite_data_files(table => 'db.sample', where => 'id = 3 and name = "foo"');

rewrite_manifests

重写表的清单文件，以优化扫描规划效率。

清单文件中的数据文件会按分区规范中的字段进行排序。此过程通过Spark作业并行执行。

使用方法

输出结果

示例

重写表db.sample的清单文件，并使清单文件与表分区规则对齐：

CALL spark_catalog.system.rewrite_manifests('db.sample');

重写表db.sample的清单文件，并禁用Spark缓存（可用于避免执行器出现内存问题）：

CALL spark_catalog.system.rewrite_manifests('db.sample', false);

rewrite_position_delete_files

Iceberg可重写位置删除文件，该操作主要实现两大功能：

• 轻度压缩：将小型位置删除文件合并为大型文件，减少清单文件中存储的元数据体积，同时降低打开小型删除文件的开销。
• 移除悬空删除：过滤掉指向已不再活跃的数据文件的位置删除记录。执行rewrite_data_files后，指向被重写数据文件的位置删除记录未必会被标记为待移除，可能仍会被表的活跃快照元数据追踪，这一现象被称为“悬空删除”问题。

使用方法

配置选项

输出结果

示例

重写表db.sample的位置删除文件：按默认筛选条件选择待重写文件，生成符合target-file-size-bytes目标大小的新文件，且重写过程中自动移除悬空删除记录。

CALL spark_catalog.system.rewrite_position_delete_files('db.sample');

重写表db.sample的所有位置删除文件：强制覆盖筛选条件，生成符合目标大小的新文件，同时移除悬空删除记录。

CALL spark_catalog.system.rewrite_position_delete_files(table => 'db.sample', options => map('rewrite-all', 'true'));

按文件数量筛选重写表db.sample的位置删除文件：仅重写“按大小标准需重写的文件数≥2”的分区中的位置删除文件，同时移除悬空删除记录。

CALL spark_catalog.system.rewrite_position_delete_files(table => 'db.sample', options => map('min-input-files','2'));

表迁移

快照和迁移存储过程可帮助测试现有Hive或Spark表，并将其迁移至Iceberg表。

snapshot

创建源表的轻量级临时副本用于测试，且不会修改源表本身。

新创建的快照表可进行修改或写入操作，不会影响源表；但快照表会引用源表的原始数据文件。当对快照表执行插入或覆盖操作时，新生成的文件会存储在快照表的指定位置，而非源表的存储位置。

快照表测试完成后，可通过执行DROP TABLE语句清理该表。

使用方法

输出结果

示例

创建隔离的Iceberg快照表：引用源表db.sample，新表名为db.snap，存储位置使用目录为db.snap分配的默认位置。

CALL spark_catalog.system.snapshot('db.sample', 'db.snap');

migrate

将现有表替换为Iceberg表，并加载源表的数据文件。

源表的表结构、分区配置、表属性及存储位置会被复制到新的Iceberg表中。

若源表的任一分区使用了不支持的文件格式，迁移操作将失败。当前支持的文件格式为Avro、Parquet和ORC。源表的现有数据文件会被添加到Iceberg表的元数据中，且可通过基于源表结构生成的“名称-ID”映射关系进行读取。

若需在测试期间保持源表完好无损，可使用snapshot过程创建共享源表数据文件和结构的临时表。

默认情况下，原始表会被保留，并命名为table_BACKUP_。

使用方法

输出结果

示例

迁移Spark默认目录中的表并添加属性：将spark_catalog.db.sample迁移为Iceberg表，并为新表添加属性'foo'='bar'。

CALL spark_catalog.system.migrate('spark_catalog.db.sample', map('foo', 'bar'));

基础迁移（无额外属性）：将当前目录下的db.sample迁移为Iceberg表，不添加额外属性

CALL spark_catalog.system.migrate('db.sample');

add_files

尝试将Hive表或基于文件的表中的文件直接添加到指定的Iceberg表中。与迁移或快照不同，add_files可从特定分区导入文件，且不会创建新的Iceberg表。此命令会为待添加的文件生成元数据，但不会移动文件本身。该过程不会分析文件的schema以验证其是否与Iceberg表的schema匹配。操作完成后，Iceberg表会将这些文件视为自身管理的文件集一部分，这意味着后续执行过期快照时，可对这些添加的文件执行物理删除。若可使用migrate或snapshot实现需求，不建议使用此方法。

使用方法

输出结果

示例

从指定分区导入文件：将会话目录中注册的Hive/Spark表db.src_tbl中，part_col_1等于A的分区文件，添加到Iceberg表db.tbl。

CALL spark_catalog.system.add_files( table => 'db.tbl', source_table => 'db.src_tbl', partition_filter => map('part_col_1', 'A') );

从文件路径导入全量文件：将路径path/to/table下基于Parquet格式的表文件，全量添加到Iceberg表db.tbl（不筛选分区）。

CALL spark_catalog.system.add_files( table => 'db.tbl',   source_table => '`parquet`.`path/to/table`' );

register_table

为已存在但尚未关联目录标识符的metadata.json文件创建目录条目。通过此操作，可将独立的Iceberg元数据文件在指定目录中注册为正式的表，使其能被目录识别和管理。

使用方法

输出结果

示例

在spark_catalog目录中，将路径为path/to/metadata/file.json的metadata.json文件注册为表db.tbl：

CALL spark_catalog.system.register_table( table => 'db.tbl', metadata_file => 'path/to/metadata/file.json' );

元数据信息

ancestors_of

查询指定快照的父快照对应的活跃快照ID，即返回该快照的祖先快照链。

使用方法

输出结果

示例

查询当前快照的所有祖先快照（默认行为）：

CALL spark_catalog.system.ancestors_of('db.tbl');

通过指定快照ID查询其所有祖先快照（使用位置参数）：

CALL spark_catalog.system.ancestors_of('db.tbl', 1);

通过指定快照ID查询其所有祖先快照（使用命名参数，推荐，提升可读性）：

CALL spark_catalog.system.ancestors_of(snapshot_id => 1, table => 'db.tbl');

Change Data Capture

create_changelog_view

创建一个包含指定表变更记录的视图，可通过该视图追踪表在不同快照或时间戳之间的所有数据变更。

使用方法

通过options参数可限定变更记录的时间范围，支持两种维度（快照ID或时间戳），若不指定则默认覆盖表的全量快照：

• start-snapshot-id：排他性起始快照ID（不包含此快照的变更），默认从表的第一个快照开始（包含）；
• end-snapshot-id：包含性结束快照ID（包含此快照的变更），默认使用表的当前快照；
• start-timestamp：排他性起始时间戳（毫秒级），默认从表的第一个快照开始（包含）；
• end-timestamp：包含性结束时间戳（毫秒级），默认使用表的当前快照。

输出结果

示例

创建视图tbl_changes，包含源表db.tbl在快照1（排他）到快照2（包含）之间的变更：

CALL spark_catalog.system.create_changelog_view(  table => 'db.tbl',  options => map('start-snapshot-id','1','end-snapshot-id', '2'));

创建视图my_changelog_view，包含源表db.tbl在时间戳1678335750489（排他）到1678992105265（包含）之间的变更：

CALL spark_catalog.system.create_changelog_view(   table => 'db.tbl',   options => map('start-timestamp','1678335750489','end-timestamp', '1678992105265'),   changelog_view => 'my_changelog_view' );

创建视图，通过标识列id和name将“删除+插入”识别为更新操作，包含快照1到2之间的变更：

CALL spark_catalog.system.create_changelog_view(   table => 'db.tbl',   options => map('start-snapshot-id','1','end-snapshot-id', '2'),   identifier_columns => array('id', 'name') )

视图创建后，可直接通过SQL查询变更记录：

SELECT * FROM tbl_changes;

SELECT * FROM tbl_changes where _change_type = 'INSERT' AND id = 3 ORDER BY _change_ordinal;

请注意，变更日志视图包含变更数据捕获（CDC）元数据列，这些列提供所追踪变更的额外信息。具体列如下：

• _change_type：变更类型，取值范围为INSERT（插入）、DELETE（删除）、UPDATE_BEFORE（更新前）或UPDATE_AFTER（更新后）。
• _change_ordinal：变更顺序。
• _commit_snapshot_id：变更发生时所属的快照ID。

以下是对应的结果示例，该示例显示第一个快照插入2条记录，第二个快照删除了1条记录。

Net Changes

此过程可移除多快照间的中间变更，仅输出净变更。以下是创建计算净变更的变更日志视图的示例：

CALL spark_catalog.system.create_changelog_view(   table => 'db.tbl',   options => map('end-snapshot-id', '87647489814522183702'),   net_changes => true );

由于Alice在第一个快照中被插入、在第二个快照中被删除，因此上述变更日志视图在启用净变更后，仅包含以下行：

Carry-over Rows

此过程默认移除结转行。结转行是采用写时复制模式时，执行行级操作（MERGE、UPDATE、DELETE）产生的结果。例如，某文件包含行1（id=1,name='Alice'）和行2（id=2,name='Bob'），采用写时复制模式删除行2时，需删除该文件并将行1保留到新文件中。尽管这并非对表的实际变更，但变更日志表仍会记录以下两行：

若需查看结转行，可按以下方式查询Spark变更表（SparkChangelogTable）：

SELECT * FROM spark_catalog.db.tbl.changes;

Pre/Post Update Images

若进行相关配置，此过程可计算更新前后镜像。更新前后镜像由一组删除行和插入行转换而来。标识列用于判断一条插入记录与一条删除记录是否指向同一行：若两条记录的标识列值相同，则视为同一行的前后状态。您可在表架构中设置标识字段，也可将其作为过程参数传入。

以下示例展示了使用标识列（id）计算更新前后镜像的过程：标识列值相同的行删除和行插入操作，会被视为单个更新操作。具体而言，假设存在以下两行记录：

在此情况下，该过程会将更新前的行标记为UPDATE_BEFORE镜像，将更新后的行标记为UPDATE_AFTER镜像，最终生成以下更新前后镜像：

表统计信息

compute_table_stats

此过程用于计算特定表的不同值数量（NDV）统计信息。默认情况下，统计信息使用表当前快照为所有列计算。该过程可选择配置为计算特定快照和/或列子集的统计信息。

输出

示例

收集表my_table最新快照的统计信息

CALL catalog_name.system.compute_table_stats('my_table'); 

收集表my_table中ID为snap1的快照的统计信息

CALL catalog_name.system.compute_table_stats(table => 'my_table', snapshot_id => 'snap1' ); 

收集表my_table中ID为snap1的快照，针对列col1和col2的统计信息

CALL catalog_name.system.compute_table_stats(table => 'my_table', snapshot_id => 'snap1', columns => array('col1', 'col2')); 

分区统计信息

compute_partition_stats

此过程从最后一个具有PartitionStatisticsFile的快照开始增量计算分区统计信息，直到给定的快照（如果未指定则使用当前快照），并将计算结果写入PartitionStatisticsFile。如果之前的分区统计信息文件不存在，则执行完整计算。它还会将PartitionStatisticsFile注册到表元数据中。

输出

示例

收集表my_table最新快照的分区统计信息

CALL catalog_name.system.compute_partition_stats('my_table'); 

收集表my_table中ID为snap1的快照的分区统计信息

CALL catalog_name.system.compute_partition_stats(table => 'my_table', snapshot_id => 'snap1'); 

表复制

rewrite_table_path过程用于准备将Iceberg表复制到其他位置。

创建一个Iceberg表元数据文件的暂存副本，其中每个绝对路径源前缀都被替换为指定的目标前缀。这可以作为完全或增量复制Iceberg表到新位置的起点。

此过程仅暂存重写的元数据文件并准备要复制的文件列表。实际的文件复制不包含在此过程中。

操作模式

• 完整重写：完整重写将重写所有可访问的元数据文件（包括metadata.json、manifest lists、manifests 和 position delete files）， 并在 file_list_location 中返回所有可访问的文件。这是此过程的默认操作模式。
• 增量重写：可选地，可以提供start_version和end_version来将范围限制为增量重写。增量重写仅重写在start_version和end_version之间添加的元数据文件，并且仅在file_list_location中返回此范围内添加的文件。

输出

文件列表

该文件包含在start_version和end_version之间添加到表中的所有文件的复制计划。

对于每个文件，它指定：

• 源路径：表中的原始文件路径，如果文件已被重写，则为暂存位置
• 目标路径：带有替换前缀的路径

以下示例显示了三个文件的复制计划：

sourcepath/datafile1.parquet,targetpath/datafile1.parquet
sourcepath/datafile2.parquet,targetpath/datafile2.parquet 
stagingpath/manifest.avro,targetpath/manifest.avro 

示例

此示例完整重写my_table的元数据路径，从OBS中的源位置到OBS中的目标位置。它将在表元数据目录下的默认暂存位置生成一组新的元数据。

CALL catalog_name.system.rewrite_table_path(
     table => 'db.my_table',
     source_prefix => 'obs://bucket/path/to/source_table',
     target_prefix => 'obs://bucket/prefix/db.db/my_table' ); 

此示例增量重写my_table在元数据版本v2.metadata.json和v20.metadata.json之间的元数据路径，新元数据文件写入指定的暂存位置。

CALL catalog_name.system.rewrite_table_path(
     table => 'db.my_table',
     source_prefix => 'obs://bucketOne/prefix/db.db/my_table',
     target_prefix => 'obs://bucketTwo/prefix/db.db/my_table',
     start_version => 'v2.metadata.json',
     end_version => 'v20.metadata.json',
     staging_location => 'obs://bucketStaging/my_table' ); 

重写完成后，第三方工具（如DistCp）可以将新创建的元数据文件和数据文件复制到目标位置。

最后，可以使用注册表方法将复制的表注册到catalog中。