如何使用 Batch 更新 Firestore 中的 500 多个文档？

Posted 2023-02-22

【中文标题】如何使用 Batch 更新 Firestore 中的 500 多个文档？

【英文标题】：How can I update more than 500 docs in Firestore using Batch?

【发布时间】：2019-02-09 10:21:43

【问题描述】：

我正在尝试使用包含 500 多个文档的集合中的 Firestore 管理员时间戳更新字段 timestamp。

const batch = db.batch();
const serverTimestamp = admin.firestore.FieldValue.serverTimestamp();

db
  .collection('My Collection')
  .get()
  .then((docs) => 
    serverTimestamp,
  , 
    merge: true,
  )
  .then(() => res.send('All docs updated'))
  .catch(console.error);

这会引发错误

 Error: 3 INVALID_ARGUMENT: cannot write more than 500 entities in a single call
    at Object.exports.createStatusError (C:\Users\Growthfile\Desktop\cf-test\functions\node_modules\grpc\src\common.js:87:15)
    at Object.onReceiveStatus (C:\Users\Growthfile\Desktop\cf-test\functions\node_modules\grpc\src\client_interceptors.js:1188:28)
    at InterceptingListener._callNext (C:\Users\Growthfile\Desktop\cf-test\functions\node_modules\grpc\src\client_interceptors.js:564:42)
    at InterceptingListener.onReceiveStatus (C:\Users\Growthfile\Desktop\cf-test\functions\node_modules\grpc\src\client_interceptors.js:614:8)
    at callback (C:\Users\Growthfile\Desktop\cf-test\functions\node_modules\grpc\src\client_interceptors.js:841:24)
  code: 3,
  metadata: Metadata  _internal_repr:  ,
  details: 'cannot write more than 500 entities in a single call' 

有没有一种方法可以让我编写一个递归方法来创建一个批处理对象，一个接一个地更新一批 500 个文档，直到所有文档都被更新。

从文档中我知道使用此处提到的递归方法可以进行删除操作：

https://firebase.google.com/docs/firestore/manage-data/delete-data#collections

但是，为了更新，我不确定如何结束执行，因为文档没有被删除。

【问题讨论】：

为什么不遍历所有 500 个文档，更新并使用最后一个文档键来构造 startAt 以创建新查询？

您可以限制然后递归批处理，遇到同样的问题，这是我的解决方案：***.com/a/61639536/2195000

【参考方案1】：

我在更新 Firestore 集合中的 500 多个文档时也遇到了问题。我想分享一下我是如何解决这个问题的。

我使用云功能来更新我在 Firestore 中的收藏，但这也应该适用于客户端代码。

解决方案会计算对批次进行的每个操作，并在达到限制后创建一个新批次并将其推送到batchArray。

所有更新完成后，代码循环通过batchArray 并提交数组内的每个批次。

计算每个操作很重要 set(), update(), delete() 对批处理进行，因为它们都计入 500 次操作限制。

const documentSnapshotArray = await firestore.collection('my-collection').get();

const batchArray = [];
batchArray.push(firestore.batch());
let operationCounter = 0;
let batchIndex = 0;

documentSnapshotArray.forEach(documentSnapshot => 
    const documentData = documentSnapshot.data();

    // update document data here...

    batchArray[batchIndex].update(documentSnapshot.ref, documentData);
    operationCounter++;

    if (operationCounter === 499) 
      batchArray.push(firestore.batch());
      batchIndex++;
      operationCounter = 0;
    
);

batchArray.forEach(async batch => await batch.commit());

return;

【讨论】：

你如何确保所有批次都成功执行，因为只有批次中的操作是原子的。有的批次执行，有的不执行，会导致数据不一致

@Adarsh 是的，你是对的。我省略了错误处理部分。我会尽快将此部分添加到答案中。我已经将我的数据库更新为一个新的数据模型，在我的例子中这是一个幂等操作。所以我可以重复代码，直到每批都成功。

所以你可以做几件事。您可以在创建云功能时检查重试选项。这将确保您的云功能在任何异常情况下都能执行。但是您必须处理您认为是transient 的失败，否则它将变成一个无限循环。此外，必须在云函数执行之间维护某种状态，以便之前执行的批处理不会再次执行。也许您可以在每次成功的批处理操作时写入实时数据库/firestore，并在下次重试时某些批处理没有时从那里继续

或者您可以编写作业详细信息（更新详细信息），例如/queue/pendingUpdates/，并编写一个按计划（例如每 5 分钟）运行的云函数来执行更新。操作成功后，您可以将作业删除/标记为已完成。否则它会在下一个间隔自动重试。这比第一个容易得多。你的想法？

@Mihae Kheel 是的，循环在达到 500 次操作后创建一个新批次，但重要的是要计算每个操作。您还需要某种形式的错误处理。

【参考方案2】：

我喜欢这个简单的解决方案：

const users = await db.collection('users').get()

const batches = _.chunk(users.docs, 500).map(userDocs => 
    const batch = db.batch()
    userDocs.forEach(doc => 
        batch.set(doc.ref,  field: 'myNewValue' ,  merge: true )
    )
    return batch.commit()
)

