如何在Logic Apps与Azure Data Factory中获取Blob Index Tag并筛选复制Blob

一、Logic Apps 操作步骤 #

1. 为Blob添加/更新Index Tag #

• 创建Consumption类型的逻辑应用，选择**当Blob添加或修改时（仅限属性）**作为触发器，关联目标存储账户和容器。
• 添加「Azure Blob Storage」连接器的设置Blob索引标记动作：
  • 选择对应存储账户、容器，Blob路径直接从触发器的动态内容中选取（比如触发Blob路径）。
  • 在「标记」字段填入键值对，示例：classification: raw-data、processed: false，直接按键值格式输入即可。
• 保存并测试：上传Blob到目标容器，查看Blob的索引标签确认设置成功。

2. 基于Index Tag判断并复制Blob #

• 在触发器后添加条件控件：
  • 条件规则选择动态内容里的「Blob索引标记」，设置判断逻辑，比如processed等于false。
  • 条件为真时，添加复制Blob动作：指定源存储账户、容器、Blob，以及目标存储账户和容器，完成复制。
  • 复制完成后，可再次调用设置Blob索引标记动作，将源Blob的processed标签改为true，防止重复处理。

二、Azure Data Factory 操作步骤 #

1. 为Blob添加/更新Index Tag #

• 创建管道，选择Azure CLI活动（推荐，操作更直观）：
  • 配置CLI命令，替换占位符为实际信息：

    az storage blob index tag set --account-name your-storage-account --container-name your-container --name @{item().name} --tags classification=raw-data processed=false --auth-mode login
    

    若要批量处理，可结合ForEach活动遍历Blob列表，动态传入Blob名称。
  • 也可使用Web活动调用Storage REST API：请求方法POST，URL为https://your-storage-account.blob.core.windows.net/your-container/your-blob?comp=tags，请求头添加x-ms-version: 2021-06-08，请求体为{"tags":{"classification":"raw-data","processed":"false"}}，并配置MSI认证确保权限。

2. 基于Index Tag筛选并复制Blob #

• 方法一：通过数据集筛选
  • 创建Azure Blob数据集，在「连接」选项卡的「筛选器」中添加动态内容，设置筛选逻辑：

    @equals(activity('Get Blob Tags').output.tags.processed, 'false')
    

    将该数据集作为复制活动的源，即可只复制符合标签条件的Blob。
• 方法二：用Lookup活动批量筛选
  • 添加Lookup活动，源数据集配置为目标Blob容器，在「查询」中添加筛选条件，获取所有带有processed: false标签的Blob列表。
  • 将Lookup活动的输出作为复制活动的源（通过@activity('Lookup Blobs').output.value动态获取），实现批量复制。
• 复制完成后，用Azure CLI/Web活动更新源Blob的processed标签为true。

三、注意事项 #

• 确保Logic Apps/ADF的托管标识/服务主体拥有存储账户的Storage Blob Data Contributor权限，否则无法读写Blob索引标签。
• Blob Index Tag的键值均为字符串类型，判断条件时需严格匹配字符串值（比如用"false"而非布尔值false）。
• Logic Apps可通过触发器配置限制监听特定容器；ADF可设置调度触发器定时执行，或用事件触发器监听Blob创建事件触发处理。

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