Cosmos DB容器反序列化失败时如何跳过问题文档而不丢弃整批数据

你之前编写的自定义CosmosSerializer无法实现跳过整个问题文档的原因是：Newtonsoft的Error事件处理是字段级别的，设置args.ErrorContext.Handled = true仅会忽略当前字段的反序列化错误，继续完成整个文档的剩余反序列化逻辑，最终得到字段为默认值的无效对象，不会终止当前文档的反序列化流程。

要实现单文档级别的错误隔离，你需要将反序列化粒度降低到单个文档维度，手动控制每个文档的反序列化逻辑和异常处理，有两种常用实现方案：

方案一：使用JObject作为中间类型（实现简单，适合中小数据量场景） #

先将查询结果读取为JObject，再逐个尝试反序列化为目标类型，捕获单文档反序列化异常后直接跳过：

var result = new List<StoredLead>();
var linqSerializerOptions = new CosmosLinqSerializerOptions
{
    PropertyNamingPolicy = CosmosPropertyNamingPolicy.CamelCase
};
// 泛型改为JObject，先读取为通用JSON对象
var iterator = container.GetItemLinqQueryable<JObject>(false, null, null, linqSerializerOptions)
    .Where(expression)
    .ToFeedIterator();

// 复用和Cosmos规则一致的序列化配置
var serializer = new JsonSerializer
{
    NullValueHandling = NullValueHandling.Ignore,
    ContractResolver = new CamelCasePropertyNamesContractResolver()
};

while (iterator.HasMoreResults)
{
    var batch = await iterator.ReadNextAsync();
    foreach (var jObj in batch)
    {
        StoredLead lead = null;
        try
        {
            // 单独反序列化当前文档
            lead = jObj.ToObject<StoredLead>(serializer);
        }
        catch (JsonException ex)
        {
            // 记录错误后跳过当前文档
            _logger.LogError(ex, "文档反序列化失败，已跳过，文档ID：{docId}", jObj["id"]?.ToString());
            continue;
        }
        // 仅反序列化成功的文档进入后续处理
        // Do some stuff with the lead
        result.Add(lead);
    }
}

return result;

方案二：流式读取原始响应（性能更高，适合大批量数据读取场景） #

直接读取Cosmos返回的原始响应流，逐元素解析，错误时跳过当前文档的剩余内容，内存开销更低：

var result = new List<StoredLead>();
var linqSerializerOptions = new CosmosLinqSerializerOptions
{
    PropertyNamingPolicy = CosmosPropertyNamingPolicy.CamelCase
};
// 将Linq表达式转换为查询定义
var queryDefinition = container.GetItemLinqQueryable<StoredLead>(linqSerializerOptions: linqSerializerOptions)
    .Where(expression)
    .ToQueryDefinition();
// 使用流式迭代器获取原始响应流
var streamIterator = container.GetItemQueryStreamIterator(queryDefinition);

var serializer = new JsonSerializer
{
    NullValueHandling = NullValueHandling.Ignore,
    ContractResolver = new CamelCasePropertyNamesContractResolver()
};

while (streamIterator.HasMoreResults)
{
    using var response = await streamIterator.ReadNextAsync();
    if (!response.IsSuccessStatusCode)
    {
        // 处理请求级别的错误，如限流、服务异常等
        continue;
    }
    // 解析响应流
    using var streamReader = new StreamReader(response.Content);
    using var jsonReader = new JsonTextReader(streamReader);
    // 定位到结果数组节点
    while (await jsonReader.ReadAsync())
    {
        if (jsonReader.TokenType == JsonToken.PropertyName && jsonReader.Value.ToString() == "Documents")
        {
            await jsonReader.ReadAsync();
            // 遍历结果数组中的每个文档
            while (await jsonReader.ReadAsync() && jsonReader.TokenType != JsonToken.EndArray)
            {
                try
                {
                    var lead = serializer.Deserialize<StoredLead>(jsonReader);
                    // Do some stuff with the lead
                    result.Add(lead);
                }
                catch (JsonException ex)
                {
                    _logger.LogError(ex, "单文档反序列化失败，已跳过");
                    // 跳过当前错误文档的所有剩余内容
                    while (jsonReader.TokenType != JsonToken.EndObject)
                    {
                        await jsonReader.ReadAsync();
                    }
                }
            }
            break;
        }
    }
}

return result;

两种方案均可实现单个文档反序列化失败不影响同批次其他文档的效果，你可以根据自己的业务数据量选择对应的实现。如果有自定义序列化规则，直接替换代码中的serializer实例为你自己封装的配置即可。

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