Databricks指定Schema类型后保存加载丢失user_id值问题排查

核心问题 #

当Schema中嵌套字段geo.latitude和geo.longitude指定为DoubleType时，读取JSON生成的DataFrame中user_id值正常，但保存为Delta格式重新加载后user_id变为null；将这两个字段改为StringType则无此问题。

触发原因分析 #

1. 嵌套类型序列化隐性bug：Spark/Delta在处理嵌套结构体中的DoubleType字段时，可能出现序列化/反序列化的字段偏移问题，导致后续字段（如user_id）无法被正确写入或解析。
2. Schema推断不一致：读取Delta表时自动推断的Schema与原始定义的Schema存在隐性差异，尽管user_id类型仍为StringType，但嵌套字段的类型处理异常会牵连后续字段值丢失。
3. 空分区字段的隐性影响：代码中date_partition从文件名提取日期，但临时目录下的JSON文件无日期路径，导致该字段为null，repartition时所有数据进入同一分区，可能触发分区处理的隐性异常，放大嵌套类型的解析问题。

验证步骤 #

1. 确认保存前的DataFrame中user_id确实非空：

   df.select("user_id").show()
   

2. 检查Delta表的实际Schema是否与原始Schema一致：

   spark.read.format("delta").load(f"{d}/df").printSchema()
   

3. 对比原始mockschema，重点查看user_id及嵌套geo字段的类型定义是否匹配。

解决方案 #

方案1：写入Delta时显式指定Schema #

避免自动推断Schema带来的不一致，写入时强制绑定原始Schema：

df_repartitioned.write.format(output_format) \
  .mode("overwrite") \
  .schema(mockschema) \
  .save(f"{d}/df")

方案2：先以StringType读取再转换类型 #

先将geo.latitude和geo.longitude以StringType读取，再显式转换为DoubleType，规避嵌套类型解析的隐性问题：

# 定义临时Schema，经纬度用StringType
temp_schema = StructType([
    StructField("bot", BooleanType(), True),
    StructField("channel", StringType(), True),
    StructField("chills", StringType(), True),
    StructField("cookies", MapType(StringType(), StringType()), True),
    StructField("geo", StructType([
        StructField("city", StringType(), True),
        StructField("country", StringType(), True),
        StructField("region", StringType(), True),
        StructField("latitude", StringType(), True),
        StructField("longitude", StringType(), True)
    ]), True),
    StructField("ip", StringType(), True),
    StructField("timestamp", LongType(), True),
    StructField("ua", StringType(), True),
    StructField("url", StringType(), True),
    StructField("user_id", StringType(), True),
    StructField("vibes", StringType(), True)
])

# 读取后转换经纬度类型
df = spark.read.schema(temp_schema).json(d)
df = df.withColumn(
    "geo",
    expr("struct(geo.city, geo.country, geo.region, cast(geo.latitude as double) as latitude, cast(geo.longitude as double) as longitude)")
)

方案3：修复空分区字段问题 #

给date_partition设置默认值，避免分区字段为null：

from pyspark.sql.functions import coalesce, current_date

df = df.withColumn(
    "date_partition", 
    coalesce(col("date_partition"), date_format(current_date(), "yyyy-MM-dd"))
)

方案4：升级Spark/Delta版本 #

该问题可能是特定版本的已知bug，升级到Spark 3.3+或Databricks Lakehouse Runtime 11.3+版本，可修复此类嵌套类型的序列化异常。

复现代码修正 #

原代码存在笔误，spark.createDataFrame([mock])应改为spark.createDataFrame([mock_data])，否则无法生成正确的JSON测试数据：

spark.createDataFrame(
    [mock_data]  # 修正此处
).write.mode("overwrite").format("json").save(d)

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