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✔️longmetadata文件中snapshots中的快照子项snapshot-id的值 根据timestamp举行回滚
存储过程名 rollback_to_timestamp
参数名称是否必填参数类型备注table✔️string表名timestamp✔️timestampmetadata文件中snapshots中的快照子项timestamp-ms的值 设置表当前见效的快照
存储过程名 set_current_snapshot
参数名称是否必填参数类型备注table✔️string表名snapshot_idlong取metadata文件中snapshots中的快照子项snapshot-id的值refstring快照引用，取metadata文件中refs中的分支名称或者标记名称 ❗️ snapshot_id与ref只能取其一。
表元数据管理
设置快照过期时间
Iceberg中的每次write/update/delete/upsert/compaction都会天生一个新快照，同时保留旧数据和元数据，以便举行快照隔离和时间观光。expire_snapshots过程可用于删除不再必要的旧快照及其文件。
这个过程将删除旧快照和那些旧快照唯一必要的数据文件。这意味着expire_snapshots过程永远不会删除未过期快照仍然必要的文件。
存储过程名 expire_snapshots
参数名称是否必填参数类型备注table✔️string表名older_than️timestamp超过该时限的快照将被删除 ，默认是: 5天前的被删除retain_lastint忽略 older_than作用， 而保留的快照数量 (defaults to 1)max_concurrent_deletesint用于删除文件操纵的线程池大小（默认情况下不使用线程池）stream_resultsboolean为true时，删除文件将按 RDD 分区发送给 Spark 驱动程序（默认情况下，全部文件都将发送给 Spark 驱动程序）。建议将该选项设置为 true，以防止 Spark 驱动程序因文件大小而发生 OOM。snapshot_idsarray 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)locationstring查找文件的目录 (defaults to the table’s location)dry_runboolean预执行，若值为true，实际并未删除文件 (defaults to false)max_concurrent_deletesint用于删除文件操纵的线程池大小（默认情况下不使用线程池）file_list_viewstring查找文件的数据集（跳过目录列表）equal_schemesmap被视为雷同的文件系统方案的映射。键是一个以逗号分隔的方案列表，值是一个方案 (defaults to map('s3a,s3n','s3')).equal_authoritiesmap被视为等同的文件系统权限映射。键是以逗号分隔的权限列表，值是权限。prefix_mismatch_modestring位置前缀（方案/授权）不匹配时的操纵行为:
ERROR- 抛出异常. (default)
IGNORE - 啥也不干.
DELETE - 删除文件. 重写数据文件
Iceberg 在一个表格中跟踪每个数据文件。数据文件越多，存储在清单文件中的元数据也就越多，而数据文件过小则会导致不必要的元数据量和文件打开资本，从而降低查询效率。
Iceberg 可以使用 Spark 的 rewriteDataFiles 操纵并行压缩数据文件。这将把小文件归并为大文件，以减少元数据开销和运行时文件打开资本。
存储过程名 rewrite_data_files
参数名称是否必填参数类型备注table✔️string表名strategystring计谋名称 - 二进制包或排序。默以为 binpack 计谋sort_orderstring对于 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'))");
General Options
属性名称Default Value备注max-concurrent-file-group-rewrites5同时重写的最大文件组数partial-progress.enabledfalse是否“分步提交”，允许在完成整个重写之前提交文件组partial-progress.max-commits10如果启用了部分进度，允许此重写产生的最大提交次数use-starting-sequence-numbertrue使用压缩开始时快照的序列号，而不是新天生快照的序列号rewrite-job-ordernonetarget-file-size-bytes536870912 (512 MB, default value of write.target-file-size-bytes from table properties)重写输出文件大小min-file-size-bytes75% of target file size低于此阈值的文件将被思量重写，而不思量任何其他标准max-file-size-bytes180% of target file size文件大小超过此阈值时，将思量重写，而不思量任何其他标准min-input-files5任何文件组的文件数超过这个数量，都将被重写，而不思量其他标准rewrite-allfalse逼迫重写全部提供的文件，优先于其他选项max-file-group-size-bytes107374182400 (100GB)单个文件组中应重写的最大数据量。整个重写操纵会根据分区和分区内文件组的大小分成多少块。这有助于分解超大分区的重写，否则由于群集的资源限制，这些分区大概无法重写。delete-file-threshold2147483647思量重写数据文件所需的最少删除次数 Options for sort strategy
属性名称Default Value备注compression-factor1.0shuffle 分区的数量以及 Spark 排序创建的输出文件数量取决于文件重写器使用的输入数据文件的大小。由于压缩，磁盘文件大小大概无法准确代表输出文件的大小。该参数允许用户调解用于估算实际输出数据大小的文件大小。系数大于 1.0 会天生比我们根据磁盘文件大小所预期的更多的文件。如果数值小于 1.0，根据磁盘上的文件大小，天生的文件将比我们预期的少shuffle-partitions-per-file1每个输出文件要使用的分区数量。iceberg会使用自界说的归并操纵，将这些已排序的分区拼接成一个已排序的文件。 Options for sort strategy with zorder sort_order
属性名称Default Value备注var-length-contribution8从长度可变的输入列（字符串、二进制）中思量的字节数max-output-size2147483647ZOrder 算法中的字节交错量 重写清单文件
重写表的清单，优化扫描规划。
清单中的数据文件按分区规范中的字段排序。该程序使用 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-rewrites5同时重写的最大文件组数partial-progress.enabledfalse是否“分步提交”，允许在完成整个重写之前提交文件组partial-progress.max-commits10如果启用了部分进度，允许此重写产生的最大提交次数rewrite-job-ordernonetarget-file-size-bytes67108864 (64MB, default value of write.delete.target-file-size-bytes from table properties)重写输出文件大小min-file-size-bytes75% of target file size低于此阈值的文件将被思量重写，而不思量任何其他标准max-file-size-bytes180% of target file size文件大小超过此阈值时，将思量重写，而不思量任何其他标准min-input-files5任何文件组的文件数超过这个数量，都将被重写，而不思量其他标准rewrite-allfalse逼迫重写全部提供的文件，优先于其他选项max-file-group-size-bytes107374182400 (100GB)单个文件组中应重写的最大数据量。整个重写操纵会根据分区和分区内文件组的大小分成多少块。这有助于分解超大分区的重写，否则由于群集的资源限制，这些分区大概无法重写。
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
