Spark写入Iceberg非空timestamp分区列报错：无法写入可空值

背景信息 #

• Avro源数据Schema：timestamp为必填无默认值的long类型（逻辑类型timestamp-millis），Schema定义如下：

{"name": "timestamp","type": {"type": "long", "logicalType": "timestamp-millis"}}

• Iceberg表定义：timestamp为非空分区列，SQL定义如下：

timestamp TIMESTAMP NOT NULL
USING iceberg
PARTITIONED BY (days(timestamp))

• 异常信息：写入时抛出：

Cannot write nullable values to non-null column 'timestamp'

• 已尝试无效方案：
  • 添加Spark配置spark.sql.iceberg.check-nullability=false，无效
  • 自定义Schema映射试图让Spark识别timestamp为非空，无效
• 当前环境：Spark 3.3.0、Iceberg 1.2.0
• 写入代码：

spark.read()
    .format("avro")
    .option("recursiveFileLookup", "true")
    .load(getS3aPath())
    .writeTo(getTableNameWithDB())
    .overwritePartitions();

可行解决方案 #

1. 显式强制标记列的非空属性 #

读取Avro数据后，通过Spark API强制将timestamp标记为非空列，覆盖自动推断的可空属性：

import org.apache.spark.sql.functions;

spark.read()
    .format("avro")
    .option("recursiveFileLookup", "true")
    .load(getS3aPath())
    // 强制转换为非空TIMESTAMP类型，同时断言字段不为空
    .withColumn("timestamp", functions.col("timestamp").cast("TIMESTAMP NOT NULL"))
    .writeTo(getTableNameWithDB())
    .overwritePartitions();

Spark 3.3+也支持直接用nonNull()方法简化操作：

.withColumn("timestamp", functions.col("timestamp").nonNull())

2. 读取时指定自定义非空Schema #

手动定义Spark Schema，明确标记timestamp为非空，替代Avro自动推断的Schema：

import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

// 注意：需包含Avro文件中所有字段，此处仅示例timestamp列
StructType customSchema = new StructType(new StructField[]{
    DataTypes.createStructField("timestamp", DataTypes.TimestampType, false)
});

spark.read()
    .format("avro")
    .schema(customSchema)
    .option("recursiveFileLookup", "true")
    .load(getS3aPath())
    .writeTo(getTableNameWithDB())
    .overwritePartitions();

3. 切换Iceberg写入模式 #

尝试使用replace模式替代overwritePartitions，规避分区写入时的空值校验逻辑差异：

spark.read()
    .format("avro")
    .option("recursiveFileLookup", "true")
    .load(getS3aPath())
    .writeTo(getTableNameWithDB())
    .replace();

若需仅覆盖特定分区，可先过滤目标分区数据后再执行replace，或结合partitionFilter指定分区范围。

4. 升级Iceberg版本 #

Iceberg 1.2.0在Spark非空列兼容处理上存在已知问题，升级到1.3.0及以上版本后，对Avro非空字段的推断逻辑更完善，可直接解决该兼容性问题。

内容的提问来源于stack exchange，提问作者Abhi