# Iceberg配置与类型 #### Spark类型和Iceberg类型对应关系 Spark和Iceberg支持不同的类型集。Iceberg会自动进行类型转换，但并非所有组合都支持，因此在设计表的列类型之前，您需要了解Iceberg中的类型转换。 **Spark类型对应的Iceberg类型** 此类型转换表描述了Spark类型如何转换为Iceberg类型。转换在创建Iceberg表和通过Spark写入Iceberg表时都适用。 表1Spark类型对应的Iceberg类型 | Spark | Iceberg | |:---|:---| | boolean | boolean | | short | integer | | byte | integer | | integer | integer | | long | long | | float | float | | double | double | | date | date | | timestamp | timestamp with timezone | | timestamp_ntz | timestamp without timezone | | char | string | | varchar | string | | string | string | | binary | binary | | decimal | decimal | | struct | struct | | array | list | | map | map | ![](https://support.huaweicloud.com/devg-aidatalake/public_sys-resources/note_3.0-zh-cn.png) 该表基于创建表时的转换表示。实际上，写入时支持类型更广泛。以下是写入时的一些要点： - Iceberg数值类型（\`integer\`、\`long\`、\`float\`、\`double\`、\`decimal\`）在写入时支持提升。例如，您可以将Spark类型\`short\`、\`byte\`、\`integer\`、\`long\`写入Iceberg类型\`long\`。 - 您可以使用Spark\`binary\`类型写入Iceberg\`fixed\`类型。请注意，将执行长度断言。 **Iceberg类型对应的Spark类型** 此类型转换表描述了Iceberg类型如何转换为Spark类型。转换在通过Spark读取Iceberg表时适用。 表2Iceberg类型对应的Spark类型 | Iceberg | Spark | |:---|:---| | boolean | boolean | | integer | integer | | long | long | | float | float | | double | double | | date | date | | time | 不支持 | | timestamp with timezone | timestamp | | timestamp without timezone | timestamp_ntz | | string | string | | uuid | string | | fixed | binary | | binary | binary | | decimal | decimal | | struct | struct | | list | array | | map | map | | nanosecond timestamp | 不支持 | | nanosecond timestamp with timezone | 不支持 | | unknown | 不支持 | | variant | variant 说明： 支持Spark4.0+ | | geometry | 不支持 | | geography | 不支持 | #### 运行时配置 **配置设置的优先级** Iceberg允许在不同级别指定配置。读或写操作的有效配置按以下优先级顺序确定： 1. DataSourceAPI读/写选项：在读/写操作中显式传递给.option(...)。 2. Spark会话配置：通过spark.conf.set(...)、spark-defaults.conf或spark-submit中的--conf全局设置。 3. 表属性：通过ALTER TABLE SET TBLPROPERTIES在Iceberg表上定义。 4. 默认值。 如果未在更高级别定义设置，则使用下一级作为后备。这允许在需要时启用全局默认值，保持灵活性。 **Spark SQL选项** Iceberg支持使用SparkSQL配置选项设置各种全局行为。可以通过spark、SparkSession设置或Spark提交参数进行设置。例如： ``` //禁用向量化 val spark = SparkSession.builder() .appName("IcebergExample") .master("local[*]") .config("spark.sql.catalog.my_catalog","org.apache.iceberg.spark.SparkCatalog") .config("spark.sql.extensions","org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") .config("spark.sql.iceberg.vectorization.enabled","false") .getOrCreate() ``` | Spark选项 | 默认值 | 描述 | |:---|:---|:---| | spark.sql.iceberg.vectorization.enabled | 继承表属性（Parquet的read.parquet.vectorization.enabled默认true；ORC的read.orc.vectorization.enabled默认false） | 启用数据文件的向量化读取。 - 设置为true时，Iceberg使用基于批次的列向量化（Columnar Vectorization）来加速查询，每次批量处理5000行（由read.parquet.vectorization.batch-size或read.orc.vectorization.batch-size控制）。 - 设置为false时将使用非向量化的逐行读取器，性能较低但兼容性更好。 | | spark.sql.iceberg.parquet.reader-type | ICEBERG | 设置Parquet读取器实现。可选值： - ICEBERG（默认值）：使用Iceberg内置的Parquet读取器。 - COMET：使用Apache DataFusion CometParquet读取器（在JVM中执行I/O和解压缩，但在原生层解码以提高性能；同时将Spark物理计划转换为原生执行）。 | | spark.sql.iceberg.check-nullability | true | 验证写入模式的可空性与表的可空性匹配。 - 设置为true时，如果写入数据中声明为NOT NULL的字段包含null值将抛出异常。 - 设置为false时跳过此检查，允许null值写入NOT NULL字段（可能导致后续查询出现问题）。 | | spark.sql.iceberg.check-ordering | true | 验证写入模式列顺序与表模式顺序匹配。 - 设置为true时，如果写入数据的列顺序与表定义不一致将抛出异常。 - 设置为false时跳过此检查，允许按名称匹配列而非按位置匹配。 | | spark.sql.iceberg.planning.preserve-data-grouping | false | - 为true时，将同一分区的扫描任务放在同一读取分片中，用于存储分区连接（Storage Partitioned Joins），可优化分区内数据的局部性。 - 设置为false时，扫描任务按默认策略分配，不保证同一分区的数据在同一分片中。 | | spark.sql.iceberg.aggregate-push-down.enabled | true | 启用聚合函数（MAX、MIN、COUNT）的下推。 - 设置为true时，Iceberg会尝试将聚合计算推入数据文件读取层，利用文件级统计信息（如min/max值）跳过不必要的数据读取。 - 设置为false时，所有聚合计算在Spark层完成。 | | spark.sql.iceberg.distribution-mode | 动态决定： - 表有排序顺序时为range - 表有分区时为hash - 否则为none | 控制写入期间的数据分布策略。可选值： - none：不重新分布数据。 - hash：按分区键进行Hash分布。适合数据在分区中均匀分布的场景 - range：按分区键或排序键进行Range分布。适合数据分布倾斜的场景 详细参考[向Iceberg表中写入数据](https://support.huaweicloud.com/devg-aidatalake/aidatalake_05_0023.html)。 | | spark.wap.id | null | 写-审计-发布（WAP）快照暂存ID。 设置后，新快照将以staged状态创建并关联此ID，需手动发布。设置为null时快照直接提交。前提需要设置表属性write.wap.enabled=true。 | | spark.wap.branch | null | WAP分支名称。设置后，快照将提交到指定分支而非主分支，用于隔离开发/测试环境。设置为null时快照提交到当前分支。 前提需要设置表属性write.wap.enabled=true。 | | spark.sql.iceberg.compression-codec | 继承表属性（write.parquet.compression-codec默认gzip；write.avro.compression-codec默认gzip；write.orc.compression-codec默认zlib） | 写入压缩编解码器。不同文件格式支持不同的编解码器： - Parquet支持uncompressed、snappy、gzip、lz4、zstd、brotli、lzop、bz2、zstd-no-seek。 - Avro支持null（无压缩）、deflate、snappy、bzip2、xz、lz4、zstd。 - ORC支持none、snappy、zlib、lz4、lzo、zstd。 | | spark.sql.iceberg.compression-level | 继承表属性（write.parquet.compression-level默认null；write.avro.compression-level默认null） | Parquet/Avro的压缩级别。通常为1-9的整数，数值越高压缩率越高但速度越慢。默认null表示使用编解码器的默认级别。仅对支持级别的编解码器（如gzip、zstd、deflate）有效。 | | spark.sql.iceberg.compression-strategy | 继承表属性write.orc.compression-strategy（默认speed） | ORC的压缩策略。可选值： - speed（默认值）：优先压缩速度，减少CPU占用。 - compression：优先压缩率，减少存储空间。 | | spark.sql.iceberg.data-planning-mode | 继承表属性read.data-planning-mode（默认AUTO） | 数据文件的扫描规划模式。可选值： - AUTO（默认值）：根据manifest文件数量自动选择local或distributed。 - LOCAL：所有文件规划在Driver进程中执行。 - DISTRIBUTED：文件规划分布到Spark Executor执行（要求spark.driver.maxResultSize\>=256MB，否则自动回退到local）。 | | spark.sql.iceberg.delete-planning-mode | 继承表属性read.delete-planning-mode（默认AUTO） | 删除文件的扫描规划模式。可选值： - AUTO（默认值）：自动选择。 - LOCAL：Driver端规划。 - DISTRIBUTED：分布式规划（要求spark.driver.maxResultSize\>=256MB，否则自动回退到local）。 | | spark.sql.iceberg.advisory-partition-size | 继承表属性write.target-file-size-bytes（默认536870912即512MB） | 启用Spark自适应查询执行（AQE）时用于写入表的建议分区大小（字节）。用于调整输出文件大小，影响每个任务写入的数据量。 | | spark.sql.iceberg.locality.enabled | false | 报告Spark任务在执行器上的本地化信息。 - 设置为true时，Iceberg会向Spark报告数据块的本地化位置，使Spark尽量将计算任务调度到存储数据的节点上，减少网络传输。对于HDFS等支持数据本地性的存储系统有效，对OBS/S3等对象存储无效。 - 设置为false时，Spark随机分配任务到执行器。 | | spark.sql.iceberg.executor-cache.enabled | true | 启用执行器端缓存。 - 设置为true时，Iceberg在执行器端缓存删除文件（DeleteFiles）等内容，避免在Row级操作时重复读取。 - 设置为false时，每次操作都从存储重新读取文件。 | | spark.sql.iceberg.executor-cache.timeout | 10（10分钟） | 缓存条目的超时时间（分钟）。超过此时间未访问的缓存条目将被清除。 | | spark.sql.iceberg.executor-cache.max-entry-size | 67108864（64MB） | 每个缓存条目的最大大小（字节）。超过此大小的文件将不会被缓存。 | | spark.sql.iceberg.executor-cache.max-total-size | 134217728（128MB） | 执行器缓存的最大总大小（字节）。当缓存总量超过此值时，将按LRU（最近最少使用）策略淘汰旧条目。 | | spark.sql.iceberg.executor-cache.locality.enabled | false | - 启用感知本地化的执行器缓存使用。 设置为true时，缓存优先在数据本地化的执行器上命中，减少跨节点缓存查找。 - 设置为false时，缓存查找不考虑数据本地性。 | | spark.sql.iceberg.merge-schema | false | 启用修改表模式以匹配写入模式。 - 设置为true时，如果写入数据包含表中不存在的新列，Iceberg会自动将这些列添加到表模式中。仅添加缺失的列，不会删除或修改已有列。 - 设置为false时，如果写入数据包含表模式中不存在的列将抛出异常。 | | spark.sql.iceberg.report-column-stats | true | 如果可用，向Spark的基于成本的优化器（CBO）报告Puffin表统计信息。 - 设置为true时，Iceberg会将列统计信息（如NDV、null值计数）传递给SparkCBO用于优化查询计划。 - 设置为false时不报告。 必须同时启用SparkCBO（spark.sql.cbo.enabled=true）才能生效。 | **读取选项** Spark读取选项在配置DataFrame Reader时传递，如下所示： ``` //时间旅行 spark.read .option("snapshot-id",10963874102873L) .table("catalog.db.table") ``` | Spark选项 | 默认值 | 描述 | |:---|:---|:---| | snapshot-id | null | 要读取的表快照的快照ID（Long类型）。用于时间旅行（TimeTravel）功能，读取指定历史版本的数据。与as-of-timestamp互斥。设置为null时读取最新快照。 | | as-of-timestamp | null | 毫秒级时间戳（Long类型）。使用的快照将是此时间点之前最新的快照。用于基于时间的历史数据查询。与snapshot-id互斥，同时设置时优先使用snapshot-id。设置为null时读取最新快照。 | | split-size | 继承表属性（read.split.target-size默认134217728即128MB；read.split.metadata-target-size默认33554432即32MB） | 同时覆盖此表的read.split.target-size（数据文件分割目标大小）和read.split.metadata-target-size（元数据文件分割目标大小）。控制每个Spark任务处理的数据量。值越小产生的任务越多。 | | lookback | 继承表属性read.split.planning-lookback（默认10） | 覆盖此表的read.split.planning-lookback。在规划splits时向前回溯的文件数量，用于将小文件合并到同一个split中。值越大合并越多但规划时间越长。 | | file-open-cost | 继承表属性read.split.open-file-cost（默认4194304即4MB） | 覆盖此表的read.split.open-file-cost。打开一个文件的估算成本（字节），用于split规划时的成本优化。 | | vectorization-enabled | 继承表属性（Parquet的read.parquet.vectorization.enabled默认true；ORC的read.orc.vectorization.enabled默认false） | 覆盖此表的向量化读取开关。设置为true启用Parquet/ORC向量化读取，使用列式批量处理提升查询性能。设置为false时使用非向量化的逐行读取器，性能较低但兼容性更好。 | | batch-size | 继承表属性（read.parquet.vectorization.batch-size默认5000；read.orc.vectorization.batch-size默认5000） | 覆盖此表的向量化批次大小。即每批处理的行数。较大的批次（如10000）提高吞吐量但增加内存消耗；较小的批次（如1000）减少内存占用但降低吞吐量。 | | stream-from-timestamp | null | 毫秒级时间戳（Long类型），用于流式传输的起始点。如果早于最旧的已知祖先快照，则使用最旧的快照。设置为null时从最新快照开始流式读取。 | | streaming-max-files-per-micro-batch | Integer.MAX_VALUE（2147483647） | 每个微批次的最大文件数。用于控制结构化流（StructuredStreaming）中每个批次的处理文件数量。设置为较小值可避免批次过大导致延迟增加。 | | streaming-max-rows-per-micro-batch | Integer.MAX_VALUE（2147483647） | 每个微批次的最大行数（软限制）。 | ![](https://support.huaweicloud.com/devg-aidatalake/public_sys-resources/note_3.0-zh-cn.png) streaming-max-rows-per-micro-batch选项设置了一个"软最大值"，一个批次将始终包含下一个未处理数据文件中的所有行，但如果包含额外文件会超过软最大值限制，则不会包含这些文件。 **写入选项** Spark写入选项在配置DataFrameWriterV2时传递，如下所示： ``` //使用Avro而不是Parquet写入 df.writeTo("catalog.db.table") .option("write-format","avro") .option("snapshot-property.key","value") .append() ``` | Spark选项 | 默认值 | 描述 | |:---|:---|:---| | write-format | 继承表属性write.format.default（默认parquet） | 此写入操作使用的文件格式。可选值： - parquet（默认） - avro - orc | | target-file-size-bytes | 继承表属性write.target-file-size-bytes（默认536870912即512MB） | 覆盖此表的write.target-file-size-bytes。控制每个输出数据文件的目标大小。较小的值产生更多小文件（查询时可能增加扫描开销），较大的值减少文件数但增加单文件大小（可能增加内存消耗）。 | | check-nullability | true | 设置字段的可空性检查。设置为true时，如果写入数据中声明为NOTNULL的字段包含null值将抛出异常。设置为false时跳过此检查，允许null值写入NOTNULL字段（可能导致后续查询行为异常）。 | | snapshot-property.*custom-key* | null | 在快照摘要中添加具有自定义键和相应值的条目。例如snapshot-property.user=alice会在快照元数据中添加{"user":"alice"}。用于审计追踪或变更标记。（DSv2API需要snapshot-property.前缀）。 | | fanout-enabled | false | 覆盖此表的write.spark.fanout.enabled。 - 设置为true时，将数据按分区扇出写入，避免单个任务写入过多分区（每个分区一个写入任务），适合写入大量分区的场景，但会增加Shuffle开销和内存消耗。 - 设置为false时，单个任务写入多个分区，可能在写入大量分区时产生内存压力。 | | check-ordering | true | 检查输入模式和表模式是否相同。 - 设置为true时，如果写入数据的列顺序与表定义不一致将抛出异常。 - 设置为false时跳过此检查，允许按名称匹配列而非按位置匹配。 | | isolation-level | null | Dataframe覆盖操作所需的隔离级别。可选值： - null（默认）：不进行检查，适合幂等写入。 - serializable：检查目标分区中的并发插入或删除，确保完全隔离。 - snapshot：仅检查目标分区中的并发删除，允许并发插入。 表属性默认： - write.delete.isolation-level默认serializable。 - write.update.isolation-level默认serializable。 - write.merge.isolation-level默认serializable。 | | validate-from-snapshot-id | null | 如果设置了隔离级别，则为检查表中并发写入冲突的基线快照ID（Long类型）。应该是任何读取表之前的快照。可以通过Table API或Snapshots表获取。如果为null，则使用表的最旧已知快照。 | | compression-codec | 继承表属性（Parquet的write.parquet.compression-codec默认gzip；Avro的write.avro.compression-codec默认gzip；ORC的write.orc.compression-codec默认zlib） | 覆盖此表的写入压缩编解码器。 - Parquet可选值：uncompressed、snappy、gzip、lz4、zstd、brotli、lzop、bz2、zstd-no-seek。 - Avro可选值：null（无压缩）、deflate、snappy、bzip2、xz、lz4、zstd。 - ORC可选值：none、snappy、zlib、lz4、lzo、zstd。 | | compression-level | 继承表属性（write.parquet.compression-level默认null；write.avro.compression-level默认null） | 覆盖此表的Parquet和Avro表的写入压缩级别。整数类型，通常1-9，数值越高压缩率越高但速度越慢。默认null表示使用编解码器的默认级别。仅对支持级别的编解码器（如gzip、zstd、deflate）有效。 | | compression-strategy | 继承表属性write.orc.compression-strategy（默认speed） | 覆盖此表的ORC表的写入压缩策略。可选值： - speed（默认）：优先压缩速度，减少CPU占用。 - compression：优先压缩率，减少存储空间。 | | distribution-mode | 动态决定（表有排序顺序为range，有分区为hash，否则none） | 覆盖此表的写入分布模式。可选值： - none：不重新分布数据。 - hash：按分区键Hash分布。适合数据在分区中均匀分布的场景。 - range：按分区键或排序键Range分布。适合数据分布倾斜的场景。 表属性默认： write.delete.distribution-mode、write.update.distribution-mode、write.merge.distribution-mode默认均为hash 详细参考[向Iceberg表中写入数据](https://support.huaweicloud.com/devg-aidatalake/aidatalake_05_0023.html)。 | | delete-granularity | file | 覆盖此表的写入删除粒度。可选值： - file（默认值）：按文件级别删除，生成positiondelete文件指向具体的数据文件行 - partition：按分区级别删除，生成partitiondelete文件标记整个分区为已删除（表属性write.delete.granularity默认值为partition） 影响DELETE操作的执行效率。file级别更精确但产生更多小文件；partition级别更高效但删除整个分区。 |