如何实现Python脚本执行完成后自动触发Azure Data Factory的ETL流程？

这里有几个实用的方案，帮你摆脱固定时间调度的等待，让ADF的ETL流水线在Python脚本完成数据上传后自动启动：

方案1：Python脚本直接调用ADF流水线（最直接可控） #

你可以借助Azure SDK for Python，在数据上传完成的代码末尾添加触发ADF流水线的逻辑，确保脚本成功执行后立即触发ETL。

步骤： #

1. 先安装Azure Data Factory的Python SDK：

   pip install azure-mgmt-datafactory azure-identity
   

2. 在Python脚本末尾添加触发代码（示例用服务主体认证，这是生产环境常用的方式）：

   from azure.identity import ClientSecretCredential
   from azure.mgmt.datafactory import DataFactoryManagementClient
   from azure.mgmt.datafactory.models import CreateRunParameters
   
   # 配置你的Azure信息
   tenant_id = "<你的租户ID>"
   client_id = "<你的服务主体客户端ID>"
   client_secret = "<你的服务主体密钥>"
   subscription_id = "<你的订阅ID>"
   resource_group_name = "<你的资源组名>"
   data_factory_name = "<你的ADF名称>"
   pipeline_name = "<要触发的流水线名称>"
   
   # 创建认证凭据和ADF客户端
   credential = ClientSecretCredential(tenant_id, client_id, client_secret)
   adf_client = DataFactoryManagementClient(credential, subscription_id)
   
   # 触发流水线
   try:
       response = adf_client.pipelines.create_run(
           resource_group_name,
           data_factory_name,
           pipeline_name,
           parameters={}  # 如果你的流水线需要参数，在这里传入
       )
       print(f"成功触发ADF流水线，运行ID：{response.run_id}")
   except Exception as e:
       print(f"触发流水线失败：{str(e)}")
       # 这里可以添加错误处理逻辑，比如发送告警
   

3. 注意：要确保你的服务主体拥有Data Factory Contributor或足够的权限来触发流水线。

方案2：利用Azure Blob Storage事件触发ADF（解耦性强） #

既然你的Python脚本是把数据上传到Blob Storage，那可以利用Azure的事件网格，当Blob上传完成时自动触发ADF的事件触发器，不需要修改太多Python代码。

步骤： #

1. 在Azure Data Factory中创建事件触发器：
   • 选择你的Blob存储账户，指定原始数据所在的容器
   • 设置触发条件为“Blob创建或修改”，可以进一步过滤文件前缀/后缀（比如只监听.csv或者parquet文件）
2. 优化触发时机（可选但推荐）：
   为了确保所有数据都完全上传完成再触发ETL，可以让Python脚本在所有数据上传完毕后，上传一个标记文件（比如_SUCCESS.txt），然后把ADF的事件触发器设置为只监听这个标记文件的创建。这样就避免了部分文件上传时误触发的问题。
3. 配置事件网格权限：确保Blob存储账户有向ADF发送事件的权限，这个在创建触发器时Azure会自动帮你配置大部分权限，只需要确认即可。

方案3：用Azure Synapse Analytics管道（如果已在使用Synapse） #

如果你已经在用Synapse，逻辑和ADF几乎一致：

• 要么用Python调用Synapse的管道API（和方案1的代码类似，只是SDK换成azure-mgmt-synapse）
• 要么同样用Blob事件触发Synapse管道
  Synapse的优势是如果你的ETL后续需要结合大数据分析、Spark作业等，一体化体验更好，但触发逻辑和ADF无本质区别。

额外注意事项： #

• 错误处理：不管用哪种方案，都要确保只有Python脚本成功完成数据上传后才触发ETL。比如在Python脚本里用try-except块，只有在try块顺利执行完才触发ADF。
• 重试机制：ADF本身有重试机制，但你也可以在Python代码里添加简单的重试逻辑，避免偶尔的网络问题导致触发失败。
• 日志监控：可以在ADF里配置流水线的运行日志，或者在Python脚本里记录触发状态，方便后续排查问题。

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