# Iceberg表存储过程管理 在Spark中使用Iceberg前，需先配置Spark目录，spark_catalog和prod目录配置请参见[Iceberg配置与类型](https://support.huaweicloud.com/devg-aidatalake/aidatalake_05_0020.html)。存储过程仅在Spark 3中使用Iceberg SQL扩展时可用。 #### 使用方法 可通过**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]。 表6expire_snapshots参数介绍 | 参数 | 类型 | 说明 | |:---|:---|:---| | table | string | 需要更新的表名，必须设置。 | | older_than | timestamp | 此时间戳之前的快照将被移除，默认值为5天前。 | | retain_last | int | 保留的祖先快照数量，默认值为1，与"older_than"设置的时间戳无关。 | | max_concurrent_deletes | int | 用于删除文件操作的线程池大小，默认情况下，不使用线程池。 | | stream_results | boolean | 当设置该参数为"true"时，删除文件将通过RDD分区发送到Spark驱动程序（默认情况下，所有文件都将发送到Spark驱动程序）。 建议将该参数设置为"true"，以防止Spark驱动程序因文件过大而发生内存溢出。 | | snapshot_ids | array of long | 要过期的快照ID数组。 | | clean_expired_metadata | boolean | 当设置为true时，将清理不再被快照（snapshots）引用的元数据，例如分区规范（partition specs）和模式（schemas）。 | 如果不设置**older_than** 和**retain_last** ，将使用表的过期属性，仍被分支或标签引用的快照不会被移除。默认情况下，分支和标签永不过期，但可通过表属性**history.expire.max-ref-age-ms**更改其保留策略，且主分支永不过期。 输出结果介绍请参见[表7]。 表7expire_snapshots操作输出结果 | 参数 | 类型 | 说明 | |:---|:---|:---| | deleted_data_files_count | long | 此操作删除的数据文件数量。 | | deleted_position_delete_files_count | long | 此操作删除的**位置删除文件**数量。 | | deleted_equality_delete_files_count | long | 此操作删除的**等值删除文件**数量。 | | deleted_manifest_files_count | long | 此操作删除的清单文件数量。 | | deleted_manifest_lists_count | long | 此操作删除的清单列表文件数量。 | | deleted_statistics_files_count | long | 此操作删除的统计文件（statistics files）数量。 | #### remove_orphan_files 用于移除未在Iceberg表的任何元数据文件中引用的文件，这些文件可被视为"孤立文件"。 | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 要清理的表名。 | | older_than | 否 | timestamp | 移除在此时间戳之前创建的孤立文件（默认值：3天前）。 | | location | 否 | string | 要查找文件的目录（默认为表的存储位置） | | dry_run | 否 | boolean | 设为true时，不实际删除文件（默认值为false）。 | | max_concurrent_deletes | 否 | int | 用于删除文件操作的线程池大小（默认情况下，不使用线程池）。 | | file_list_view | 否 | string | 用于查找文件的数据集（跳过目录列表）。 | | equal_schemes | 否 | map | 被视为相等的文件系统方案映射。键是用逗号分隔的方案列表，值是一个方案（默认值为map('s3a,s3n','s3')）。 | | equal_authorities | 否 | map | 被视为相等的文件系统权限映射。键是用逗号分隔的权限列表，值是一个权限。 | | prefix_mismatch_mode | 否 | string | 当位置前缀（方案/权限）不匹配时的行为： - ERROR：抛出异常（默认值）。 - IGNORE：不执行任何操作DELETE-删除文件。 | | prefix_listing | 否 | boolean | 当设置为true时，将通过SupportsPrefixOperations 接口使用基于前缀的文件列举（prefix-based file listing）。启用此标志时，表的FileIO实现必须支持SupportsPrefixOperations接口（默认值为false）。 | 输出结果 | 输出名称 | 类型 | 描述 | |:---|:---|:---| | orphan_file_location | String | 此命令确定为孤立文件的每个文件的路径。 | **示例** 对表执行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 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并行压缩数据文件。此操作会将小文件合并为大文件，以减少元数据开销和运行时的文件打开成本。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 要更新的表名。 | | strategy | 否 | string | 策略名称，可选binpack（默认）或sort。 | | sort_order | 否 | string | 排序方式： - 若使用Zorder，格式为zorder(c1,c2,c3)（逗号分隔的列名列表）。 - 其他情况，格式为逗号分隔的排序规则(ColumnName SortDirection NullOrder)，其中SortDirection可为ASC或DESC，NullOrder可为NULLSFIRST或NULLSLAST默认使用表的排序规则。 | | options | 否 | map | 用于操作的选项参数。 | | where | 否 | string | 作为字符串的过滤条件，用于筛选需要重写的文件。注意：所有可能包含与过滤条件匹配数据的文件都将被选中进行重写。 | **配置选项** | 名称 | 默认值 | 描述 | |:---|:---|:---| | max-concurrent-file-group-rewrites | 5 | 同时重写的文件组最大数量。 | | partial-progress.enabled | false | 启用在整个重写完成前提交文件组。 | | partial-progress.max-commits | 10 | 若启用部分进度，此重写操作允许的最大提交次数。 | | partial-progress.max-failed-commits | partial-progress.max-commits的值 | 若启用部分进度，作业失败前允许的最大失败提交次数。 | | use-starting-sequence-number | true | 使用压缩开始时快照的序列号，而非新生成快照的序列号。 | | rewrite-job-order | none | 强制重写作业的顺序： - bytes-asc：优先重写最小的作业组。 - bytes-desc：优先重写最大的作业组。 - files-asc：优先重写文件数量最少的作业组。 - files-desc：优先重写文件数量最多的作业组。 - none：按计划顺序重写作业组（无特定排序）。 | | target-file-size-bytes | 536870912（512MB，默认值来自表属性write.target-file-size-bytes） | 目标输出文件大小。 | | min-file-size-bytes | 目标文件大小的75% | 低于此阈值的文件将被视为需要重写，无论其他条件如何。 | | max-file-size-bytes | 目标文件大小的180% | 大于此阈值的文件将被视为需要重写，无论其他条件如何。 | | min-input-files | 5 | 任何文件组的文件数量超过此值时，无论其他条件如何，都将被重写。 | | rewrite-all | false | 强制重写所有提供的文件，覆盖其他选项。 | | max-file-group-size-bytes | 107374182400（100GB） | 单个文件组中应重写的最大数据量。整个重写操作会根据分区拆分，在分区内再根据大小拆分为文件组。这有助于拆分大型分区的重写任务，避免因集群资源限制而无法重写。 | | delete-file-threshold | 2147483647 | 与数据文件关联的删除操作数量的最小值，达到此值时数据文件将被考虑重写。 | | delete-ratio-threshold | 0.3 | 数据文件被纳入重写（rewriting）范畴所需达到的最低删除比例（minimum deletion ratio)。 | | output-spec-id | 当前分区spec ID | 输出分区规范的标识符。重写过程中，数据将被重新组织以匹配输出分区方式。 | | remove-dangling-deletes | false | 重写后移除悬空的位置删除和等值删除文件。若删除文件不应用于任何活动数据文件，则视为悬空文件。启用此选项将额外生成一个提交来执行删除操作。 | **排序策略选项** | 名称 | 默认值 | 描述 | |:---|:---|:---| | compression-factor | 1.0 | Spark排序操作创建的shuffle分区数量（进而决定输出文件数量）基于此文件重写器中使用的输入数据文件大小。由于压缩，磁盘文件大小可能无法准确反映输出文件的实际大小。此参数允许用户调整用于估算实际输出数据大小的文件大小。大于1.0的值会生成比基于磁盘文件大小预期更多的文件；小于1.0的值会生成比基于磁盘大小预期更少的文件。 | | shuffle-partitions-per-file | 1 | 每个输出文件使用的shuffle分区数量。Iceberg会使用自定义的合并操作将这些排序后的分区重新拼接成一个单一的排序文件。 | **ZORDER排序顺序的排序策略选项** | 名称 | 默认值 | 描述 | |:---|:---|:---| | var-length-contribution | 8 | 从可变长度类型（String、Binary）的输入列中截取的字节数。 | | max-output-size | 2147483647 | ZOrder算法中间交错处理的字节量。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | rewritten_data_files_count | int | 此命令重写的数据文件数量。 | | added_data_files_count | int | 此命令写入的新数据文件数量。 | | rewritten_bytes_count | long | 此命令写入的字节数。 | | failed_data_files_count | int | 当partial-progress.enabled为true时，重写失败的数据文件数量。 | | removed_delete_files_count | int | 此命令删除的文件数。 | **示例** 使用默认的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作业并行执行。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 待更新的表名。 | | use_caching | 否 | boolean | 操作期间是否使用Spark缓存（默认值为true）。 | | spec_id | 否 | int | 待重写清单文件对应的分区规范ID（默认值为当前分区规范ID）。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | rewritten_manifests_count | int | 此命令重写的清单文件数量。 | | added_manifests_count | int | 此命令写入的新清单文件数量。 | **示例** 重写表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后，指向被重写数据文件的位置删除记录未必会被标记为待移除，可能仍会被表的活跃快照元数据追踪，这一现象被称为"悬空删除"问题。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 待更新的表名。 | | options | 否 | map | 用于此过程的配置选项。 | | where | 否 | string | 用于过滤文件的字符串形式的过滤条件。 | **配置选项** | 名称 | 默认值 | 描述 | |:---|:---|:---| | max-concurrent-file-group-rewrites | 5 | 可同时重写的文件组最大数量。 | | partial-progress.enabled | false | 启用"部分进度提交"，允许在整个重写操作完成前提交部分文件组。 | | partial-progress.max-commits | 10 | 若启用部分进度，此重写操作允许的最大提交次数。 | | rewrite-job-order | none | 强制指定重写作业的执行顺序： - bytes-asc：优先重写数据量最小的文件组。 - bytes-desc：优先重写数据量最大的文件组。 - files-asc：优先重写文件数量最少的文件组。 - files-desc：优先重写文件数量最多的文件组。 - none：按作业规划顺序重写（无特定排序）。 | | target-file-size-bytes | 67108864（64MB，默认值来自表属性write.delete.target-file-size-bytes） | 输出文件的目标大小。 | | min-file-size-bytes | 目标文件大小的75% | 低于此阈值的文件，无论其他条件如何，均会被纳入重写范围。 | | max-file-size-bytes | 目标文件大小的180% | 超过此阈值的文件，无论其他条件如何，均会被纳入重写范围。 | | min-input-files | 5 | 任何文件组的文件数量超过此值时，无论其他条件如何，均会被重写。 | | rewrite-all | false | 强制重写所有提供的文件，覆盖其他筛选条件。 | | max-file-group-size-bytes | 107374182400（100GB） | 单个文件组中可重写的最大数据量。整个重写操作会按分区拆分，分区内再按大小拆分为文件组，避免因集群资源限制导致大型分区无法重写。 | | max-files-to-rewrite | null | 此选项设置了将被重写的符合条件的文件数量上限。如果未指定此选项，所有符合条件的文件都将被重写。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | rewritten_delete_files_count | int | 此命令移除的旧位置删除文件数量。 | | added_delete_files_count | int | 此命令新增的位置删除文件数量。 | | rewritten_bytes_count | long | 此命令移除的所有旧位置删除文件的总字节数。 | | added_bytes_count | long | 此命令新增的所有位置删除文件的总字节数。 | **示例** 重写表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语句清理该表。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | source_table | 是 | string | 待生成快照的源表名。 | | table | 是 | string | 要创建的新Iceberg快照表名。 | | location | 否 | string | 新快照表的存储位置（默认由目录自动分配，目前在AI DataLake表迁移场景下，禁止设置该参数）。 | | properties | 否 | map | 要添加到新快照表的属性（如表级配置）。 | | parallelism | 否 | int | 用于读取文件的线程数（默认值为1）。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | imported_files_count | long | 添加到新快照表的文件数量（即引用的源表数据文件数）。 | **示例** 创建隔离的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_。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 待迁移的源表名。 | | properties | 否 | map | 要添加到新Iceberg表的属性（会覆盖源表中同名属性）。 | | drop_backup | 否 | boolean | 设为true时，不会保留原始表作为备份（默认值为false）。 | | backup_table_name | 否 | string | 保留的原始备份表名称（默认值为源表名_BACKUP_）。 | | parallelism | 否 | int | 用于读取文件的线程数（默认值为1）。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | migrated_files_count | long | 追加到新Iceberg表中的数据文件数量。 | **示例** 迁移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实现需求，不建议使用此方法。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 待添加文件的目标Iceberg表名。 | | source_table | 是 | string | 文件来源表（可为Hive/Spark注册的表，或格式为file_format.path的文件路径，如parquet./data/path）。 | | partition_filter | 否 | map | 用于筛选源表分区的映射（仅导入符合筛选条件的分区文件）。 | | check_duplicate_files | 否 | boolean | 是否防止已存在于目标表的文件被重复添加（默认值为true）。 | | parallelism | 否 | int | 用于读取文件的线程数（默认值为1）。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | added_files_count | long | 此命令成功添加的文件数量。 | | changed_partition_count | long | 受影响的分区数量（若可统计）。 | **示例** 从指定分区导入文件：将会话目录中注册的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元数据文件在指定目录中注册为正式的表，使其能被目录识别和管理。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 待注册的表名。 | | metadata_file | 是 | string | 待注册的metadata.json文件路径。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | current_snapshot_id | long | 新注册Iceberg表的当前快照ID。 | | total_records_count | long | 新注册Iceberg表的总记录数。 | | total_data_files_count | long | 新注册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，即返回该快照的祖先快照链。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 需查询祖先快照的表名。 | | snapshot_id | 否 | long | 用于查询父快照活跃ID的指定快照ID。 | **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | snapshot_id | long | 祖先快照的ID。 | | timestamp | long | 该祖先快照的创建时间。 | **示例** 查询当前快照的所有祖先快照（默认行为）： ``` 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** 创建一个包含指定表变更记录的视图，可通过该视图追踪表在不同快照或时间戳之间的所有数据变更。 **使用方法** | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 生成变更记录的源表名。 | | changelog_view | 否 | string | 要创建的变更视图名称（若不指定，默认生成与源表关联的视图名）。 | | options | 否 | map | Spark读取选项映射，用于指定变更记录的时间范围（常用选项见下文）。 | | net_changes | 否 | boolean | 是否仅输出"净变更"（移除中间变更，仅保留最终结果），默认值为false。当compute_updates=true时，此参数必须为false。 | | compute_updates | 否 | boolean | 是否计算"更新前后镜像"（将删除+插入组合识别为更新），默认规则：-若提供identifier_columns，默认值为true；-若未提供，默认值为false。 | | identifier_columns | 否 | array | 用于识别"更新操作"的标识列列表（如array('id','name')）。若compute_updates=true且未指定此参数，则使用表的当前标识字段。 | 通过options参数可限定变更记录的时间范围，支持两种维度（快照ID或时间戳），若不指定则默认覆盖表的全量快照： - start-snapshot-id：排他性起始快照ID（不包含此快照的变更），默认从表的第一个快照开始（包含）； - end-snapshot-id：包含性结束快照ID（包含此快照的变更），默认使用表的当前快照； - start-timestamp：排他性起始时间戳（毫秒级），默认从表的第一个快照开始（包含）； - end-timestamp：包含性结束时间戳（毫秒级），默认使用表的当前快照。 **输出结果** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | changelog_view | string | 已创建的变更视图名称。 | **示例** 创建视图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条记录。 | id | name | _change_type | _change_ordinal | _commit_snapshot_id | |:---|:---|:---|:---|:---| | 1 | Alice | INSERT | 0 | 5390529835796506035 | | 2 | Bob | INSERT | 0 | 5390529835796506035 | | 1 | Alice | DELETE | 1 | 8764748981452218370 | **Net Changes** 此过程可移除多快照间的中间变更，仅输出净变更。以下是创建计算净变更的变更日志视图的示例： ``` CALL spark_catalog.system.create_changelog_view( table => 'db.tbl', options => map('end-snapshot-id', '87647489814522183702'), net_changes => true ); ``` 由于Alice在第一个快照中被插入、在第二个快照中被删除，因此上述变更日志视图在启用净变更后，仅包含以下行： | id | name | _change_type | _change_ordinal | _commit_snapshot_id | |:---|:---|:---|:---|:---| | 2 | Bob | INSERT | 0 | 5390529835796506035 | **Carry-over Rows** 此过程默认移除结转行。结转行是采用写时复制模式时，执行行级操作（MERGE、UPDATE、DELETE）产生的结果。例如，某文件包含行1（id=1,name='Alice'）和行2（id=2,name='Bob'），采用写时复制模式删除行2时，需删除该文件并将行1保留到新文件中。尽管这并非对表的实际变更，但变更日志表仍会记录以下两行： | id | name | _change_type | |:---|:---|:---| | 1 | Alice | DELETE | | 1 | Alice | INSERT | 若需查看结转行，可按以下方式查询Spark变更表（SparkChangelogTable）： ``` SELECT * FROM spark_catalog.db.tbl.changes; ``` #### Pre/Post Update Images 若进行相关配置，此过程可计算更新前后镜像。更新前后镜像由一组删除行和插入行转换而来。标识列用于判断一条插入记录与一条删除记录是否指向同一行：若两条记录的标识列值相同，则视为同一行的前后状态。您可在表架构中设置标识字段，也可将其作为过程参数传入。 以下示例展示了使用标识列（id）计算更新前后镜像的过程：标识列值相同的行删除和行插入操作，会被视为单个更新操作。具体而言，假设存在以下两行记录： | id | name | _change_type | |:---|:---|:---| | 3 | Robert | DELETE | | 3 | Dan | INSERT | 在此情况下，该过程会将更新前的行标记为UPDATE_BEFORE镜像，将更新后的行标记为UPDATE_AFTER镜像，最终生成以下更新前后镜像： | id | name | _change_type | |:---|:---|:---| | 3 | Robert | UPDATE_BEFORE | | 3 | Dan | UPDATE_AFTER | #### 表统计信息 **compute_table_stats** 此过程用于计算特定表的不同值数量（NDV）统计信息。默认情况下，统计信息使用表当前快照为所有列计算。该过程可选择配置为计算特定快照和/或列子集的统计信息。 | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 表名。 | | snapshot_id | 否 | string | 要收集统计信息的快照ID。 | | columns | 否 | array | 要收集统计信息的列。 | **输出** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | statistics_file | string | 此命令创建的统计信息文件路径。 | **示例** 收集表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注册到表元数据中。 | 参数名称 | 是否必填 | 类型 | 描述 | |:---|:---|:---|:---| | table | 是 | string | 表名。 | | snapshot_id | 否 | string | 要计算分区统计信息的快照ID。默认为当前快照ID。 | **输出** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | partition_statistics_file | string | 此命令创建的分区统计信息文件路径。 | **示例** 收集表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表到新位置的起点。 此过程仅暂存重写的元数据文件并准备要复制的文件列表。实际的文件复制不包含在此过程中。 | 参数名称 | 是否必填 | 默认值 | 类型 | 描述 | |:---|:---|:---|:---|:---| | table | 是 | - | string | 表名。 | | source_prefix | 是 | - | string | 要替换的现有前缀。 | | target_prefix | 是 | - | string | source_prefix的替换前缀。 | | start_version | 否 | 表元数据日志中的第一个metadata.json | string | 要重写的按时间顺序第一个metadata.json的名称或路径。 | | end_version | 否 | 表元数据日志中的最新metadata.json | string | 要重写的按时间顺序最后一个metadata.json的名称或路径。 | | staging_location | 否 | 表元数据目录下的新目录 | string | 新重写的元数据文件的输出位置。 | **操作模式** - **完整重写**：完整重写将重写所有可访问的元数据文件（包括metadata.json、manifest lists、manifests 和 position delete files）， 并在 file_list_location 中返回所有可访问的文件。这是此过程的默认操作模式。 - **增量重写**：可选地，可以提供start_version和end_version来将范围限制为增量重写。增量重写仅重写在start_version和end_version之间添加的元数据文件，并且仅在file_list_location中返回此范围内添加的文件。 **输出** | 输出名称 | 类型 | 描述 | |:---|:---|:---| | latest_version | string | 此过程重写的最新元数据文件名称。 | | file_list_location | string | 包含源路径到目标路径映射的CSV文件路径。 | **文件列表** 该文件包含在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中。