数据湖湖仓一体大数据【免费下载链接】icebergApache Iceberg项目地址https://gitcode.com/gh_mirrors/iceberg5/iceberg点击查看免费下载导读本指南以 docs/docs/java-api-quickstart.md 为核心面向希望绕过 SQL 引擎、直接用 Java 代码操作 Apache Iceberg 表的开发者。你将掌握通过Catalog/Tables接口创建与加载表、手工构造Schema与PartitionSpec、从 Avro/Spark 转换 Schema以及使用ManageSnapshotsAPI 完成分支Branch与标签Tag的创建、写入、读取、替换与保留策略管理等完整能力。文中所有 API 均可在当前仓库的api、core、hive-metastore等模块源码中找到对应实现示例可直接复制改造运行。创建表Catalog 与 Tables 接口Iceberg 的表可以通过Catalog接口或Tables接口的实现来创建。Catalog是面向命名空间namespace组织的完整接口定义了createTable、loadTable、renameTable、dropTable、listTables等表管理方法Tables则更适合按路径直接加载单表。生产环境最常用的是Catalog的两个标准实现HiveCatalog与HadoopCatalog。使用 Hive CatalogHiveCatalog通过连接 Hive MetastoreHMS来跟踪 Iceberg 表的元数据。初始化时只需给它一个名称和一组属性import java.util.HashMap; import java.util.Map; import org.apache.iceberg.hive.HiveCatalog; HiveCatalog catalog new HiveCatalog(); catalog.setConf(spark.sparkContext().hadoopConfiguration()); // 可选复用 Spark 的 Hadoop 配置 MapString, String properties new HashMapString, String(); properties.put(warehouse, ...); // 数据仓库根路径 properties.put(uri, ...); // Hive Metastore 的 Thrift URI catalog.initialize(hive, properties);从源码看HiveCatalog.initialize见 HiveCatalog.java会做以下几件事若未显式设置 HadoopConfiguration则回退到默认环境配置将uri属性写入HiveConf.ConfVars.METASTOREURIS将warehouse写入METASTOREWAREHOUSE默认使用HadoopFileIO读写文件可通过io-impl属性替换创建带缓存和连接池的 HMS 客户端CachedClientPool。HiveCatalog实现了Catalog接口因此可以直接进行表操作。创建表时传入TableIdentifier、Schema以及初始元数据如分区规范import org.apache.iceberg.Table; import org.apache.iceberg.catalog.TableIdentifier; TableIdentifier name TableIdentifier.of(logging, logs); Table table catalog.createTable(name, schema, spec); // 或者加载一张已存在的表 Table table catalog.loadTable(name);其中schema与spec的创建方法见下文 Schemas 与 Partitioning 两节。TableIdentifier.of(logging, logs)表示命名空间为logging、表名为logs即通常所说的库.表两级结构。使用 Hadoop CatalogHadoopCatalog不依赖 Hive Metastore直接把仓库目录下的目录结构映射为命名空间与表创建db.tbl时会在仓库目录下创建db/tbl目录并在其中写入表元数据。它的限制是只能用于支持原子重命名atomic rename的文件系统如 HDFS在本地文件系统或 S3 上多个客户端并发写同一张表是不安全的。import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.hadoop.HadoopCatalog; Configuration conf new Configuration(); String warehousePath hdfs://host:8020/warehouse_path; HadoopCatalog catalog new HadoopCatalog(conf, warehousePath);HadoopCatalog同样实现Catalog接口具备createTable、loadTable、dropTable等方法。该构造器在内部以warehouse为属性调用initialize(hadoop, ...)见 HadoopCatalog.java并会强制校验 warehousePath 非空。需要特别注意的是从源码看HadoopCatalog不支持renameTable直接抛出UnsupportedOperationException见 HadoopCatalog.java也不支持为表设置自定义 location。创建表的示例import org.apache.iceberg.Table; import org.apache.iceberg.catalog.TableIdentifier; TableIdentifier name TableIdentifier.of(logging, logs); Table table catalog.createTable(name, schema, spec); // 或者加载一张已存在的表 Table table catalog.loadTable(name);Spark 中使用表Spark 可以通过名称直接使用HiveCatalog创建的表前提是 Spark session 配置了对应的 catalog// spark.sql.catalog.hive_prod org.apache.iceberg.spark.SparkCatalog // spark.sql.catalog.hive_prod.type hive spark.table(logging.logs);Spark 也可以按路径加载由HadoopCatalog创建的表spark.read.format(iceberg).load(hdfs://host:8020/warehouse_path/logging/logs);注意这里的路径就是HadoopCatalog的warehousePath拼接命名空间与表名与源码中defaultWarehouseLocation的拼接逻辑warehouseLocation/namespace/table见 HadoopCatalog.java完全一致。Schemas定义与转换表结构创建 Schema下面的例子为logs表创建了一个 schema包含日志级别、事件时间、消息正文与可选的调用栈列表import org.apache.iceberg.Schema; import org.apache.iceberg.types.Types; Schema schema new Schema( Types.NestedField.required(1, level, Types.StringType.get()), Types.NestedField.required(2, event_time, Types.TimestampType.withZone()), Types.NestedField.required(3, message, Types.StringType.get()), Types.NestedField.optional(4, call_stack, Types.ListType.ofRequired(5, Types.StringType.get())) );关键点直接使用 Iceberg API 时必须显式给出字段 ID上面的1、2、3…。ID 是 Iceberg 元数据中跟踪字段身份、支撑 schema 演进的基石从其他格式Spark、Avro、Parquet转换而来的 schema 会自动分配新 ID。顶层字段 ID 与嵌套字段 ID 必须在整个 schema 内唯一如上例列表元素的 ID 为5。创建表时schema 中的所有 ID 会被重新分配以保证唯一性因此即便手写 ID 有冲突建表阶段也会被规范化。required与optional决定字段是否允许为空TimestampType.withZone()表示带时区的时间戳对应 SQL 的TIMESTAMP WITH TIME ZONE。从 Avro Schema 转换借助AvroSchemaUtil中的转换器可以把现成的 Avro schema 转换为 Iceberg schemaimport org.apache.avro.Schema; import org.apache.avro.Schema.Parser; import org.apache.iceberg.avro.AvroSchemaUtil; Schema avroSchema new Parser().parse({\type\: \record\, ...}); Schema icebergSchema AvroSchemaUtil.toIceberg(avroSchema);AvroSchemaUtil.toIceberg的实现位于 AvroSchemaUtil.java它通过AvroSchemaVisitor把 Avro 类型访问器转换为 Iceberg 的Type再包装为Schema。转换时字段 ID 来自 Avro 属性如field-id、element-id缺失时会自动分配。从 Spark 表转换如果表已经存在于 Spark 中可以用SparkSchemaUtil直接取其 schemaimport org.apache.iceberg.spark.SparkSchemaUtil; Schema schema SparkSchemaUtil.schemaForTable(sparkSession, tableName);schemaForTable内部调用convert(spark.table(name).schema())见 SparkSchemaUtil.java会把 Spark 的StructType完整转换为 IcebergSchemaSpark/Hive 分区列也会包含其中。仓库中spark/v3.3、spark/v3.4、spark/v3.5三个版本目录下均有该工具类。Partitioning创建分区规范分区规范PartitionSpec描述 Iceberg 如何把记录分组到数据文件中——同一分区值内的记录会落入相同的数据文件从而让扫描器在规划阶段跳过无关文件。分区规范基于表的 schema 通过 builder 创建。下面的例子为logs表创建分区规范按日志事件时间的小时粒度分区再按日志级别做 identity 分区import org.apache.iceberg.PartitionSpec; PartitionSpec spec PartitionSpec.builderFor(schema) .hour(event_time) .identity(level) .build();Iceberg 内置的分区转换transform在仓库的 format/spec.md 中有完整定义汇总如下转换名说明适用的源类型结果类型identity源值原样保留任意源类型bucket[N]对值取 32 位 Murmur3 哈希后对N取模int、long、decimal、date、time、timestamp、timestamptz、string、uuid、fixed、binary等inttruncate[W]按宽度W截断int、long、decimal、string、binary源类型year/month/day/hour提取时间单位的整数自 1970 起date、timestamp、timestamptz等intvoid恒为null用于 v1 表中“丢弃”分区字段任意源类型或int例如bucket[N]的精确语义为bucket_N(x) (murmur3_x86_32_hash(x) Integer.MAX_VALUE) % Ntruncate[W]对字符串截取不超过L个 code point 的 UTF-8 子串。所有转换对null输入都必须返回null。分区规范还可以随时间演进增加、删除、重命名、重排字段演进产生新 spec ID但不应改变既有分区字段 ID。Branching and Tagging分支与标签Iceberg 的 snapshot 引用机制branch / tag让表可以像 Git 一样拥有多条并行历史适用于审计、灰度、按版本发布等场景。Java 库中所有相关操作都通过table.manageSnapshots()返回的ManageSnapshots接口完成该接口定义在 ManageSnapshots.java核心实现是core模块的SnapshotManager见 SnapshotManager.java。所有修改都必须调用commit()才会真正生效。创建分支与标签/* 创建分支 test-branch 该分支整体保留 1 周7 天 604800000 ms 分支上最新的 2 个 snapshot 始终保留 分支上最近 1 小时内3600000 ms创建的 snapshot 也会被保留。 */ String branch test-branch; table.manageSnapshots() .createBranch(branch, 3) // 在 snapshot 3 处创建分支 .setMinSnapshotsToKeep(branch, 2) // 最少保留的 snapshot 数 .setMaxSnapshotAgeMs(branch, 3600000) // snapshot 最大保留时长毫秒 .setMaxRefAgeMs(branch, 604800000) // 分支引用本身的保留时长毫秒 .commit(); // 在 snapshot 10 处创建标签 historical-tag标签保留 1 天86400000 ms String tag historical-tag; table.manageSnapshots() .createTag(tag, 10) .setMaxRefAgeMs(tag, 86400000) .commit();理解这几个保留参数定义见 ManageSnapshots.javasetMinSnapshotsToKeep(branch, n)分支至少保留的 snapshot 数量setMaxSnapshotAgeMs(branch, ms)分支上 snapshot 的最大保留时长超过即视为可清理setMaxRefAgeMs(ref, ms)引用本身branch/tag的保留时长到期后引用会被过期清理机制移除。createBranch(name)不带 snapshot ID会指向当前 snapshot若表尚无 snapshot则指向一个新建的空 snapshot见 SnapshotManager.java。向分支提交写入写入某个分支只需在对应写操作上指定toBranch(...)。更多可指定toBranch的更新操作列表参见 docs/docs/api.md 中的 Update operations 一节newAppend、newFastAppend、newOverwrite、newDelete、newRewrite、newRowDelta等均继承自Table见 Table.java。// 把 FILE_A 追加到 test-branch String branch test-branch; table.newAppend() .appendFile(FILE_A) .toBranch(branch) .commit(); // 在 test-branch 上执行行级更新新增数据文件 删除文件 table.newRowDelta() .addRows(DATA_FILE) .addDeletes(DELETES) .toBranch(branch) .commit(); // 在 test-branch 上执行重写用 compactedFile 替换 SMALL_FILE_1 与 SMALL_FILE_2 table.newRewrite() .rewriteFiles(ImmutableSet.of(SMALL_FILE_1, SMALL_FILE_2), ImmutableSet.of(compactedFile)) .toBranch(branch) .commit();从分支和标签读取读取分支或标签与普通表扫描一致只需在TableScan上调用useRef(...)。传入分支时使用的是该分支的 head snapshot。需要注意目前不支持在按分支读取的同时再指定asOfSnapshotIduseRef的定义见 TableScan.java。// 读取 test-branch 的 head snapshot TableScan branchRead table.newScan().useRef(test-branch); // 读取 audit-tag 所指向的 snapshot TableScan tagRead table.newScan().useRef(audit-tag);替换与快进分支和标签已有分支/标签指向的 snapshot 可以通过replace系列 API 更新。快进fast forward操作类似 Git 的 fast-forward当目标分支是源分支或标签的祖先时可以把目标分支推进到源分支的 head。无论是 fast forward 还是 replace目标分支的保留属性默认都会保持不变。// 把 test-branch 更新为指向 snapshot 4 table.manageSnapshots() .replaceBranch(branch, 4) .commit(); String tag audit-tag; // 把 audit-tag 替换为指向 snapshot 3并更新其保留时长 table.manageSnapshots() .replaceBranch(tag, 4) .setMaxRefAgeMs(1000) .commit();replaceBranch(from, to)与fastForwardBranch(from, to)还支持把分支推进到另一个引用的位置见 ManageSnapshots.java其中fastForwardBranch会校验from必须是to的祖先否则抛出IllegalArgumentException。更新保留属性分支和标签的保留属性可以随时调整引用本身的保留时长用setMaxRefAgeMs更新分支内 snapshot 的保留策略用setMinSnapshotsToKeep与setMaxSnapshotAgeMs更新。String branch test-branch; // 更新 test-branch 的保留属性至少保留 10 个 snapshot // snapshot 最大保留 2 小时分支引用保留 1 周 table.manageSnapshots() .setMinSnapshotsToKeep(branch, 10) .setMaxSnapshotAgeMs(branch, 7200000) .setMaxRefAgeMs(branch, 604800000) .commit(); // 更新 test-tag 的保留时长保留 1 周 table.manageSnapshots() .setMaxRefAgeMs(test-tag, 604800000) .commit();删除分支和标签使用removeBranch与removeTag分别删除分支与标签接口定义见 ManageSnapshots.java。注意引用不存在时会抛出IllegalArgumentException。// 删除 test-branch table.manageSnapshots() .removeBranch(test-branch) .commit(); // 删除 test-tag table.manageSnapshots() .removeTag(test-tag) .commit();小结本文从建表出发覆盖了 Iceberg Java API 的核心链路HiveCatalog/HadoopCatalog的初始化与区别、TableIdentifierSchemaPartitionSpec的组合建表、Avro/Spark schema 转换、以及基于ManageSnapshots的完整分支标签生命周期管理。后续可进一步阅读 docs/docs/api.md完整 Java API 与事务机制、docs/docs/configuration.mdCatalog 与表级配置参数如warehouse、uri、clients、lock-impl等以及 format/spec.md分区转换与表格式规范的权威定义构建完整的 Iceberg 应用知识体系。赞分享数据湖湖仓一体大数据【免费下载链接】icebergApache Iceberg项目地址https://gitcode.com/gh_mirrors/iceberg5/iceberg点击查看免费下载相关推荐Apache Iceberg Spark DDL 完全指南建表、Schema 演进、分区演化与分支标签管理Apache Iceberg Spark DDL 完全指南建表、Schema 演进、分区演化与分支标签管理 本文基于 Apache Iceberg 官方 Sp数据湖大数据数据存储Apache Iceberg 分支与标签机制深度解析Apache Iceberg 分支与标签机制深度解析 概述 在现代数据湖架构中数据版本控制和时间旅行查询是核心需求。Apache Iceberg 通过创新的快大数据数据湖OLAP数据存储Apache Iceberg 入门指南在 Spark 3 中快速上手 Iceberg 表Apache Iceberg 入门指南在 Spark 3 中快速上手 Iceberg 表 导读 本文是 Apache Iceberg 官方文档中的 Spark大数据数据湖OLAP数据存储上一篇GPU 核心绑定与中断调优Windows 游戏降延迟下一篇Civitai Bulk Image Manager 迁移审计从 Retool 40 个查询到 7 个服务函数的批量图片治理实战创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考