Databricks Autoloader：动态Schema的文件存储格式及读取方式咨询

一、Schema的存储方式与格式 #

推荐格式：JSON #

Spark的Schema（StructType）可直接序列化为JSON格式，该格式可读性强、体积小，且能被Spark直接解析还原为StructType，是存储Schema的最优选择。

存储步骤 #

1. 从样本数据（或任意一批符合目标Schema的入站数据）中提取Schema；
2. 将Schema序列化为JSON字符串；
3. 将JSON字符串写入云存储路径（如DBFS、S3、ADLS等）保存为文件。

二、Autoloader读取预存Schema的方法 #

• 不能用cloudFiles.schemaLocation读取自定义Schema文件：这个参数的作用是让Autoloader自动推断Schema并将其保存到指定的checkpoint路径，用于后续流任务的Schema演化，并非读取用户预存的Schema文件。
• 正确方式：读取JSON文件解析为StructType，传入.schema()参数：先加载预存的Schema JSON文件，解析成Spark的StructType对象，再在readStream时通过.schema()指定该Schema。

三、代码示例 #

1. 存储Schema到JSON文件 #

# 从样本数据中读取Schema（以读取一批parquet文件为例）
sample_df = spark.read.format("parquet").load("<path_to_sample_data>")
target_schema = sample_df.schema

# 将Schema序列化为JSON字符串
schema_json = target_schema.json()

# 将JSON字符串写入存储路径（示例为DBFS路径，可替换为其他云存储路径）
dbutils.fs.put("<path_to_save_schema>/schema.json", schema_json, overwrite=True)

2. 读取预存Schema并用于Autoloader流读取 #

# 读取预存的Schema JSON文件
schema_json_str = dbutils.fs.head("<path_to_save_schema>/schema.json")

# 将JSON字符串解析为Spark StructType
from pyspark.sql.types import StructType
custom_schema = StructType.fromJson(schema_json_str)

# 使用Autoloader读取流数据，指定预存的Schema
stream_df = spark.readStream \
    .format("cloudFiles") \
    .schema(custom_schema) \
    .option("cloudFiles.schemaLocation", "<path_to_checkpoint>") \
    .option("cloudFiles.format", "parquet") \
    .load("<path_to_source_data>")

注意：如果后续入站数据Schema发生变化，Autoloader可根据cloudFiles.schemaLocation路径中的元数据进行Schema演化（需开启相关配置，如cloudFiles.schemaEvolutionMode）；若需强制使用固定Schema，可忽略演化配置。

内容的提问来源于stack exchange，提问作者Thiru Balaji G