如何使用Azure Data Factory将输出错误保存到Blob存储

方案一：使用ADF内置活动直接实现（无需额外服务） #

完全依赖ADF自带组件，无需编写代码：

• 捕获错误输出：在Web活动的失败分支中，通过表达式获取错误内容。需完整错误结构用@activity('你的Web活动名称').error，仅需错误消息用@activity('你的Web活动名称').error.message。
• 存储错误到变量：添加Set Variable活动，创建字符串类型变量（如errorContent），值设为上述表达式。
• 写入Blob：添加Copy Data活动：
  • 数据源选择Inline，类型设为DelimitedText（或Json，按需选择），在Inline dataset的内容字段填入@variables('errorContent')。
  • 目标数据集绑定到你的Blob存储容器，文件名用动态内容生成（如error_@{utcNow('yyyyMMddHHmmss')}.txt），避免文件重名。

方案二：调整Azure Function实现（解决之前的失败问题） #

若之前用Function未成功，大概率是权限或代码逻辑问题，可按以下步骤调整：

• 函数配置：创建HTTP触发的Function，允许匿名访问（或配置ADF托管标识MSI的访问权限）。
• 代码逻辑（以Python为例）：

  import azure.functions as func
  from azure.storage.blob import BlobServiceClient
  import os
  import json
  from datetime import datetime
  
  def main(req: func.HttpRequest) -> func.HttpResponse:
      try:
          req_body = req.get_json()
          # 将错误信息格式化为易读的JSON字符串
          error_content = json.dumps(req_body, indent=2, ensure_ascii=False)
          
          # 通过环境变量获取Blob存储连接字符串
          blob_service_client = BlobServiceClient.from_connection_string(os.environ["BlobStorageConnStr"])
          container_client = blob_service_client.get_container_client("error-logs")
          # 生成唯一文件名
          blob_name = f"error_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}.txt"
          
          container_client.upload_blob(blob_name, error_content, overwrite=True)
          
          return func.HttpResponse("错误信息已保存", status_code=200)
      except Exception as e:
          return func.HttpResponse(f"保存失败: {str(e)}", status_code=500)
  

• ADF配置：在Web活动的失败分支中，调用Function的HTTP端点，请求方法选POST，请求体填@activity('你的Web活动名称').error，确保Function的连接字符串配置正确，且ADF的MSI（或Function身份）拥有Blob存储的写入权限。

方案三：使用Azure Logic Apps无代码实现 #

Logic Apps可视化流程适合快速搭建这类场景：

• 创建Logic App，触发方式选择当收到HTTP请求时，生成专属端点。
• 添加创建Blob动作，连接到目标Blob存储，设置好目标容器，Blob名称用动态内容生成（如error_@{utcNow('yyyyMMddHHmmss')}.txt），内容选择触发动作的Body（即ADF传递的错误信息）。
• 在ADF的Web活动失败分支中，调用这个Logic App的端点，请求体传入@activity('你的Web活动名称').error即可。

内容的提问来源于stack exchange，提问作者Antônio Farias