使用Iceberg表格式为DataFrame Schema添加自定义元数据失效如何解决

Iceberg自身有独立的Schema元数据管理体系，默认不会同步Spark端StructField携带的自定义metadata：写入Iceberg时，Spark侧StructField的metadata会被默认过滤，不会存入Iceberg的元数据存储；读取Iceberg表生成DataFrame时，也不会主动把Iceberg存储的字段属性回填到StructField的metadata字段中，就会出现空值情况。

方法1：Iceberg 0.14+版本（推荐） #

Iceberg从0.14版本开始原生支持Spark侧字段元数据的自动同步，只需要在写入和读取前开启配置即可：

# 开启字段元数据自动同步开关
spark.conf.set("spark.sql.iceberg.field.metadata.enabled", "true")

# 正常写入带自定义metadata的DataFrame到Iceberg即可
df.writeTo("catalog.db.table_name").createOrReplace()

# 读取时也会自动回填metadata
read_df = spark.read.table("catalog.db.table_name")
# 直接读取即可拿到自定义元数据
print(read_df.schema.fields[0].metadata)

方法2：低版本Iceberg手动适配 #

如果使用的Iceberg版本低于0.14，需要手动完成元数据的同步和回填：

写入阶段同步元数据 #

先把DataFrame写入Iceberg表，再批量把StructField的metadata设置为Iceberg的字段属性：

from pyspark.sql.types import StructField, StructType, IntegerType, StringType

# 你的带自定义元数据的原始Schema
custom_schema = StructType([
    StructField("id", IntegerType(), False, metadata={"desc": "用户ID", "level": "一级字段"}),
    StructField("name", StringType(), True, metadata={"desc": "用户姓名", "encrypt": "AES"})
])

# 写入数据到Iceberg
df = spark.createDataFrame([(1, "张三"), (2, "李四")], schema=custom_schema)
table_name = "catalog.db.user_info"
df.writeTo(table_name).create()

# 同步所有字段的自定义元数据到Iceberg
for field in custom_schema.fields:
    for k, v in field.metadata.items():
        spark.sql(f"""
        ALTER TABLE {table_name} ALTER COLUMN {field.name} SET TBLPROPERTIES ('{k}' = '{v}')
        """)

读取阶段回填元数据 #

读取Iceberg表时，主动获取Iceberg存储的字段属性，重构带自定义元数据的Schema：

# 读取Iceberg表的字段属性
field_properties = spark.sql(f"SELECT name, properties FROM {table_name}.fields").collect()
prop_map = {row.name: row.properties for row in field_properties}

# 重构带元数据的Schema
original_schema = spark.read.table(table_name).schema
new_fields = []
for field in original_schema.fields:
    merged_meta = {**field.metadata, **prop_map.get(field.name, {})}
    new_fields.append(StructField(field.name, field.dataType, field.nullable, merged_meta))
new_schema = StructType(new_fields)

# 用新Schema加载数据
df_with_meta = spark.read.schema(new_schema).table(table_name)
# 验证元数据
print(df_with_meta.schema.fields[0].metadata)

• 字符串类型的元数据值不需要额外转义，但若元数据值包含单引号，需要提前做转义处理避免ALTER语句执行失败
• 分区字段的元数据同步规则和普通字段完全一致，不需要额外适配

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