从 Firestore 中删除集合的最快方法？

Question

我有一个应用程序将数百万个文档加载到一个集合中，使用 30-80 个工作人员同时加载数据。有时，我发现加载过程没有顺利完成，对于其他数据库，我可以简单地删除表并重新开始，但对于 Firestore 集合则不行。我必须列出文档并删除它们，但我还没有找到一种方法来扩展它，使其具有与加载过程相同的容量。我现在正在做的是我有两个 AppEngine 托管的 Flask/Python 方法，一个获取 1000 个文档的页面并传递给另一个方法来删除它们。这样，列出文档的过程就不会被删除它们的过程阻止。仍然需要几天才能完成，这太长了。

获取文档列表并创建删除它们的任务的方法，这是单线程的：

@app.route('/delete_collection/<collection_name>/<batch_size>', methods=['POST'])
def delete_collection(collection_name, batch_size):
    batch_size = int(batch_size)
    coll_ref = db.collection(collection_name)
    print('Received request to delete collection {} {} docs at a time'.format(
        collection_name,
        batch_size
    ))
    num_docs = batch_size
    while num_docs >= batch_size:
        docs = coll_ref.limit(batch_size).stream()
        found = 0
        deletion_request = {
            'doc_ids': []
        }
        for doc in docs:
            deletion_request['doc_ids'].append(doc.id)
            found += 1
        num_docs = found
        print('Creating request to delete docs: {}'.format(
            json.dumps(deletion_request)
        ))
        # Add to task queue
        queue = tasks_client.queue_path(PROJECT_ID, LOCATION, 'database-manager')

        task_meet = {
            'app_engine_http_request': {  # Specify the type of request.
                'http_method': 'POST',
                'relative_uri': '/delete_documents/{}'.format(
                    collection_name
                ),
                'body': json.dumps(deletion_request).encode(),
                'headers': {
                    'Content-Type': 'application/json'
                }
            }
        }
        task_response_meet = tasks_client.create_task(queue, task_meet)
        print('Created task to delete {} docs: {}'.format(
            batch_size,
            json.dumps(deletion_request)
        ))

这是我用来删除文档的方法，可以缩放。实际上，它一次只处理 5-10 个，受其他方法传递 doc_ids 页面以删除的速率的限制。将两者分开会有所帮助，但作用不大。

@app.route('/delete_documents/<collection_name>', methods=['POST'])
def delete_documents(collection_name):
    # Validate we got a body in the POST
    if flask.request.json:
        print('Request received to delete docs from :{}'.format(collection_name))
    else:
        message = 'No json found in request: {}'.format(flask.request)
        print(message)
        return message, 400

    # Validate that the payload includes a list of doc_ids
    doc_ids = flask.request.json.get('doc_ids', None)
    if doc_ids is None:
        return 'No doc_ids specified in payload: {}'.format(flask.request.json), 400
    print('Received request to delete docs: {}'.format(doc_ids))
    for doc_id in doc_ids:
        db.collection(collection_name).document(doc_id).delete()
    return 'Finished'


if __name__ == '__main__':
    # Set environment variables for running locally
    app.run(host='127.0.0.1', port=8080, debug=True)

我已经尝试运行多个并发执行 delete_collection()，但不确定是否有帮助，因为我不确定每次调用 limit(batch_size).stream() 时是否会获得一组不同的文档或可能正在重复。

我怎样才能让它运行得更快？

Answer 1

在此 public documentation 中描述了如何使用可调用的云函数，您可以利用 Firebase Command Line Interface 中的 firestore delete 命令每秒删除多达 4000 个文档。

Answer 2

这是我用来测试批量删除的简单 Python 脚本。就像@Chris32 所说的那样，如果延迟不太严重，批处理模式将每秒删除数千个文档。

from time import time
from uuid import uuid4
from google.cloud import firestore

DB = firestore.Client()

def generate_user_data(entries = 10):
    print('Creating {} documents'.format(entries))
    now = time()
    batch = DB.batch()
    for counter in range(entries):
        # Each transaction or batch of writes can write to a maximum of 500 documents.
        # https://cloud.google.com/firestore/quotas#writes_and_transactions
        if counter % 500 == 0 and counter > 0:
            batch.commit()

        user_id = str(uuid4())
        data = {
            "some_data": str(uuid4()),
            "expires_at": int(now)
            }
        user_ref = DB.collection(u'users').document(user_id)
        batch.set(user_ref, data)
    batch.commit()
    print('Wrote {} documents in {:.2f} seconds.'.format(entries, time() - now))

def delete_one_by_one():
    print('Deleting documents one by one')
    now = time()
    docs = DB.collection(u'users').where(u'expires_at', u'<=', int(now)).stream()
    counter = 0
    for doc in docs:
        doc.reference.delete()
        counter = counter + 1
    print('Deleted {} documents in {:.2f} seconds.'.format(counter, time() - now))

def delete_in_batch():
    print('Deleting documents in batch')
    now = time()
    docs = DB.collection(u'users').where(u'expires_at', u'<=', int(now)).stream()
    batch = DB.batch()
    counter = 0
    for doc in docs:
        counter = counter + 1
        if counter % 500 == 0:
            batch.commit()
        batch.delete(doc.reference)
    batch.commit()
    print('Deleted {} documents in {:.2f} seconds.'.format(counter, time() - now))


generate_user_data(10)
delete_one_by_one()
print('###')
generate_user_data(10)
delete_in_batch()
print('###')
generate_user_data(2000)
delete_in_batch()

Answer 3

这就是我想出的。它不是超级快（每秒 120-150 个文档），但我在 python 中找到的所有其他示例都不起作用：

now = datetime.now()
then = now - timedelta(days=DOCUMENT_EXPIRATION_DAYS)
doc_counter = 0
commit_counter = 0
limit = 5000
while True:
    docs = []
    print('Getting next doc handler')
    docs = [snapshot for snapshot in db.collection(collection_name)
        .where('id.time', '<=', then)
        .limit(limit)
        .order_by('id.time', direction=firestore.Query.ASCENDING
      ).stream()]
    batch = db.batch()
    for doc in docs:
        doc_counter = doc_counter + 1
        if doc_counter % 500 == 0:
            commit_counter += 1
            print('Committing batch {} from {}'.format(commit_counter, doc.to_dict()['id']['time']))
            batch.commit()
        batch.delete(doc.reference)
    batch.commit()
    if len(docs) == limit:
        continue
    break

print('Deleted {} documents in {} seconds.'.format(doc_counter, datetime.now() - now))

正如在其他 cmets 中提到的，.stream() 有 60 秒的最后期限。这个迭代结构设置了 5000 的限制，之后再次调用 .stream()，使其保持在 60 秒的限制之下。如果有人知道如何加快速度，请告诉我。