使用Azure Data Factory为Cosmos DB JSON数据的嵌套数组新增字段

方案1：修改ADF数据流派生列配置（适配现有流水线逻辑） #

你遇到的结构异常是因为直接对Addresses数组的子字段做提取操作，没有保留数组的嵌套结构，只需调整派生列的表达式即可解决：

• 选中Addresses作为要修改的派生列，不要单独新建子字段
• 填入如下表达式：

map(Addresses, item, setProperties(item, {
  Line1LowerCase: toLower(item.Line1)
}))

• 表达式说明：
  • map函数用于遍历数组的每一个元素，处理后返回新数组，完全保留原有数组结构
  • setProperties函数用于给现有对象新增属性，不会覆盖原有字段
  • toLower函数将Line1的内容转为小写
    写入前确认Cosmos DB数据集的序列化设置中，嵌套结构保留为对象/数组，不要勾选「将嵌套数组展开为平级字段」的选项。

方案2：使用Cosmos DB存储过程批量处理（百万级数据更推荐） #

如果不需要跨库同步，直接在Cosmos DB侧执行批量修改效率更高，可避免数据进出的IO开销：

1. 新建Cosmos DB存储过程，代码示例如下：

function bulkUpdate() {
  var collection = getContext().getCollection();
  var collectionLink = collection.getSelfLink();
  var response = getContext().getResponse();
  var documentsUpdated = 0;

  tryQueryAndUpdate();

  function tryQueryAndUpdate(continuation) {
    var query = {query: "SELECT * FROM c"};
    var requestOptions = {continuation: continuation};

    var isAccepted = collection.queryDocuments(
      collectionLink,
      query,
      requestOptions,
      function (err, feed, responseOptions) {
        if (err) throw err;

        if (feed.length > 0) {
          asyncUpdateDocuments(feed, 0, function () {
            if (responseOptions.continuation) {
              tryQueryAndUpdate(responseOptions.continuation);
            } else {
              response.setBody("更新完成，共修改文档数：" + documentsUpdated);
            }
          });
        } else {
          response.setBody("没有需要更新的文档，共修改文档数：" + documentsUpdated);
        }
      });

    if (!isAccepted) {
      response.setBody("查询被限流，请重试，已修改文档数：" + documentsUpdated);
    }
  }

  function asyncUpdateDocuments(documents, index, callback) {
    if (index >= documents.length) {
      callback();
      return;
    }

    var doc = documents[index];
    if (doc.Addresses && Array.isArray(doc.Addresses)) {
      doc.Addresses.forEach(addr => {
        addr.Line1LowerCase = addr.Line1 ? addr.Line1.toLowerCase() : "";
      });
    }

    var isAccepted = collection.replaceDocument(doc._self, doc, {}, function (err) {
      if (err) throw err;
      documentsUpdated++;
      asyncUpdateDocuments(documents, index + 1, callback);
    });

    if (!isAccepted) {
      setTimeout(function () {
        asyncUpdateDocuments(documents, index, callback);
      }, 1000);
    }
  }
}

2. 执行存储过程即可，百万级数据可分批次执行，避免触碰到Cosmos DB的单次请求RU限制。

验证步骤 #

修改完成后随机抽取10-20条文档校验：

• Addresses字段依然为数组类型
• 每个数组元素包含Line1LowerCase字段，值为对应Line1的小写格式
• 原有字段无丢失、结构无异常

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