PySpark流处理、Delta表创建及schemaEvolutionMode相关技术咨询

1. readStream/writeStream中option的调用顺序是否有要求？ #

没有特定顺序要求。Spark的DataFrameReader/DataFrameWriter采用构建器模式，每个.option()调用只是向配置集合中添加键值对，最终执行.load()/.start()时才会读取所有配置生效。你提供的示例代码完全没问题，调整.option()、.schema()、.format()的顺序不会影响流作业的行为。

2. 创建Delta表时同时使用tableName和location是否合理？ #

完全合理，这是创建外部Delta表的标准方式：

• 指定tableName是在Spark Catalog（或Hive元存储）中注册表的元数据，让你能通过SQL或spark.table()直接访问表；
• 指定location是将表的数据存储到自定义路径，而非Databricks默认的内部表存储路径。

你观察到的现象：

• 同时使用两者时，指定路径下出现.parquet（Delta存储的实际数据文件）、_delta_log（Delta的事务日志）是正常的Delta Lake存储结构；checkpoint文件则是流写入作业时配置的检查点路径，和表创建本身无关，是writeStream.option("checkpointLocation", ...)设置的结果。
• 仅使用tableName时创建的是内部（管理）表，数据存储在Databricks默认的元数据关联路径，因此能直接在Catalog中看到。

另外你提供的Delta表创建语法是可行的，这是通过Delta Lake API创建外部表的正确写法：

(DeltaTable.createIfNotExists(spark) 
      .tableName("%s.%s_%s" % (layer, domain, deltaTable)) 
      .addColumn("x", "INTEGER")
      .location(path)
      .execute())

该语句会检查指定的三层命名空间（layer.domain.deltaTable）下的表是否存在，不存在则创建带有指定列的外部Delta表，数据存储在path指定的位置。

3. readStream输出中未显示schemaEvolutionMode对应的rescued_data列如何解决？ #

rescued_data是Databricks Auto Loader（即format("cloudFile")）的特性，仅在满足以下条件时才会生成：

1. 必须启用Auto Loader的schema演进：在readStream中设置option("cloudFiles.schemaEvolutionMode", "addNewColumns")（或其他允许schema演进的模式，如rescue）；
2. 指定schema存储路径：必须设置option("cloudFiles.schemaLocation", "<your-schema-path>")，Auto Loader需要该路径存储schema的历史版本，才能识别新字段；
3. 数据中存在原schema未定义的字段：只有当流入的数据包含你指定的.schema(schema)中没有的字段时，rescued_data列才会生成，用来存放这些未匹配的字段。

如果以上条件都满足仍未出现该列，可以检查：

• 确认流作业已经读取了包含新字段的数据；
• 检查是否误将cloudFiles.schemaEvolutionMode设为了failOnNewColumns（该模式会在遇到新字段时报错，不会生成rescued_data）。

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