Iceberg配置与类型

Spark类型和Iceberg类型对应关系

Spark和Iceberg支持不同的类型集。Iceberg会自动进行类型转换，但并非所有组合都支持，因此在设计表的列类型之前，您需要了解Iceberg中的类型转换。

Spark类型对应的Iceberg类型

此类型转换表描述了Spark类型如何转换为Iceberg类型。转换在创建Iceberg表和通过Spark写入Iceberg表时都适用。

表1 Spark类型对应的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

该表基于创建表时的转换表示。实际上，写入时支持类型更广泛。以下是写入时的一些要点：

Iceberg数值类型（`integer`、`long`、`float`、`double`、`decimal`）在写入时支持提升。例如，您可以将Spark类型`short`、`byte`、`integer`、`long`写入Iceberg类型`long`。
您可以使用Spark`binary`类型写入Iceberg`fixed`类型。请注意，将执行长度断言。

Iceberg类型对应的Spark类型

此类型转换表描述了Iceberg类型如何转换为Spark类型。转换在通过Spark读取Iceberg表时适用。

表2 Iceberg类型对应的Spark类型
Iceberg	Spark
boolean	boolean
integer	integer
long	long
float	float
double	double
date	date
time	不支持
timestampwithtimezone	timestamp
timestampwithouttimezone	timestamp_ntz
string	string
uuid	string
fixed	binary
binary	binary
decimal	decimal
struct	struct
list	array
map	map
nanosecondtimestamp	不支持
nanosecondtimestampwithtimezone	不支持
unknown	不支持
variant	variant 说明：支持Spark4.0+
geometry	不支持
geography	不支持

运行时配置

配置设置的优先级

Iceberg允许在不同级别指定配置。读或写操作的有效配置按以下优先级顺序确定：

DataSourceAPI读/写选项：在读/写操作中显式传递给.option(...)。
Spark会话配置：通过spark.conf.set(...)、spark-defaults.conf或spark-submit中的--conf全局设置。
表属性：通过ALTER TABLE SET TBLPROPERTIES在Iceberg表上定义。
默认值。

如果未在更高级别定义设置，则使用下一级作为后备。这允许在需要时启用全局默认值，保持灵活性。

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表中写入数据。
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）	每个微批次的最大行数（软限制）。

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表中写入数据。
delete-granularity	file	覆盖此表的写入删除粒度。可选值： file（默认值）：按文件级别删除，生成positiondelete文件指向具体的数据文件行 partition：按分区级别删除，生成partitiondelete文件标记整个分区为已删除（表属性write.delete.granularity默认值为partition）影响DELETE操作的执行效率。file级别更精确但产生更多小文件；partition级别更高效但删除整个分区。