如何在Autoloader中启用Schema Evolution以自动新增摄入列？

要让Autoloader在新列到来时自动添加并摄入，必须配置以下关键参数，否则会因未知列导致流任务失败：

核心配置项 #

• 指定schema存储路径：设置cloudFiles.schemaLocation，用于Spark持久化和跟踪schema的变化，这是自动进化的基础。
• 开启schema自动进化：将cloudFiles.schemaEvolutionMode设为addNewColumns，允许任务自动识别并添加新列。
• 可选：自动推断初始schema：如果没有预先定义schema，设置cloudFiles.inferSchema为true让Spark自动推断初始结构。

代码示例（Python） #

stream_df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "parquet")  # 替换为你的文件格式（csv/json等）
    .option("cloudFiles.schemaLocation", "/dbfs/path/to/schema/storage")  # 持久化schema的路径
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")  # 开启新列自动添加
    .option("cloudFiles.inferSchema", "true")  # 自动推断初始schema（按需开启）
    .load("/dbfs/path/to/source/data")

常见错误原因 #

• 未配置schemaLocation：Spark无法保存schema元数据，无法跟踪后续的列变化。
• schemaEvolutionMode未设置为addNewColumns：默认值为failOnNewColumns，遇到新列直接触发任务失败，与你的报错匹配。
• 使用硬编码固定schema：如果显式指定了schema参数，即使开启进化，Spark也不会自动添加新列，需改用从schemaLocation读取或自动推断的方式。

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