await Promise.all(batches)

记得在顶部添加import * as _ from "lodash"。基于this answer。

【讨论】：

“使用打字稿” ...我没有看到任何打字稿

这应该是官方文档的一部分。或者至少类似的东西不依赖于 lodash。奇迹般有效！ :)

@MattFletcher loadash 在 Vanilla JS 中编写，如果你想要类型支持安装 @types/lodash

【参考方案3】：

如上所述，@Sebastian 的回答很好，我也对此表示赞同。尽管在一次更新 25000 多个文档时遇到了问题。 逻辑调整如下。

console.log(`Updating documents...`);
let collectionRef = db.collection('cities');
try 
  let batch = db.batch();
  const documentSnapshotArray = await collectionRef.get();
  const records = documentSnapshotArray.docs;
  const index = documentSnapshotArray.size;
  console.log(`TOTAL SIZE=====$index`);
  for (let i=0; i < index; i++) 
    const docRef = records[i].ref;
    // YOUR UPDATES
    batch.update(docRef, isDeleted: false);
    if ((i + 1) % 499 === 0) 
      await batch.commit();
      batch = db.batch();
    
  
  // For committing final batch
  if (!(index % 499) == 0) 
    await batch.commit();
  
  console.log('write completed');
 catch (error) 
  console.error(`updateWorkers() errored out : $error.stack`);
  reject(error);

【参考方案4】：

简单的解决方案 只开火两次？ 我的数组是“resultsFinal” 我一次以 490 的限制发射批次，第二次以阵列的长度限制（results.lenght） 对我来说很好:) 你是怎么检查的？ 你去firebase并删除你的收藏，firebase说你已经删除了XXX文档，和你的数组长度一样？好的，你可以走了

async function quickstart(results) 
    // we get results in parameter for get the data inside quickstart function
    const resultsFinal = results;
    // console.log(resultsFinal.length);
    let batch = firestore.batch();
    // limit of firebase is 500 requests per transaction/batch/send 
    for (i = 0; i < 490; i++) 
        const doc = firestore.collection('testMore490').doc();
        const object = resultsFinal[i];
        batch.set(doc, object);
    
    await batch.commit();
    // const batchTwo = firestore.batch();
    batch = firestore.batch();

    for (i = 491; i < 776; i++) 
        const objectPartTwo = resultsFinal[i];
        const doc = firestore.collection('testMore490').doc();
        batch.set(doc, objectPartTwo);
    
    await batch.commit();

【参考方案5】：

以前的 cmets 给出的解释已经解释了这个问题。

我正在分享我为我构建和工作的最终代码，因为我需要以更解耦的方式工作的东西，而不是上面介绍的大多数解决方案的工作方式。

import  FireDb  from "@services/firebase"; // = firebase.firestore();

type TDocRef = FirebaseFirestore.DocumentReference;
type TDocData = FirebaseFirestore.DocumentData;

let fireBatches = [FireDb.batch()];
let batchSizes = [0];
let batchIdxToUse = 0;

export default class FirebaseUtil 
  static addBatchOperation(
    operation: "create",
    ref: TDocRef,
    data: TDocData
  ): void;
  static addBatchOperation(
    operation: "update",
    ref: TDocRef,
    data: TDocData,
    precondition?: FirebaseFirestore.Precondition
  ): void;
  static addBatchOperation(
    operation: "set",
    ref: TDocRef,
    data: TDocData,
    setOpts?: FirebaseFirestore.SetOptions
  ): void;
  static addBatchOperation(
    operation: "create" | "update" | "set",
    ref: TDocRef,
    data: TDocData,
    opts?: FirebaseFirestore.Precondition | FirebaseFirestore.SetOptions
  ): void 
    // Lines below make sure we stay below the limit of 500 writes per
    // batch
    if (batchSizes[batchIdxToUse] === 500) 
      fireBatches.push(FireDb.batch());
      batchSizes.push(0);
      batchIdxToUse++;
    
    batchSizes[batchIdxToUse]++;

    const batchArgs: [TDocRef, TDocData] = [ref, data];
    if (opts) batchArgs.push(opts);

    switch (operation) 
      // Specific case for "set" is required because of some weird TS
      // glitch that doesn't allow me to use the arg "operation" to
      // call the function
      case "set":
        fireBatches[batchIdxToUse].set(...batchArgs);
        break;
      default:
        fireBatches[batchIdxToUse][operation](...batchArgs);
        break;
    
  

  public static async runBatchOperations() 
    // The lines below clear the globally available batches so we
    // don't run them twice if we call this function more than once
    const currentBatches = [...fireBatches];
    fireBatches = [FireDb.batch()];
    batchSizes = [0];
    batchIdxToUse = 0;

    await Promise.all(currentBatches.map((batch) => batch.commit()));
  

【参考方案6】：

您可以使用默认的BulkWriter。此方法使用 500/50/5 规则。

例子：

let bulkWriter = firestore.bulkWriter();

bulkWriter.create(documentRef, foo: 'bar');
bulkWriter.update(documentRef2, foo: 'bar');
bulkWriter.delete(documentRef3);
await close().then(() => 
  console.log('Executed all writes');
);