Databricks Autoloader运行后无法自动终止问题求助

我是Databricks新手，需要从Azure Blob存储中多个子目录的.parquet文件拉取数据，通过Autoloader处理后，用SQL执行create table ... using delta location ''操作（每个.parquet文件都在父目录下的独立子目录中，需要遍历所有子目录读取）。但测试时，仅3条记录的文件处理完成后，进程不会自动终止，必须设置超时或手动终止。最初在循环中直接调用.start()，后来改成把流存入列表再启动等待，还是要等满超时时间才停止，没法适配后续真实数据规模。数据实际几秒内就加载完成了，但.start()无法自行终止，请问哪里操作有误？

原代码 #

from pathlib import Path
import os
import time
import types
configs = {"fs.azure.account.auth.type": "OAuth",
           "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
           "fs.azure.account.oauth2.client.id": "<azure client id>",
           "fs.azure.account.oauth2.client.secret": dbutils.secrets.get(scope="<scope>",key="<your key>"),
           "fs.azure.account.oauth2.client.endpoint": "https://login.microsoftonline.com/<tenantId>/oauth2/token"}

#commented out things really only have to be run once
#dbutils.fs.unmount("/mounted/data_location")
#dbutils.fs.mount(
#  source = "abfss://data_location@<blob>.dfs.core.windows.net/",
#  mount_point = "/mounted/data_location",
#  extra_configs = configs)

#dbutils.fs.mount(
#  source = "abfss://<location>@<blob>.dfs.core.windows.net/",
#  mount_point = "/mounted/dbfs/",
#  extra_configs = configs)

#dbutils.fs.mount(
#  source = "abfss://<location>@<blob>.dfs.core.windows.net/",
#  mount_point = "/mounted/tmp/",
#  extra_configs = configs)

#create a unique dir for each schema load attempt
snapshot = str(int(time.time()))
#source directory for data
info = dbutils.fs.ls("/mounted/data_location")
block_sz = 1024
threads = []

#played with these & re-using schema location instead of unique location
#.option("overwriteSchema", "true") \
#.option("mergeSchema", "true") \
#.option("allowOverwrites", "true") \

for dir in info:
  display('Starting:',dir.name[:-1])
  #autoload destination
  target_location = dir.path.replace('/mounted/data_location','/mounted/dbfs/autoload')
  display('Clearing:', target_location)
  #assume it already exists and wipe it out for now for testing
  dbutils.fs.rm(target_location.replace('dbfs:',''), True)
  schema_location = dir.path.replace('/mounted/data_location','/mounted/dbfs/schema/' + snapshot)
  display(target_location, ' ', schema_location)
  Path(schema_location).mkdir(parents=True, exist_ok=True)
  Path(target_location).mkdir(parents=True, exist_ok=True)
  thread = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "parquet") \
    .option("cloudFiles.inferColumnTypes", "True") \
    .option("cloudFiles.schemaLocation", schema_location) \
    .option("pathGlobFilter", "*.parquet") \
    .option('cloudFiles.useIncrementalListing', 'false') \
    .load(dir.path) \
    .writeStream \
    .option("checkpointLocation", schema_location) \
    .option("parquet.block.size", block_sz)

  item = types.SimpleNamespace()
  item.thread = thread
  item.path = target_location
  threads.append(item)
  display('Constructed',dir)

for t in threads:
  work = t.thread.start(t.path)
  t.work = work

for t in threads:
  display(t)
  t.work.awaitTermination(15) #it should already be done in a few seconds, but waits the full 15
  t.work.stop()

display('Script end.')
#dbutils.notebook.exit("complete")

问题根源与修复方案 #

1. 核心问题：误用流处理模式 #

spark.readStream是用于持续监听目录新增文件的流处理场景，默认会一直运行等待新数据写入。而你的需求是一次性读取现有所有.parquet文件，不需要持续监听，完全不需要用流处理模式。

2. 两种修复方案 #

方案一：改用Autoloader批处理模式（推荐）

直接用spark.read搭配cloudFiles格式，一次性读取所有现有文件，处理完成后自动结束进程：

from pathlib import Path
import os
import time
import types

configs = {"fs.azure.account.auth.type": "OAuth",
           "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
           "fs.azure.account.oauth2.client.id": "<azure client id>",
           "fs.azure.account.oauth2.client.secret": dbutils.secrets.get(scope="<scope>",key="<your key>"),
           "fs.azure.account.oauth2.client.endpoint": "https://login.microsoftonline.com/<tenantId>/oauth2/token"}

# 挂载代码仅需运行一次，已注释
# dbutils.fs.unmount("/mounted/data_location")
# dbutils.fs.mount(...)

# 创建唯一目录用于本次schema加载
snapshot = str(int(time.time()))
info = dbutils.fs.ls("/mounted/data_location")
block_sz = 1024

for dir in info:
  display('Starting:', dir.name[:-1])
  target_location = dir.path.replace('/mounted/data_location','/mounted/dbfs/autoload')
  display('Clearing:', target_location)
  dbutils.fs.rm(target_location.replace('dbfs:',''), True)
  schema_location = dir.path.replace('/mounted/data_location','/mounted/dbfs/schema/' + snapshot)
  display(target_location, ' ', schema_location)
  Path(schema_location).mkdir(parents=True, exist_ok=True)
  Path(target_location).mkdir(parents=True, exist_ok=True)
  
  # 改用批处理读取Autoloader数据
  df = spark.read.format("cloudFiles") \
    .option("cloudFiles.format", "parquet") \
    .option("cloudFiles.inferColumnTypes", "True") \
    .option("cloudFiles.schemaLocation", schema_location) \
    .option("pathGlobFilter", "*.parquet") \
    .option('cloudFiles.useIncrementalListing', 'false') \
    .load(dir.path)
  
  # 批处理写入Delta格式到目标位置
  df.write.format("delta") \
    .option("parquet.block.size", block_sz) \
    .save(target_location)
  
  display('Completed processing:', dir.name[:-1])

display('Script end.')
# dbutils.notebook.exit("complete")

方案二：流处理模式下使用Trigger.Once()

如果后续需要支持「一次性读取现有数据+后续监听新增文件」的混合场景，可以保留流处理，但添加trigger(once=True)配置，让流处理执行一次批处理后自动终止：

# 在原writeStream部分添加trigger配置
thread = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "parquet") \
  .option("cloudFiles.inferColumnTypes", "True") \
  .option("cloudFiles.schemaLocation", schema_location) \
  .option("pathGlobFilter", "*.parquet") \
  .option('cloudFiles.useIncrementalListing', 'false') \
  .load(dir.path) \
  .writeStream \
  .option("checkpointLocation", schema_location) \
  .option("parquet.block.size", block_sz) \
  .trigger(once=True)  # 关键配置：执行一次后自动终止
  .start(target_location)

# 等待流处理完成，无需设置超时
thread.awaitTermination()

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