数据湖湖仓一体大数据【免费下载链接】icebergApache Iceberg项目地址https://gitcode.com/gh_mirrors/iceberg5/iceberg点击查看免费下载Apache Iceberg 是面向大规模分析型数据湖的开放表格式而 Spark 是目前 Iceberg 操作中功能最为丰富的计算引擎。本文以 spark-getting-started.md 为骨架带你从零开始在 Spark 3 环境中引入 Iceberg 运行时、配置基于路径或 Hive Metastore 的 Catalog、用 SQL 与 DataFrame API 完成建表、写入与读取并透彻理解 Spark 与 Iceberg 之间的类型转换规则。读完本文你将具备一套可直接运行的 Iceberg Spark 3 实战方案并知道如何进一步深入 DDL、查询、写入与表维护文档。为什么推荐用 Spark 上手 Iceberg在 Apache Iceberg 的生态中Spark 被官方文档定位为功能最丰富的计算引擎。官方文档明确建议初学者先通过 Spark 来理解 Iceberg 的核心概念与特性见 spark-getting-started.md。如果你使用的是其他计算引擎如 Flink、Trino、Hive可以在 多引擎支持 页面找到对应文档。Iceberg 与 Spark 的集成基于 Spark 3 的DataSourceV2DSv2API实现数据源与 Catalog 接口这也是 Spark 3 中 Iceberg 表能被 SQL 与 DataFrame 无缝操作的技术基础。在 Spark 3 中引入 Iceberg使用--packages在 Spark Shell 中加载最简单的方式是在启动 Spark Shell 时通过--packages参数自动从 Maven 中央仓库下载 Iceberg 的 Spark 运行时构件spark-shell --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:{{ icebergVersion }}其中{{ icebergVersion }}是文档模板中的版本占位符在实际发布版本文档中会被替换为具体的 Iceberg 版本号。构件名iceberg-spark-runtime-3.5_2.12中的3.5表示对应的 Spark 大版本2.12表示 Scala 版本。提示如果希望把 Iceberg 固化到你的 Spark 安装环境中可以将对应的iceberg-spark-runtime-3.5_2.12Jar 文件放入 Spark 的jars目录之后无需每次通过--packages加载。从当前仓库的构建配置gradle.properties可以看到仓库维护了 Spark 3.3、3.4、3.5 三套版本目录systemProp.knownSparkVersions3.3,3.4,3.5默认 Spark 版本为 3.5默认 Scala 版本为 2.12。因此本文示例统一使用iceberg-spark-runtime-3.5_2.12其他版本组合可对应替换构件名。添加 Iceberg SQL 扩展Iceberg 在 Spark 3 中通过一个扩展模块提供额外的 SQL 命令例如存储过程CALL、ALTER TABLE ... WRITE ORDERED BY等。启用方式是在启动参数中设置spark.sql.extensionsSpark 扩展属性Iceberg 扩展实现spark.sql.extensionsorg.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions从源码看该扩展类的实现位于 spark/v3.5/spark-extensions/src/main/scala/org/apache/iceberg/spark/extensions/IcebergSparkSessionExtensions.scala它通过注入 Spark 的 Parser、Analyzer、Optimizer 与 Planner 规则为MERGE INTO、UPDATE、DELETE FROM、分区演进、分支与标签等能力提供 SQL 解析和执行支持v3.3、v3.4、v3.5 目录下均有同名实现。添加 Catalog让 SQL 能够按名称管理表Iceberg 提供多种 Catalog 实现使得 SQL 命令可以管理表并按名称加载表。Catalog 通过spark.sql.catalog.(catalog_name)前缀下的属性进行配置。完整示例路径型 Catalog 内置 Catalog 扩展下面的命令创建了一个名为local的基于路径Hadoop的 Catalog表数据存放在$PWD/warehouse下同时把 Spark 内置的spark_catalog替换为 Iceberg 的SparkSessionCatalog从而为内置 Catalog 增加 Iceberg 表支持spark-sql --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:{{ icebergVersion }}\ --conf spark.sql.extensionsorg.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \ --conf spark.sql.catalog.spark_catalogorg.apache.iceberg.spark.SparkSessionCatalog \ --conf spark.sql.catalog.spark_catalog.typehive \ --conf spark.sql.catalog.localorg.apache.iceberg.spark.SparkCatalog \ --conf spark.sql.catalog.local.typehadoop \ --conf spark.sql.catalog.local.warehouse$PWD/warehouse配置项说明配置项作用spark.sql.extensions加载 Iceberg SQL 扩展行级更新、存储过程等命令必需spark.sql.catalog.spark_catalog将 Spark 内置 Catalog 替换为SparkSessionCatalog让同一 Hive Metastore 中的非 Iceberg 表仍可正常加载spark.sql.catalog.spark_catalog.type底层实现类型此处为hivespark.sql.catalog.local注册名为local的 Catalog实现类为SparkCatalogspark.sql.catalog.local.type底层 Iceberg Catalog 类型hadoop表示基于 HDFS/本地路径的HadoopCatalogspark.sql.catalog.local.warehouse仓库根路径表数据与元数据均存放于此两种 Catalog 实现的区别Iceberg 为 Spark 提供两个 Catalog 适配类均可在 spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ 目录下找到源码org.apache.iceberg.spark.SparkCatalog支持 Hive Metastore、Hadoop 仓库、REST、Glue、JDBC、Nessie 等底层实现是一个专职的 Iceberg Catalogorg.apache.iceberg.spark.SparkSessionCatalog为 Spark 内置 Catalog 增加 Iceberg 表支持非 Iceberg 表则委托给内置 Catalog 加载从而允许同一 Hive Metastore 同时存放 Iceberg 与非 Iceberg 表。从 SparkCatalog.java 的类注释与buildIcebergCatalog实现可以看到它支持type、uri、warehouse、catalog-impl、io-impl、cache-enabled、table-default.*、table-override.*等属性并把当前 Spark 应用的applicationId与用户信息写入 Catalog 属性。Catalog 通用配置属性精选在 spark-configuration.md 中给出了更完整的 Catalog 配置表常用属性如下属性取值示例说明spark.sql.catalog.名称.typehive、hadoop、rest、glue、jdbc、nessie底层 Iceberg Catalog 实现类型使用自定义 Catalog 时可留空spark.sql.catalog.名称.catalog-impl自定义类名自定义Catalog实现类若type为空则必须指定spark.sql.catalog.名称.urithrift://host:port/http://localhost:8080Hive Metastore 地址hive 类型或 REST 地址rest 类型spark.sql.catalog.名称.warehousehdfs://nn:8020/warehouse/path仓库基础路径spark.sql.catalog.名称.cache-enabledtrue/false是否启用 Catalog 缓存默认truespark.sql.catalog.名称.default-namespacedefaultCatalog 的默认命名空间spark.sql.catalog.名称.table-default.键—由该 Catalog 创建的表默认携带的属性值spark.sql.catalog.名称.table-override.键—强制性的表属性值用户无法覆盖spark.sql.catalog.名称.hadoop.*fs.s3a.endpoint...仅作用于 Iceberg 表的 per-Catalog Hadoop 配置优先级高于spark.hadoop.*REST / Hive / Hadoop Catalog 快速配置参考# Hive Metastore 型 Catalog省略 uri 时使用 hive-site.xml 中的 hive.metastore.uris spark.sql.catalog.hive_prod org.apache.iceberg.spark.SparkCatalog spark.sql.catalog.hive_prod.type hive spark.sql.catalog.hive_prod.uri thrift://metastore-host:port # REST 型 Catalog spark.sql.catalog.rest_prod org.apache.iceberg.spark.SparkCatalog spark.sql.catalog.rest_prod.type rest spark.sql.catalog.rest_prod.uri http://localhost:8080 # Hadoop 型 Catalog目录型 spark.sql.catalog.hadoop_prod org.apache.iceberg.spark.SparkCatalog spark.sql.catalog.hadoop_prod.type hadoop spark.sql.catalog.hadoop_prod.warehouse hdfs://nn:8020/warehouse/pathCatalog 名称在 SQL 中用于标识表。例如注册了hive_prod后SELECT * FROM hive_prod.db.table就会从该 Catalog 加载db.table。Spark 3 还会跟踪当前 Catalog 与命名空间执行USE hive_prod.db后可直接写SELECT * FROM table并可通过SHOW CURRENT NAMESPACE查看当前上下文。创建你的第一张 Iceberg 表在spark-sqlShell 中或通过spark.sql(...)执行CREATE TABLE即可在上一步定义的localCatalog 中建表-- local 是上文定义的基于路径的 Catalog CREATE TABLE local.db.table (id bigint, data string) USING iceberg;USING iceberg是 Spark 3 创建 Iceberg 表的固定写法。Iceberg Catalog 支持完整的 SQL DDL 命令集包括CREATE TABLE ... PARTITIONED BY声明分区CREATE TABLE ... AS SELECTCTAS 原子建表ALTER TABLE列增删改、属性设置、分区演进等DROP TABLE注意 0.14 起DROP TABLE只移除 Catalog 中的表条目删除数据需使用DROP TABLE PURGE在仓库测试代码中可以看到大量同类建表用法例如 TestSparkCatalogOperations.java 中同时覆盖了CREATE TABLE ... USING iceberg、CTASCREATE TABLE ... USING iceberg AS SELECT与 RTASREPLACE TABLE ... AS SELECT场景可以作为验证上述语法的参考用例。PARTITIONED BY还支持 Iceberg 的隐藏分区转换表达式例如bucket(16, id)、days(ts)、truncate(L, col)等详见 partitioning.md 与 spark-ddl.md。写入数据用 SQL 追加数据建表后即可用INSERT INTO写入数据既支持常量列表也支持子查询INSERT INTO local.db.table VALUES (1, a), (2, b), (3, c); INSERT INTO local.db.table SELECT id, data FROM source WHERE length(data) 1;行级更新MERGE INTOIceberg 为 Spark 带来了行级 SQL 更新能力即MERGE INTO与DELETE FROMMERGE INTO local.db.target t USING (SELECT * FROM updates) u ON t.id u.id WHEN MATCHED THEN UPDATE SET t.count t.count u.count WHEN NOT MATCHED THEN INSERT *;MERGE INTO的底层实现是重写包含待更新行的数据文件以一次overwrite提交完成更新详见 spark-writes.md。官方文档推荐优先使用MERGE INTO而非INSERT OVERWRITE因为 Iceberg 只重写受影响的数据文件且动态覆盖在表分区演进后行为可能变化。需要注意MERGE INTO、UPDATE、行级DELETE均依赖上文中配置的 Iceberg SQL 扩展。用 DataFrame v2 写入 API 追加数据Iceberg 支持新的 v2 DataFrame 写入 APIspark.table(source).select(id, data) .writeTo(local.db.table).append()旧的writeAPI 虽然仍受支持但官方明确不推荐使用。v2 API 的优势包括支持 CTAS/RTAS、统一按列名写入、支持隐藏分区表达式、覆盖行为显式可控且每个操作与 SQL 语句一一对应append()对应INSERT INTO、overwritePartitions()对应动态INSERT OVERWRITE等。相关的读写选项如write-format、target-file-size-bytes、snapshot-property.*、时间旅行读选项snapshot-id、as-of-timestamp、branch、tag等可在 spark-configuration.md 中查到完整表格其常量定义也见源码 SparkReadOptions.java。读取数据用 SQL 查询SQL 读取直接使用 Iceberg 表名即可SELECT count(1) as count, data FROM local.db.table GROUP BY data;SQL 也是官方推荐的检查表方式。Iceberg 提供了一系列元数据表metadata tables以表名为命名空间追加元数据表名即可访问。例如查看表中所有快照SELECT * FROM local.db.table.snapshots;------------------------------------------------------------------------------------------------------------------------ | committed_at | snapshot_id | parent_id | operation | manifest_list | ... | ------------------------------------------------------------------------------------------------------------------------ | 2019-02-08 03:29:51.215 | 57897183625154 | null | append | s3://.../table/metadata/snap-57897183625154-1.avro | ... | | | | | | | ... | | | | | | | ... | | ... | ... | ... | ... | ... | ... | ------------------------------------------------------------------------------------------------------------------------快照是 Iceberg 表状态的一次完整记录每次写入append/overwrite/delete 等都会产生一个新快照snapshot_id、parent_id、operation与manifest_list共同描述了表的演化历史。除snapshots外还有history、files、manifests、partitions、refs、metadata_log_entries以及跨快照的all_*系列元数据表均可在 spark-queries.md 中查阅。元数据表同样支持与时间旅行结合例如SELECT * FROM prod.db.table.manifests TIMESTAMP AS OF 2021-09-20 08:00:00。用 DataFrame 读取DataFrame 读取也已支持可通过spark.table直接按表名引用val df spark.table(local.db.table) df.count()spark.table/spark.read.format(iceberg).load(...)对标识符的解析有明确的优先级file:///path/to/table按路径加载、单段名称解析为当前Catalog.当前Namespace.表名、多段名称按catalog.namespace.table形式匹配Catalog 优先于 Namespace 解析。此外Spark 3.3 及更高版本支持 SQL 时间旅行TIMESTAMP AS OF/VERSION AS OFVERSION AS OF还可接受分支或标签名详见 spark-queries.md。类型兼容性Spark 与 Iceberg 的类型转换Spark 与 Iceberg 支持的类型集合并不完全相同。Iceberg 会自动进行类型转换但并非所有组合都支持因此在设计表列类型之前先理解下面的转换规则会很有帮助。Spark 类型 → Iceberg 类型该表描述创建 Iceberg 表以及通过 Spark 写入 Iceberg 表时的类型转换SparkIceberg备注booleanbooleanshortintegerbyteintegerintegerintegerlonglongfloatfloatdoubledoubledatedatetimestamptimestamp with timezonetimestamp_ntztimestamp without timezonecharstringvarcharstringstringstringbinarybinarydecimaldecimalstructstructarraylistmapmap在 SparkTypeToType.java 的atomic方法中可以看到上述映射的具体实现ShortType、ByteType与IntegerType统一转换为 IcebergIntegerTypeCharType、VarcharType与StringType统一转换为StringTypeTimestampType转换为带时区时间戳、TimestampNTZType转换为不带时区时间戳。提示上述表格反映的是建表时的类型表示实际写入时支持的范围更广Iceberg 数值类型integer、long、float、double、decimal在写入时支持类型提升例如可以用 Spark 的short、byte、integer、long写入 Iceberg 的long列可以用 Spark 的binary写入 Iceberg 的fixed类型此时会对长度进行断言检查。Iceberg 类型 → Spark 类型该表描述从 Iceberg 表读取时转换到 Spark 类型的规则IcebergSpark备注booleanbooleanintegerintegerlonglongfloatfloatdoubledoubledatedatetime不支持timestamp with timezonetimestamptimestamp without timezonetimestamp_ntzstringstringuuidstringfixedbinarybinarybinarydecimaldecimalstructstructlistarraymapmap值得特别留意的是 Iceberg 的time类型在读回 Spark 时不受支持因此在设计表结构时如果需要通过 Spark 读取应避免使用该类型。uuid读回为string、fixed读回为binary也都是常见的降级映射。下一步学习路径完成本文的快速入门后可以按需深入以下主题链接均指向当前仓库文档DDL 命令CREATE、ALTER、DROP含分区演进、排序与分布、分支与标签等扩展能力查询数据SELECT查询、时间旅行、增量读取与各类元数据表写入数据INSERT INTO、MERGE INTO、INSERT OVERWRITE、DataFrame v2 写入、Schema 合并与写入分布模式维护表通过存储过程执行快照过期、文件重写等维护操作Spark 配置Catalog 配置、SQL 扩展、读写选项的完整参考至此你已经掌握了在 Spark 3 中运行 Iceberg 的完整链路引入运行时 → 配置 Catalog → 建表 → 写入 → 查询 → 理解类型映射。接下来可以结合 Spark 配置 与 表配置 文档把这份入门方案落地到你的真实数据仓库中。赞分享数据湖湖仓一体大数据【免费下载链接】icebergApache Iceberg项目地址https://gitcode.com/gh_mirrors/iceberg5/iceberg点击查看免费下载相关推荐Apache Iceberg 与 Spark 快速上手Catalog 配置、建表读写与类型兼容性实战指南Apache Iceberg 与 Spark 快速上手Catalog 配置、建表读写与类型兼容性实战指南 Apache Iceberg 是一套面向大规模分析型数据湖大数据数据存储Apache Iceberg 入门指南在 Spark 3 中快速上手 Iceberg 表Apache Iceberg 入门指南在 Spark 3 中快速上手 Iceberg 表 导读 本文是 Apache Iceberg 官方文档中的 Spark大数据数据湖OLAP数据存储Apache Iceberg Spark 快速上手指南从 Docker Compose 到表读写与 Catalog 配置Apache Iceberg Spark 快速上手指南从 Docker Compose 到表读写与 Catalog 配置 Apache Iceberg™数据湖大数据数据存储上一篇幻兽帕鲁存档编辑终极指南3分钟掌握游戏数据自由下一篇如何在Chrome浏览器中轻松下载网页视频Video Download Helper扩展使用指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考