将Azure Blob Storage变更捕获至Azure Cosmos DB的Azure Function示例方案

数据老陈

2026-10-2

前置依赖配置

提前准备好以下资源，可避免后续部署运行出错：

已创建的Azure存储账户，且订阅下已启用Microsoft.EventGrid资源提供程序
已创建的Azure Cosmos DB账户，提前建好目标数据库和存储事件的容器，分区键建议设为/blobUrl或/eventTime
Azure Function运行环境，本示例使用.NET 8隔离进程模型，生产环境兼容性最优

步骤1：本地配置文件

先配置local.settings.json用于本地调试，部署到Azure时直接在Function配置页替换对应值即可：

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "Azure存储账户连接字符串，用于Function运行时",
    "FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
    "CosmosDBConnection": "Azure Cosmos DB账户连接字符串"
  }
}

步骤2：完整Function代码示例

代码内置Event Grid触发器和Cosmos DB输出绑定，无需手动实现Cosmos客户端写入逻辑：

using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;
using System;
using System.Text.Json.Serialization;

// Blob存储事件的Data字段结构
public class BlobStorageEventData
{
    [JsonPropertyName("api")]
    public string Api { get; set; }

    [JsonPropertyName("clientRequestId")]
    public string ClientRequestId { get; set; }

    [JsonPropertyName("requestId")]
    public string RequestId { get; set; }

    [JsonPropertyName("eTag")]
    public string ETag { get; set; }

    [JsonPropertyName("contentType")]
    public string ContentType { get; set; }

    [JsonPropertyName("contentLength")]
    public long ContentLength { get; set; }

    [JsonPropertyName("blobType")]
    public string BlobType { get; set; }

    [JsonPropertyName("url")]
    public string BlobUrl { get; set; }

    [JsonPropertyName("sequencer")]
    public string Sequencer { get; set; }
}

// 存入Cosmos DB的事件记录结构
public class BlobChangeEventRecord
{
    [JsonPropertyName("id")]
    public string Id { get; set; }

    [JsonPropertyName("eventType")]
    public string EventType { get; set; }

    [JsonPropertyName("eventTime")]
    public DateTime EventTime { get; set; }

    [JsonPropertyName("subject")]
    public string Subject { get; set; }

    [JsonPropertyName("blobDetail")]
    public BlobStorageEventData BlobDetail { get; set; }
}

public class BlobChangeCaptureFunction
{
    private readonly ILogger<BlobChangeCaptureFunction> _logger;

    public BlobChangeCaptureFunction(ILogger<BlobChangeCaptureFunction> logger)
    {
        _logger = logger;
    }

    [Function("BlobChangeCaptureFunction")]
    // Cosmos DB输出绑定，替换为你自己的数据库名、容器名
    [CosmosDBOutput(databaseName: "你的Cosmos数据库名", 
                    containerName: "你的Cosmos容器名", 
                    Connection = "CosmosDBConnection")]
    public BlobChangeEventRecord Run([EventGridTrigger] BlobEventGridEvent eventGridEvent)
    {
        _logger.LogInformation("收到Blob事件，类型：{EventType}", eventGridEvent.EventType);

        var blobData = eventGridEvent.Data.ToObjectFromJson<BlobStorageEventData>();

        return new BlobChangeEventRecord
        {
            // 用事件ID作为Cosmos文档ID，避免重复写入
            Id = eventGridEvent.Id,
            EventType = eventGridEvent.EventType,
            EventTime = eventGridEvent.EventTime,
            Subject = eventGridEvent.Subject,
            BlobDetail = blobData
        };
    }
}

// Event Grid事件通用结构
public class BlobEventGridEvent
{
    [JsonPropertyName("id")]
    public string Id { get; set; }

    [JsonPropertyName("subject")]
    public string Subject { get; set; }

    [JsonPropertyName("eventType")]
    public string EventType { get; set; }

    [JsonPropertyName("eventTime")]
    public DateTime EventTime { get; set; }

    [JsonPropertyName("data")]
    public BinaryData Data { get; set; }
}

步骤3：部署后事件订阅配置

Function部署完成后，需要到存储账户侧绑定事件触发规则：

进入目标存储账户，左侧菜单选择「事件」，点击「+ 事件订阅」
名称自定义，事件架构选择「事件网格架构」
事件类型勾选需要捕获的修改操作：Blob 删除、Blob 创建、Blob 重命名等
终结点类型选择「Azure Function」，选中你刚部署的BlobChangeCaptureFunction
如果需要只捕获特定容器的事件，可在「过滤器」标签页添加主题前缀匹配规则：/blobServices/default/containers/你的目标容器名
保存配置后即可生效

常见运行失败排障

Microsoft.EventGrid资源提供程序注册后需要等待3-5分钟才会生效
Cosmos DB连接字符串需使用读写密钥，不可用只读密钥
如果Function开启了身份验证，需要给事件网格授予Function的调用权限

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