如何在自动加载器上启用模式演化

Question

我想在自动加载器上进行 schemaEvolution，以便在新列到达和摄取时添加它们

当我单击显示流时应该失败并告诉我有未知列

Answer 1

模式位置目录会随着时间的推移跟踪您的数据模式

了解更多关于 在 Auto Loader 中配置模式推断和演化

当您为选项

cloudFiles.schemaLocation

指定目标目录时，它会启用模式推断和演化。

如果您愿意，可以将同一目录用于

checkpointLocation

。

以下是语法：

(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target")
)

结果：

   df = (spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .option("cloudFiles.schemaLocation", schema_loc)
      .load(Source_data_loc)
      .writeStream
      .option("checkpointLocation", schema_loc)
      .start(target_data_loc))