Python 多线程读取 ES 数据并回填 phone 字段

以下是一个简单的 Python 多线程读取 ES 数据，并将数据存入队列中，然后从队列中获取数据并修改 ES 的示例代码：

from elasticsearch import Elasticsearch
from queue import Queue
import threading

# 创建 ES 客户端
es = Elasticsearch()

# 创建队列
queue = Queue()

# 读取 ES 数据的函数
def read_data_from_es():
    # 查询所有文档的 name 字段
    results = es.search(index='your_index', body={
        'query': {
            'match_all': {}
        },
        '_source': ['name']
    })
    
    # 将 name 字段存入队列
    for hit in results['hits']['hits']:
        name = hit['_source']['name']
        queue.put(name)

# 修改 ES 数据的函数
def update_data_in_es():
    while True:
        # 从队列中获取 name 字段
        name = queue.get()
        
        # 查询与该 name 对应的文档的 phone 字段
        results = es.search(index='your_index', body={
            'query': {
                'match': {
                    'name': name
                }
            },
            '_source': ['phone']
        })
        
        # 修改该文档的 phone 字段
        for hit in results['hits']['hits']:
            doc_id = hit['_id']
            phone = hit['_source']['phone']
            es.update(index='your_index', id=doc_id, body={
                'doc': {
                    'phone': phone + ' modified'
                }
            })
        
        # 标记该任务完成
        queue.task_done()

# 创建多线程并启动
read_thread = threading.Thread(target=read_data_from_es)
update_threads = [threading.Thread(target=update_data_in_es) for i in range(5)]
read_thread.start()
for thread in update_threads:
    thread.start()

# 等待队列中所有任务完成
queue.join()

这个示例代码中，首先创建了一个 ES 客户端和一个队列。然后创建了两个函数：read_data_from_es() 用于从 ES 中读取数据并将 name 字段存入队列，update_data_in_es() 用于从队列中获取 name 字段并查询对应的 phone 字段，并将 phone 字段修改后回填到 ES 中。最后创建了一个读线程和 5 个更新线程，并启动了它们。最后调用 queue.join() 等待队列中的所有任务完成。