如何在Azure Function中增改CosmosDB数据并同步汽车经销商库存

我来一步步帮你解决这两个Azure Function结合CosmosDB的常见问题，都是实际项目中经常遇到的场景：

首先，你需要确保Function项目中引用了Microsoft.Azure.Cosmos NuGet包，并且通过依赖注入获取CosmosClient实例（这是目前推荐的最佳实践，避免频繁创建客户端连接）。

下面用C#示例展示三种常用操作：

核心操作说明 #

• Upsert（插入/更新自动判断）：最省心的方式，如果文档不存在就插入，存在则直接覆盖更新，适合大多数场景。
• Create（仅插入）：严格只做插入操作，如果文档已存在会抛出冲突异常，适合需要确保数据全新的场景。
• Replace（仅更新）：必须基于已存在的文档进行更新，需要指定文档ID和分区键，不存在则抛出未找到异常。

代码示例（Http触发Function） #

using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Azure.Cosmos;
using System.Net;
using System.Text.Json;

public class CosmosDbCarOperations
{
    private readonly Container _carContainer;

    // 通过依赖注入获取CosmosClient，初始化容器
    public CosmosDbCarOperations(CosmosClient cosmosClient)
    {
        _carContainer = cosmosClient.GetContainer("CarInventoryDB", "Cars");
    }

    /// <summary>
    /// Upsert车辆数据：自动处理插入或更新
    /// </summary>
    [Function("UpsertCar")]
    public async Task<HttpResponseData> UpsertCar([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req)
    {
        var car = await JsonSerializer.DeserializeAsync<Car>(req.Body);
        if (car == null)
        {
            return req.CreateResponse(HttpStatusCode.BadRequest, "无效的车辆数据");
        }

        // 假设DealerId是分区键，VIN是车辆唯一标识，这里把VIN设为CosmosDB的Id确保唯一性
        car.Id = car.VIN;
        var response = await _carContainer.UpsertItemAsync(car, new PartitionKey(car.DealerId));

        var result = req.CreateResponse(HttpStatusCode.OK);
        await JsonSerializer.SerializeAsync(result.Body, response.Resource);
        return result;
    }

    /// <summary>
    /// 仅插入新车辆，已存在则报错
    /// </summary>
    [Function("InsertNewCar")]
    public async Task<HttpResponseData> InsertNewCar([HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req)
    {
        var car = await JsonSerializer.DeserializeAsync<Car>(req.Body);
        if (car == null)
        {
            return req.CreateResponse(HttpStatusCode.BadRequest, "无效的车辆数据");
        }

        car.Id = car.VIN;
        try
        {
            var response = await _carContainer.CreateItemAsync(car, new PartitionKey(car.DealerId));
            var result = req.CreateResponse(HttpStatusCode.Created);
            await JsonSerializer.SerializeAsync(result.Body, response.Resource);
            return result;
        }
        catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.Conflict)
        {
            return req.CreateResponse(HttpStatusCode.Conflict, $"VIN为{car.VIN}的车辆已存在");
        }
    }

    /// <summary>
    /// 更新已存在的车辆数据
    /// </summary>
    [Function("UpdateExistingCar")]
    public async Task<HttpResponseData> UpdateExistingCar([HttpTrigger(AuthorizationLevel.Function, "put")] HttpRequestData req, string vin)
    {
        var updatedCar = await JsonSerializer.DeserializeAsync<Car>(req.Body);
        if (updatedCar == null || string.IsNullOrEmpty(vin))
        {
            return req.CreateResponse(HttpStatusCode.BadRequest, "无效的请求参数");
        }

        try
        {
            // 必须指定文档Id（这里是vin）和分区键
            var response = await _carContainer.ReplaceItemAsync(updatedCar, vin, new PartitionKey(updatedCar.DealerId));
            var result = req.CreateResponse(HttpStatusCode.OK);
            await JsonSerializer.SerializeAsync(result.Body, response.Resource);
            return result;
        }
        catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
        {
            return req.CreateResponse(HttpStatusCode.NotFound, $"未找到VIN为{vin}的车辆");
        }
    }
}

// 车辆实体类
public class Car
{
    public string Id { get; set; } // CosmosDB默认文档ID
    public string VIN { get; set; } // 车辆唯一标识（核心唯一键）
    public string DealerId { get; set; } // 分区键（建议按经销商划分）
    public int StockQuantity { get; set; } // 库存数量
    public decimal Price { get; set; } // 价格
    // 其他业务字段...
}

你的需求核心是确保CosmosDB中每个车辆（按唯一标识）只保留最新的库存、价格信息，最佳方案是结合Blob触发的Azure Function + CosmosDB批量Upsert操作，具体步骤如下：

实现思路 #

1. 触发机制：用Blob存储触发Function，当每日新的库存文件上传到指定Blob容器时，自动触发同步逻辑；也可以用Timer触发，每天固定时间拉取文件。
2. 文件解析：读取上传的文件（比如JSON、CSV格式），解析为Car对象列表。
3. 批量Upsert：对每个Car执行Upsert操作，利用车辆唯一标识（比如VIN）作为CosmosDB的文档ID，或者给容器设置唯一键约束，确保旧数据被新数据覆盖。
4. 性能优化：使用CosmosDB的批量操作，减少网络请求次数，提升同步效率。

代码示例（Blob触发Function） #

using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Blob;
using Microsoft.Azure.Cosmos;
using System.IO;
using System.Text.Json;

public class CarInventorySync
{
    private readonly Container _carContainer;

