数据湖大数据数据存储【免费下载链接】icebergApache Iceberg项目地址https://gitcode.com/gh_mirrors/icebe/iceberg点击查看免费下载本文是一份面向数据工程师与平台架构师的 Apache Iceberg 表迁移技术指南围绕 Iceberg 官方文档中的 Table Migration 主线系统讲解全量数据迁移与原地元数据迁移两类路径的取舍并深入剖析原地迁移中三大核心动作 ——Snapshot Table快照表、Migrate Table迁移表与Add Files追加文件的原理、Spark SQL 调用方式、参数语义与底层实现。读完本文你将能够根据业务对停机时间、数据隔离性和历史保留的需求为 Hive、Delta Lake 等存量表制定并落地一套安全、可回滚的 Iceberg 迁移方案。一、两种迁移路径全量数据迁移 vs 原地元数据迁移Apache Iceberg 支持将其他格式的存量表转换为 Iceberg 表官方将迁移方式划分为两大类全量数据迁移Full Data Migration把源表的全部数据文件拷贝到一张全新的 Iceberg 表中。其最大优点是新表与源表完全隔离——后续对源表的任何清理、删除操作都不会影响新表代价是迁移速度慢且需要双倍存储空间。在实践中全量迁移通常借助以下手段完成Create-Table-As-SelectCTAS 语法INSERT INTO 语句各类 Change-Data-CaptureCDC同步管道。原地元数据迁移In-Place Metadata Migration保留源表现有数据文件不动仅在数据之上叠加 Iceberg 元数据metadata。这种方式速度更快且无需复制数据但新表与源表之间并不完全隔离——如果源表侧有任何进程对数据文件执行了清理vacuum新表也会连带受到影响。本指南的核心内容即围绕原地元数据迁移展开。Iceberg 的原地元数据迁移共包含三个重要动作Snapshot Table、Migrate Table与Add Files三者分别应对无停机探测性迁移原地接管替换与迁移后的增量补齐三类场景。二、Snapshot Table零停机创建独立快照表Snapshot Table动作会以源表相同的 schema 与分区方式创建一张名称不同的全新 Iceberg 表动作执行期间及执行之后源表都保持不变源表上的既有读写任务可以继续运行不受任何影响。整个流程分为三步创建新表以源表的元数据schema、partition spec 等为模板创建一张名称不同、位置独立的 Iceberg 表。源表上的 Readers 与 Writers 可以继续正常工作无需停机。提交数据文件将源表所有分区的数据文件全部提交到新 Iceberg 表中。此时源表仍然不变读侧Readers可以先切换到新 Iceberg 表。切换写侧待读侧验证无误后将全部 Writers 切换到新 Iceberg 表。当所有写任务完成切换后迁移流程即宣告完成。从源码实现来看SnapshotTable是定义在 api 模块的 Action 接口 中的一流动作一等公民其方法链包括as(destTableIdent)指定新表标识、tableLocation(location)指定新表位置、tableProperties(...)/tableProperty(...)设置表属性、executeWith(ExecutorService)指定并行读文件的线程池以及ignoreMissingFiles()用于跳过已消失的源数据文件执行结果通过importedDataFilesCount()返回导入的数据文件数。具体的不可变实现由 core 模块的 BaseSnapshotTable 通过 Immutables 框架生成。在 SnapshotTableSparkAction 中doExecute()的核心逻辑清晰地呈现了先暂存、后提交的安全模式通过stageDestTable()创建暂存表StagedSparkTable强校验源表位置与暂存表位置不得重叠包括互为前缀的情况否则会混合两张表的文件而直接报错调用SparkTableUtil.importSparkTable(...)为源表生成 Iceberg 元数据成功后commitStagedChanges()提交一旦抛错则进入finally分支执行abortStagedChanges()回滚暂存变更。值得注意的细节是快照表默认写入gc.enabledfalse并标记snapshottrue表属性见destTableProps()因为快照表与源表共享数据文件、并非这些文件的唯一所有者所以禁止对快照表执行会物理删除数据文件的expire_snapshots之类操作仅影响元数据的 Iceberg 删除如 DELETE 语句产生的 delete 文件仍然允许。相应地对源表执行 DELETE 移除原始数据文件也会破坏快照表的完整性。三、Migrate Table原地接管并替换源表Migrate Table动作同样会创建一张与源表 schema、分区方式一致的 Iceberg 表但区别在于动作执行过程中会锁定并从 catalog 中移除drop源表。因此Migrate Table 要求在执行前停止所有正在操作源表的修改任务支持 Iceberg 的读者Readers可以继续读取。流程同样分为三步停止写侧停止所有与源表交互的 Writers。备份并建新表创建一张与源表相同标识和元数据schema、partition spec 等的 Iceberg 表同时把源表重命名为备份表默认后缀_BACKUP_以备失败时回滚。提交并清理将源表所有分区的数据文件提交到新 Iceberg 表然后删除源表此时 Writers 即可开始向新 Iceberg 表写入。迁移完成后默认保留的备份表如db.sample_BACKUP_可通过drop_backuptrue参数选择删除。Migrate 的实现同样遵循暂存 提交的安全模式。在 MigrateTableSparkAction 中可以看到常量BACKUP_SUFFIX _BACKUP_定义了默认备份名规则构造函数即生成backupIdentrenameAndBackupSourceTable()先把源表重命名为备份表从而冻结源表、暂停一切修改并为其后的暂存建表腾出位置若备份名已存在则抛出AlreadyExistsException源表不存在则抛出NoSuchTableException后续流程从备份表而非源表导入数据文件到暂存的 Iceberg 表失败时restoreSourceTable()会把备份表重命名回原标识完成回滚成功后若开启dropBackup()则删除备份表destTableProps()会为迁移表写入migratedtrue属性并继承源表位置putIfAbsent(LOCATION, sourceTableLocation())确保新表原地接管原数据目录。从源码还可推断Migrate 对源 catalog 有较强约束checkSourceCatalog要求源 catalog 必须是SparkSessionCatalog即当前实现只支持从 Spark Session Catalog 中的非 Iceberg 表进行迁移。此外Migrate 会拒绝迁移使用不支持文件格式仅支持 Avro、Parquet、ORC的分区表也会因分桶无法在 Iceberg 中保留而直接失败。四、Add Files补齐迁移窗口期的新增数据在完成初始迁移无论采用 Snapshot Table 还是 Migrate Table之后经常会发现还有部分数据文件未被迁移。这些文件通常来自并发写入者——它们在迁移过程中或迁移结束后仍继续向源表写入数据。具体到不同格式对于 Hive 表这些未迁移文件是新增的 Hive 数据文件对于 Delta Lake 表这些未迁移文件是新产生的 snapshot版本。Add Files动作正是为将这些遗漏文件纳入 Iceberg 表而设计的。它不创建新表而是直接向一张已存在的 Iceberg 表追加来自 Hive/文件型表的数据文件且可以只导入指定分区Iceberg 会为这些文件生成元数据但不会移动文件本身。从 AddFilesProcedure 的源码看其源标识还支持以parquet.path、orc.path、avro.path形式直接指向文件型表位置isFileIdentifier()负责识别这类命名空间。使用 Add Files 前必须明确两个重要警告不校验 schema该过程不会分析文件 schema 是否与 Iceberg 表匹配添加 schema 不一致的文件会引发数据问题文件所有权转移一旦添加完成Iceberg 会将这些文件视为自己拥有的文件后续expire_snapshot等操作将能够物理删除这些文件因此只要可能应优先使用migrate或snapshot而非add_files。五、实战从 Hive 迁移到 IcebergHive 的 ORC、Parquet、Avro 三种文件格式均可迁移到 Iceberg。由于 Hive 表没有 snapshot 概念迁移过程本质上是用现有 schema 创建一张新的 Iceberg 表并把所有分区的数据文件一次性提交进去初始迁移之后的新增数据文件则通过 Add Files 动作持续补齐。这些动作由 Spark 集成模块以 Spark Procedure存储过程的形式提供已打包进 Spark runtime jar见 releases 下载页 中的 Spark runtime 产物。对应的过程定义与参数细节可参考 Spark Procedures 文档。5.1 Snapshot Hive 表CALL catalog_name.system.snapshot(db.source, db.dest)snapshot过程的完整参数如下详见 spark-procedures.md#snapshot参数是否必填类型说明source_table✔️string要快照的源表名table✔️string要创建的新 Iceberg 表名locationstring新表的位置默认交给 catalog 决定propertiesmapstring, string添加到新表的属性parallelismint文件读取线程数默认 1ignore_missing_filesboolean为 true 时跳过找不到的源数据文件而不是失败默认 false输出为imported_files_countlong即添加到新表的文件数。典型用法-- 在 catalog 默认位置创建引用 db.sample 的隔离快照表 db.snap CALL catalog_name.system.snapshot(db.sample, db.snap); -- 在指定位置 /tmp/temptable/ 创建快照表 CALL catalog_name.system.snapshot(db.sample, db.snap, /tmp/temptable/);快照表适合测试场景测试完成后用DROP TABLE清理即可。在 SnapshotTableProcedure 中可以看到各参数的默认行为——parallelism必须大于 0ignore_missing_files默认为 false且源表与目标表名不能相同。5.2 Migrate Hive 表CALL catalog_name.system.migrate(db.sample)migrate过程会复制源表的 schema、分区、属性和位置并用源表数据文件填充新表详见 spark-procedures.md#migrate参数是否必填类型说明table✔️string要迁移的表名propertiesmapstring, string新 Iceberg 表的属性drop_backupboolean为 true 时不再保留原表作为备份默认 falsebackup_table_namestring备份表名称默认table_BACKUP_parallelismint文件读取线程数默认 1ignore_missing_filesboolean为 true 时跳过找不到的源数据文件而不是失败默认 false输出为migrated_files_countlong即追加到 Iceberg 表的文件数。典型用法-- 迁移并在新表上添加属性 foobar CALL catalog_name.system.migrate(spark_catalog.db.sample, map(foo, bar)); -- 不添加额外属性直接迁移当前 catalog 中的表 CALL catalog_name.system.migrate(db.sample);5.3 从 Hive 表追加文件到 Iceberg 表CALL spark_catalog.system.add_files( table db.tbl, source_table db.src_tbl )add_files过程参数详见 spark-procedures.md#add_files参数是否必填类型说明table✔️string要追加文件的目标 Iceberg 表source_table✔️string文件来源表也支持file_format.path形式的路径partition_filtermapstring, string只导入指定分区的文件check_duplicate_filesboolean是否阻止添加已存在于表中的文件默认 trueparallelismint文件读取线程数默认 1输出为added_files_countlong与changed_partition_countlong未知时为空。注意当表属性compatibility.snapshot-id-inheritance.enabled为 true 或表格式版本大于 1 时changed_partition_count会返回 NULL。-- 只导入 part_col_1A 分区中的文件 CALL spark_catalog.system.add_files( table db.tbl, source_table db.src_tbl, partition_filter map(part_col_1, A) ); -- 从 parquet 文件型表位置导入全部文件 CALL spark_catalog.system.add_files( table db.tbl, source_table parquet.path/to/table );六、实战从 Delta Lake 迁移到 IcebergDelta Lake 采用 Parquet 文件格式并支持时间旅行time travel与版本管理。与 Hive 不同从 Delta Lake 迁移时通常希望保留全部历史因此常见的做法是把 Delta Lake 的所有 snapshot 都迁移过来以维持数据历史。目前 Iceberg 对 Delta Lake 只支持Snapshot Table动作由于 Delta Lake 表维护事务日志源表所有可用的事务会被按顺序提交到新 Iceberg 表作为对应的事务。初始迁移之后 Delta 表新增的数据文件会包含在其对应事务中后续通过Add Transaction动作Add Files 的变体目前仍在开发中追加到新表。6.1 启用迁移能力iceberg-delta-lake模块不会随 Spark、Flink 引擎运行时打包需要额外添加以下依赖iceberg-delta-lakeMaven 坐标org.apache.iceberg:iceberg-delta-lakedelta-standalone-0.6.0io.delta:delta-standalone_2.13:0.6.0delta-storage-2.2.0io.delta:delta-storage:2.2.0该模块基于Delta Standalone 0.6.0构建与测试支持的 Delta Lake 表协议版本为minReaderVersion: 1、minWriterVersion: 2协议版本语义参见 Delta Lake 官方的 Table Protocol Versioning 说明。6.2 API 与默认实现模块提供了DeltaLakeToIcebergMigrationActionsProvider接口包含动作snapshotDeltaLakeTable将一张已有 Delta Lake 表快照为 Iceberg 表。接口的默认实现可通过以下方式获取DeltaLakeToIcebergMigrationActionsProvider defaultActions DeltaLakeToIcebergMigrationActionsProvider.defaultActions()snapshotDeltaLakeTable动作会读取 Delta Lake 表的事务在一个 Iceberg 事务内将其转换为一张具有相同 schema 与分区方式的新 Iceberg 表源 Delta Lake 表保持原样。新表可以独立读写而不影响源表但快照使用的是源表的数据文件——通过从源表 schema 生成的 name-to-id 映射name mapping来读取。快照表上的 INSERT / OVERWRITE 产生的新文件会写入快照表自身的位置该位置默认与源 Delta Lake 表相同也可通过 API 另行指定。SnapshotDeltaLakeTable动作接口定义在 delta-lake 模块其必需输入与配置方式如下必需输入配置方式说明源表位置参数sourceTableLocation源 Delta Lake 表的位置新表标识APIas(TableIdentifier)指定新 Iceberg 表的 namespace 与表名Iceberg catalogAPIicebergCatalog(Catalog)用于创建新表的 catalogHadoop 配置APIdeltaLakeConfiguration(Configuration)读取源 Delta Lake 表所需的 Hadoop 配置输出为imported_files_countlong即添加到新表的文件数。动作执行后还会为新表写入以下默认属性属性名值说明snapshot_sourcedelta标记该表由 Delta Lake 表快照而来original_location源 Delta Lake 表位置源表的绝对路径schema.name-mapping.default由 schema 推导的 JSON name mapping用于读取 Delta Lake 数据文件的 name mapping6.3 Java 调用示例import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.catalog.Catalog; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.delta.DeltaLakeToIcebergMigrationActionsProvider; String sourceDeltaLakeTableLocation s3://my-bucket/delta-table; String destTableLocation s3://my-bucket/iceberg-table; TableIdentifier destTableIdentifier TableIdentifier.of(my_db, my_table); Catalog icebergCatalog ...; // 从 Spark 等引擎获取或通过 CatalogUtil.loadCatalog 创建 Configuration hadoopConf ...; // 从引擎获取、且已配置访问 Delta Lake 表所需文件系统的 Hadoop Configuration DeltaLakeToIcebergMigrationActionsProvider.defaultActions() .snapshotDeltaLakeTable(sourceDeltaLakeTableLocation) .as(destTableIdentifier) .icebergCatalog(icebergCatalog) .tableLocation(destTableLocation) .deltaLakeConfiguration(hadoopConf) .tableProperty(my_property, my_value) .execute();七、源码实现与测试验证7.1 Action 接口体系三类迁移动作在 Iceberg 中都被建模为Action接口位于 api 模块的 actions 包SnapshotTable提供as、tableLocation、tableProperties、tableProperty、executeWith、ignoreMissingFiles等方法结果返回importedDataFilesCountMigrateTable提供tableProperties、tableProperty、dropBackup、backupTableName、executeWith、ignoreMissingFiles等方法结果返回migratedDataFilesCount。它们的不可变实现分别由 BaseSnapshotTable 与 BaseMigrateTable 通过 Immutables 注解生成保证 Action 配置后不可变、可安全复用。7.2 Spark 侧的暂存提交机制在 Spark 集成中两个核心 Action 实现都依赖StagedSparkTableStagingTableCatalog的暂存机制SnapshotTableSparkAction 要求源 catalog 必须是 session catalogspark_catalog并校验新旧表位置不得重叠MigrateTableSparkAction 要求源 catalog 为SparkSessionCatalog通过重命名备份 → 暂存建表 → 提交 → 失败回滚/成功删备份的流程实现原子替换。对应的过程层封装SnapshotTableProcedure、MigrateTableProcedure、AddFilesProcedure把 Action 的参数校验与默认值落实为可调用的 Spark Procedure。7.3 测试覆盖仓库中的测试为上述行为提供了可复现的验证TestSnapshotTableAction 覆盖了并行任务快照、位置重叠时报错、非重叠位置等场景TestMigrateTableAction 覆盖了并行任务下的迁移流程。八、方案选型与注意事项场景推荐方案原因不想停机、先验证再切换Snapshot Table源表全程可用读侧先切、写侧后切迁移失败无影响可以接受短暂停机、希望原地接管Migrate Table原标识原地替换自动保留备份可回滚位置继承源表迁移后有并发写入者遗留的数据Add Files按分区精准补齐新文件无需重建表几点必须牢记的约束隔离性差异Snapshot 与 Migrate 都共享源数据文件源表侧一旦 vacuum/删除数据文件新表会受影响也不要对快照表执行expire_snapshots等物理删除操作格式支持原地迁移仅支持 Avro、Parquet、ORC 文件格式分桶表无法迁移分桶语义无法保留Add Files 的风险不校验 schema且添加后的文件归 Iceberg 所有、可被物理删除能不用就不用Delta Lake 迁移特殊性需额外引入三个依赖且当前只支持 Snapshot 路径增量事务补齐Add Transaction仍在开发中。通过上述三种动作的组合你可以在控制停机时间与数据冗余成本的前提下将 Hive、Delta Lake 等存量表平稳迁移到 Apache Iceberg并获得 Iceberg 的事务、版本管理与 ACID 能力。更细的语法与参数可继续参阅 Hive 迁移文档、Delta Lake 迁移文档 与 Spark Procedures 参考。赞分享数据湖大数据数据存储【免费下载链接】icebergApache Iceberg项目地址https://gitcode.com/gh_mirrors/icebe/iceberg点击查看免费下载相关推荐Apache Iceberg Hive表迁移完全指南Apache Iceberg Hive表迁移完全指南 概述 Apache Iceberg作为新一代数据湖表格式相比传统Hive表具有诸多优势包括ACID事务大数据数据湖OLAP数据存储Apache Iceberg Hive表迁移完全指南Apache Iceberg Hive表迁移完全指南 概述 在现代数据架构中将传统Hive表迁移到Apache Iceberg表已成为提升数据管理能力的重要步数据湖大数据数据存储Apache Iceberg 迁移指南使用 snapshotDeltaLakeTable 将 Delta Lake 表完整迁移至 IcebergApache Iceberg 迁移指南使用 snapshotDeltaLakeTable 将 Delta Lake 表完整迁移至 Iceberg Delta数据湖大数据数据存储上一篇HiddenVM终极指南5大高级技巧实现桌面环境无痕使用下一篇Windows Terminal 主题联动快速上手4步实现配色方案自动切换创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考