如何为Cosmos DB所有文档添加新属性？

完全不用删除重建现有文档！批量给Cosmos DB集合里的所有文档添加新属性，其实可以通过批量读取+更新的方式轻松实现，我给你整理了C#和JavaScript两种实战验证过的方案，直接就能用：

C# 实现方案 #

前置准备 #

先确保你已经安装了Microsoft.Azure.Cosmos NuGet包，这是官方的Cosmos DB SDK。

完整代码示例 #

using Microsoft.Azure.Cosmos;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

class CosmosDbBulkUpdate
{
    // 替换为你自己的Cosmos DB配置
    private static readonly string EndpointUri = "你的Cosmos DB端点地址";
    private static readonly string PrimaryKey = "你的Cosmos DB主键";
    private static readonly string DatabaseId = "目标数据库ID";
    private static readonly string ContainerId = "目标集合ID";
    private static readonly string PartitionKeyField = "你的分区键字段名"; // 比如"userId"

    static async Task Main(string[] args)
    {
        // 初始化Cosmos客户端
        using var cosmosClient = new CosmosClient(EndpointUri, PrimaryKey);
        var container = cosmosClient.GetDatabase(DatabaseId).GetContainer(ContainerId);

        string continuationToken = null;
        do
        {
            // 分页读取集合内的所有文档
            var query = new QueryDefinition("SELECT * FROM c");
            var resultIterator = container.GetItemQueryIterator<dynamic>(query, continuationToken);

            while (resultIterator.HasMoreResults)
            {
                var response = await resultIterator.ReadNextAsync();
                continuationToken = response.ContinuationToken;

                // 批量执行更新操作
                var updateTasks = new List<Task>();
                foreach (var doc in response)
                {
                    // 给文档添加新属性，这里示例添加"IsArchived"，默认值为false
                    doc.IsArchived = false;
                    // 执行替换更新（必须保留id和分区键）
                    updateTasks.Add(
                        container.ReplaceItemAsync(
                            doc, 
                            doc.id, 
                            new PartitionKey(doc[PartitionKeyField])
                        )
                    );
                }

                await Task.WhenAll(updateTasks);
                Console.WriteLine($"已完成 {updateTasks.Count} 个文档的更新");
            }
        } while (!string.IsNullOrEmpty(continuationToken));

        Console.WriteLine("所有文档更新完成！");
    }
}

注意事项 #

• 替换代码中的配置参数为你自己的实际信息，尤其是分区键字段名
• 如果集合文档量极大，建议控制单次批量更新的数量，避免触发Cosmos DB的限流（429错误），SDK默认会自动重试，也可以自定义重试策略
• 若不需要全量更新，可以修改查询语句过滤目标文档，比如SELECT * FROM c WHERE c.IsArchived = null

JavaScript 实现方案（Node.js） #

前置准备 #

先安装官方SDK：npm install @azure/cosmos

完整代码示例 #

const { CosmosClient } = require("@azure/cosmos");

// 替换为你自己的Cosmos DB配置
const config = {
    endpoint: "你的Cosmos DB端点地址",
    key: "你的Cosmos DB主键",
    databaseId: "目标数据库ID",
    containerId: "目标集合ID",
    partitionKeyField: "你的分区键字段名"
};

async function bulkAddProperty() {
    const client = new CosmosClient(config);
    const container = client.database(config.databaseId).container(config.containerId);

    let continuationToken = undefined;
    do {
        // 分页读取文档
        const { resources: documents, continuationToken: nextToken } = await container.items
            .query("SELECT * FROM c", { continuationToken })
            .fetchAll();

        continuationToken = nextToken;

        // 批量更新文档
        const updatePromises = documents.map(async (doc) => {
            // 添加新属性，示例添加"LastUpdated"为当前时间
            doc.LastUpdated = new Date().toISOString();
            return container.item(doc.id, doc[config.partitionKeyField]).replace(doc);
        });

        await Promise.all(updatePromises);
        console.log(`已完成 ${updatePromises.length} 个文档的更新`);
    } while (continuationToken);

    console.log("所有文档更新完成！");
}

// 执行更新
bulkAddProperty().catch(err => {
    console.error("更新过程出错：", err);
});

进阶优化：使用存储过程（适合超大数据量） #

如果你的集合文档量特别大，推荐使用Cosmos DB的存储过程在服务器端执行更新，减少网络往返开销。以下是存储过程的示例代码：

// 存储过程代码，需要上传到你的Cosmos DB集合中
function bulkAddNewProperty() {
    const collection = getContext().getCollection();
    const response = getContext().getResponse();
    let continuationToken = null;

    // 定义要添加的属性和值
    const newProperty = { "IsProcessed": true };

    function queryNextBatch() {
        const queryOptions = { continuationToken: continuationToken };
        const isQueryAccepted = collection.queryDocuments(
            collection.getSelfLink(),
            "SELECT * FROM c",
            queryOptions,
            (err, docs, options) => {
                if (err) throw err;
                if (docs.length === 0) {
                    response.setBody("全量更新完成");
                    return;
                }

                // 更新当前批次的文档
                let updateCount = 0;
                docs.forEach(doc => {
                    // 合并新属性
                    Object.assign(doc, newProperty);
                    const isUpdateAccepted = collection.replaceDocument(
                        doc._self,
                        doc,
                        (err) => {
                            if (err) throw err;
                            updateCount++;
                            if (updateCount === docs.length) {
                                continuationToken = options.continuationToken;
                                continuationToken ? queryNextBatch() : response.setBody("全量更新完成");
                            }
                        }
                    );
                    if (!isUpdateAccepted) throw new Error("更新请求被拒绝");
                });
            }
        );
        if (!isQueryAccepted) throw new Error("查询请求被拒绝");
    }

    queryNextBatch();
}

内容的提问来源于stack exchange，提问作者John von No Man