如何并行化for循环以提升Elasticsearch双索引查询人脸比对效率

我正在开发一个涉及多次从Elasticsearch查询提取数据的实时业务项目，需要同时查询Elasticsearch的两个不同索引，对处理耗时要求极高。现有代码包含两个独立for循环分别处理两个索引的返回结果，当前循环耗时如下：

现有核心代码：

t2 = time.time()

for i in range(docs_enter):
    enter_t = req_enter['hits']['hits'][i]['_source']['Time']
    enter_time = datetime.strptime(enter_t,time_format)
    total_duration = now - enter_time
    if total_duration < threshold_hours:
        enter_face = req_enter['hits']['hits'][i]['_source']['FaceID']
        enter_img = np.array(enter_face).astype(np.uint8)
        match_enter = verifyFace(crop, enter_img)
        acc_enter.append(match_enter)

t3 = time.time()

for j in range(docs_guichet):
    guichet_t = req_guichet['hits']['hits'][j]['_source']['Time']
    guichet_time = datetime.strptime(guichet_t,time_format)
    total_duration = now - guichet_time
    if total_duration < threshold_hours:
        guichet_face = req_guichet['hits']['hits'][j]['_source']['FaceID']
        guichet_img = np.array(guichet_face).astype(np.uint8)
        match_guichet = verifyFace(crop, guichet_img)
        acc_guichet.append(match_guichet)

t4 = time.time()

print(f"\n------ 1st loop time {t3 - t2} -------\n")
print(f"\n------ 2nd loop time {t4 - t3} -------\n")

需要实现两个目标：优化单for循环运行耗时、实现两个for循环并行执行。

1. 单循环性能优化 #

• 提前下推过滤逻辑到Elasticsearch：在ES查询语句中新增range时间过滤条件，直接只返回符合threshold_hours阈值的文档，大幅减少循环迭代次数，这是优先级最高的优化手段。
• 减少重复字典查找：提前将req_enter['hits']['hits']、req_guichet['hits']['hits']赋值给局部变量，避免循环内重复查询字典路径，同时直接迭代hit对象无需下标访问，进一步提升效率。
• 优化时间解析速度：若使用Python3.7+且时间格式符合ISO标准，将datetime.strptime替换为性能高3~5倍的datetime.fromisoformat。
• 批量调用人脸验证接口：若verifyFace支持批量输入，可先收集所有符合条件的人脸数据后一次性调用，减少单次函数调用的 overhead。

优化后的单循环示例：

# 提前提取hit列表，减少重复查找
enter_hits = req_enter['hits']['hits']
for hit in enter_hits:
    enter_t = hit['_source']['Time']
    # 替换为更快的时间解析方法
    enter_time = datetime.fromisoformat(enter_t)
    if (now - enter_time) < threshold_hours:
        enter_face = hit['_source']['FaceID']
        enter_img = np.array(enter_face, dtype=np.uint8)
        acc_enter.append(verifyFace(crop, enter_img))

2. 双循环并行实现 #

两个循环无数据依赖，可通过concurrent.futures实现并行，根据verifyFace的类型选择执行器：

• 若verifyFace为CPU密集型运算：使用ProcessPoolExecutor规避GIL限制
• 若verifyFace为IO密集型调用：使用ThreadPoolExecutor降低进程切换开销

并行实现代码示例：

from concurrent.futures import ThreadPoolExecutor, as_completed

# 封装两个循环为独立函数
def process_enter_hits():
    acc_enter = []
    enter_hits = req_enter['hits']['hits']
    for hit in enter_hits:
        enter_t = hit['_source']['Time']
        enter_time = datetime.fromisoformat(enter_t)
        if (now - enter_time) < threshold_hours:
            enter_face = hit['_source']['FaceID']
            enter_img = np.array(enter_face, dtype=np.uint8)
            acc_enter.append(verifyFace(crop, enter_img))
    return acc_enter

def process_guichet_hits():
    acc_guichet = []
    guichet_hits = req_guichet['hits']['hits']
    for hit in guichet_hits:
        guichet_t = hit['_source']['Time']
        guichet_time = datetime.fromisoformat(guichet_t)
        if (now - guichet_time) < threshold_hours:
            guichet_face = hit['_source']['FaceID']
            guichet_img = np.array(guichet_face, dtype=np.uint8)
            acc_guichet.append(verifyFace(crop, guichet_img))
    return acc_guichet

# 并行执行两个函数
t2 = time.time()
with ThreadPoolExecutor(max_workers=2) as executor:
    futures = {
        executor.submit(process_enter_hits): 'enter',
        executor.submit(process_guichet_hits): 'guichet'
    }
    for future in as_completed(futures):
        if futures[future] == 'enter':
            acc_enter = future.result()
        else:
            acc_guichet = future.result()
t3 = time.time()
print(f"\n------ 双循环总耗时 {t3 - t2} -------\n")

优化后整体耗时可降低为原两个循环耗时的最大值，而非总和。

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