如何配置Spark Autoloader指定Azure存储账户，避免跨存储事件读取错误

问题原因 #

当配置cloudFiles.resourceGroup参数时，Spark Auto Loader会监听整个资源组下所有存储账户的文件事件。切换source_path到STORAGE_ACCOUNT_B后，作业的检查点仍保留着之前针对STORAGE_ACCOUNT_A的容器预期值，导致新的事件（来自STORAGE_ACCOUNT_B）与预期不符，触发报错。

解决方案 #

1. 新增存储账户限定参数 #

在Auto Loader配置中添加cloudFiles.accountName参数，明确指定要监听的目标存储账户，这样Auto Loader会自动过滤掉资源组内其他存储账户的事件：

df = spark.readStream.format("cloudFiles")
.option("cloudFiles.useNotifications", "true")
.option("cloudFiles.tenantId", "XXXX")
.option("cloudFiles.subscriptionID", "XXXX")
.option("cloudFiles.resourceGroup", "XXXX")
.option("cloudFiles.clientId", "XXXX")
.option("cloudFiles.clientSecret", "XXXX")
.option("cloudFiles.accountName", "STORAGE_ACCOUNT_B")  # 限定仅监听此存储账户
.option("cloudFiles.format", "csv")
.option("header", "true")
.schema(schema)
.load(source_path)

2. 更换作业检查点目录 #

旧的检查点目录保存了作业之前的状态（包括预期的容器信息），必须更换新的检查点路径，避免作业沿用旧状态导致冲突。在writeStream配置中修改checkpointLocation：

df.writeStream
.format("delta")
.option("checkpointLocation", "/new/checkpoint/path/in/STORAGE_ACCOUNT_B")  # 新的检查点路径
.start("/target/path")

3. 手动指定存储账户级事件订阅 #

如果需要更精确的控制，可以手动为STORAGE_ACCOUNT_B的目标容器创建存储账户级别的事件网格订阅（而非资源组级别），然后在Auto Loader中指定该订阅的ID：

df = spark.readStream.format("cloudFiles")
.option("cloudFiles.useNotifications", "true")
.option("cloudFiles.tenantId", "XXXX")
.option("cloudFiles.subscriptionID", "XXXX")
.option("cloudFiles.resourceGroup", "XXXX")
.option("cloudFiles.clientId", "XXXX")
.option("cloudFiles.clientSecret", "XXXX")
.option("cloudFiles.eventSubscriptionId", "YOUR_STORAGE_B_EVENT_SUBSCRIPTION_ID")  # 指定订阅ID
.option("cloudFiles.format", "csv")
.option("header", "true")
.schema(schema)
.load(source_path)

这种方式确保作业只接收来自STORAGE_ACCOUNT_B的事件通知，彻底隔离其他存储账户的事件干扰。

内容的提问来源于stack exchange，提问作者question.it