Azure Data Factory存储事件触发器未触发管道问题求助

问题根源分析 #

你遇到的问题主要有两个可能原因：

1. Blob名称格式错误：代码中Blob名称以/开头，不符合Azure Blob命名规范，会导致触发器的路径过滤规则无法匹配到目标Blob。
2. 上传方式未触发完整Blob创建事件：当使用分块或流上传时，若未完成最终的提交/刷新操作，Azure存储不会触发BlobCreated事件，进而导致ADF触发器不生效。

修复后的代码 #

以下是调整后的TypeScript代码，同时解决上述两个问题：

import { BlobServiceClient } from "@azure/storage-blob"

async function handleWriteToBlob(blobData, insurance) {
  const jsonData = sendJson(blobData);
  const containerName = "raw-layer";
  const fileName = blobData['timestamp'];
  // 移除Blob名称开头的斜杠，符合Azure Blob命名规范
  const blobName = `folder_name/${insurance}/${insurance}-insurance_form-${fileName.toISOString()}.json`;

  const blobServiceClient = new BlobServiceClient(`https://${accountName}.blob.core.windows.net?${sasToken}`);
  const containerClient = blobServiceClient.getContainerClient(containerName);
  const blockBlobClient = containerClient.getBlockBlobClient(blobName);

  const options = {
    blobHTTPHeaders: {
      blobContentType: 'application/json'
    },
    bufferThreshold: 4 * 1024 * 1024, // 4MB
    maxBuffers: 20 // 80MB
  };

  // 使用uploadData方法确保触发BlobCreated事件
  const result = await blockBlobClient.uploadData(jsonData, options);

  if (result.errorCode) {
    throw Error(result.errorCode);
  }

  console.log(`Uploaded BlockBlob ${blobName} successfully`);
}

function sendJson(blobData) {
  const jsonStr = JSON.stringify(blobData);
  console.log(jsonStr);
  return jsonStr;
}

额外说明：流上传时的Flush操作 #

如果场景需要处理大文件、使用流上传，则必须调用流的close()方法（内部自动执行flush和块提交），示例代码如下：

async function handleStreamUploadToBlob(blobData, insurance) {
  const jsonData = sendJson(blobData);
  const containerName = "raw-layer";
  const fileName = blobData['timestamp'];
  const blobName = `folder_name/${insurance}/${insurance}-insurance_form-${fileName.toISOString()}.json`;

  const blobServiceClient = new BlobServiceClient(`https://${accountName}.blob.core.windows.net?${sasToken}`);
  const containerClient = blobServiceClient.getContainerClient(containerName);
  const blockBlobClient = containerClient.getBlockBlobClient(blobName);

  const writeStream = blockBlobClient.createWriteStream({
    blobHTTPHeaders: {
      blobContentType: 'application/json'
    }
  });

  // 写入数据
  writeStream.write(jsonData);
  // 关闭流，自动完成flush和块提交，触发BlobCreated事件
  writeStream.end();

  return new Promise((resolve, reject) => {
    writeStream.on('finish', resolve);
    writeStream.on('error', reject);
  });
}

验证步骤 #

1. 执行修改后的代码上传Blob
2. 查看ADF触发器的运行历史，确认是否触发管道
3. 若仍未触发，检查ADF触发器的路径过滤规则是否与Blob名称完全匹配（注意大小写、路径分隔符）

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