    public CarInventorySync(CosmosClient cosmosClient)
    {
        _carContainer = cosmosClient.GetContainer("CarInventoryDB", "Cars");
    }

    /// <summary>
    /// 当库存文件上传到Blob容器时触发同步
    /// </summary>
    [Function("SyncCarInventoryFromBlob")]
    public async Task Run([BlobTrigger("daily-inventory/{fileName}", Connection = "AzureWebJobsStorage")] Stream fileStream, string fileName)
    {
        // 1. 读取并解析文件
        string fileContent;
        using (var reader = new StreamReader(fileStream))
        {
            fileContent = await reader.ReadToEndAsync();
        }

        var carList = JsonSerializer.Deserialize<List<Car>>(fileContent);
        if (carList == null || !carList.Any())
        {
            // 日志记录：文件为空或解析失败
            return;
        }

        // 2. 按分区键分组，批量Upsert（同一分区的文档才能放在一个批量里）
        var groupedCars = carList.GroupBy(c => c.DealerId);
        foreach (var group in groupedCars)
        {
            var batch = _carContainer.CreateBatch(new PartitionKey(group.Key));
            foreach (var car in group)
            {
                // 用VIN作为文档ID，确保唯一性
                car.Id = car.VIN;
                batch.UpsertItemOperation(car);

                // 批量操作最多支持100条，达到上限就执行
                if (batch.Count >= 100)
                {
                    await ExecuteBatchWithRetry(batch);
                    batch = _carContainer.CreateBatch(new PartitionKey(group.Key));
                }
            }

            // 执行剩余的批量操作
            if (batch.Count > 0)
            {
                await ExecuteBatchWithRetry(batch);
            }
        }
    }

    /// <summary>
    /// 带重试的批量操作执行，处理临时异常
    /// </summary>
    private async Task ExecuteBatchWithRetry(Batch batch)
    {
        try
        {
            await _carContainer.ExecuteBatchAsync(batch);
        }
        catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.TooManyRequests)
        {
            // 遇到限流，等待后重试（简单示例，实际可使用更智能的重试策略）
            await Task.Delay(TimeSpan.FromSeconds(ex.RetryAfter.Value.TotalSeconds));
            await _carContainer.ExecuteBatchAsync(batch);
        }
    }
}

额外注意事项 #

• 唯一键约束：如果不想把VIN作为文档ID，可以在CosmosDB容器的设置中添加唯一键（比如/VIN），这样即使ID不同，只要VIN重复，Upsert也会自动更新旧文档。
• 分区键选择：选择经销商ID作为分区键是合理的，因为同一经销商的车辆查询和同步都会更高效，避免跨分区操作的性能损耗。
• 错误处理：实际项目中建议添加日志记录（比如用ILogger），记录同步成功/失败的车辆数量，以及失败的具体原因，方便排查问题。
• 文件格式：如果是CSV文件，可以用CsvHelper等库进行解析，替换代码中的JSON序列化部分即可。

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