Python场景下如何高效同步更新MongoDB与ElasticSearch？

• 优先做批量消费+批量写入，当前代码单条消费、单条写入的模式会产生大量重复的网络IO请求，是开销高的核心原因。可调整Kafka消费者配置，设置max_poll_records参数控制单次拉取的消息数量，攒够指定批量大小后，再调用MongoDB的bulk_write接口、ES的bulk接口做批量写入，把多次网络请求合并为一次，大幅降低开销。参考修改逻辑如下：

from pymongo import UpdateOne
from elasticsearch import helpers

# 可根据实时性要求调整批量大小，实时性要求低可设为100~1000，要求高可设为10~50
BATCH_SIZE = 100
batch = []

if __name__ == "__main__":
    consumer = utils.kafka_conn()
    consumer.subscribe(('topic_name',))
    # 初始化全局Mongo、ES客户端，全程复用不要每次请求新建
    mongo_col = utils.get_mongo_collection()
    es_client = utils.get_es_client()
    for message in consumer:
        field1 = handling_field1()
        field2 = handling_field2()
        batch.append((field1, field2, message.offset))
        if len(batch) >= BATCH_SIZE:
            # 批量写入MongoDB
            mongo_ops = [UpdateOne({'_id': your_record_id}, {'$set': {"field1": f1, "field2": f2}}) for f1, f2, _ in batch]
            mongo_col.bulk_write(mongo_ops)
            # 批量写入ES
            es_actions = [{"_index": "your_index", "_id": your_record_id, "_source": {"field1": f1, "field2": f2}} for f1, f2, _ in batch]
            helpers.bulk(es_client, es_actions)
            # 写入成功后提交Kafka offset，避免数据丢失
            consumer.commit()
            batch.clear()

• 复用客户端连接，不要在update_mongodb、update_es函数中每次调用都新建连接，连接创建和销毁的开销极高，将MongoDB客户端、ES客户端设为全局单例，程序启动时初始化一次，全程复用即可。
• 异步化并行写入，当前同步逻辑会等MongoDB写入完成后再执行ES写入，两个写入操作是串行的，可改用异步客户端实现并行写入：Kafka用aiokafka、MongoDB用motor、ES用AsyncElasticsearch，通过asyncio.gather同时发起两个写入请求，减少等待时间。参考逻辑如下：

import asyncio
from aiokafka import AIOKafkaConsumer
from motor.motor_asyncio import AsyncIOMotorClient
from elasticsearch import AsyncElasticsearch, helpers

async def main():
    consumer = AIOKafkaConsumer('topic_name', bootstrap_servers='your_kafka_addr', enable_auto_commit=False)
    await consumer.start()
    mongo_client = AsyncIOMotorClient('your_mongo_uri')
    es_client = AsyncElasticsearch('your_es_addr')
    mongo_col = mongo_client['db_name']['col_name']
    BATCH_SIZE = 100
    batch = []
    try:
        async for msg in consumer:
            field1 = handling_field1()
            field2 = handling_field2()
            batch.append((field1, field2))
            if len(batch) >= BATCH_SIZE:
                mongo_ops = [UpdateOne({'_id': your_id}, {'$set': {"field1": f1, "field2": f2}}) for f1, f2 in batch]
                es_actions = [{"_index": "your_index", "_id": your_id, "_source": {"field1": f1, "field2": f2}} for f1, f2 in batch]
                # 并行执行两个写入请求
                await asyncio.gather(
                    mongo_col.bulk_write(mongo_ops),
                    helpers.async_bulk(es_client, es_actions)
                )
                await consumer.commit()
                batch.clear()
    finally:
        await consumer.stop()
        await es_client.close()
        mongo_client.close()

if __name__ == "__main__":
    asyncio.run(main())

• 参数调优：调整Kafka消费者的fetch_min_bytes、fetch_max_wait_ms参数，避免频繁拉取过小的消息批次；调整MongoDB、ES客户端的连接池大小，匹配写入并发量，避免连接不足导致的等待开销。

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