Python:用于数据库查询的并行化循环

Question

我将这个问题分为两部分。

我有与此类似的代码

for data in data_list:
rslt = query in databae where data == 10 # just some pseudo database query to give example but this single query usually takes around 30-50seconds.
if rslt.property == 'some condition here':
return rslt

这里的条件是

1. 我们必须在查询后返回与条件匹配的data_list的第一个元素。
2. 每个元素的每个数据库查询大约需要 30-40 秒。
3. data_list通常非常大，大约 15-20k 元素
4. 不幸的是，我们不能对整个data_list进行单个数据库查询。我们必须循环或一次一个元素来执行此操作。

现在我的问题是，

1. 如何优化此过程。目前，整个过程大约需要3-4小时。
2. 我读过关于python线程和多处理的信息，但我对在这种情况下哪一个合适感到困惑。

Answer 1

您可以考虑使用多处理Pool。然后，您可以使用map将可迭代的块发送给Pool的工作线程，以便根据给定的函数进行处理。因此，假设您的查询是一个函数，例如query(data)：

def query(data):
rslt = query in databae where data == 10
if rslt.property == 'some condition here':
return rslt

我们将像这样使用池：

from multiprocessing import Pool
with Pool() as pool:
results = pool.map(query, data_list)

现在，根据您的要求，我们将找到第一个：

print(next(filter(None, results)))

请注意，以这种方式query函数意味着results将是rslts 和Nones 的列表，我们正在寻找第一个非None结果。

---

一些注意事项：

1. 请注意，Pool的构造者的第一个参数是processes它允许您选择池将容纳多少个进程：

   如果进程None则使用 os.cpu_count() 返回的数字。

2. 请注意，map具有默认为 1 的chunksize参数，并允许选择传递给 worker 的块的大小：

   此方法将可迭代对象切成多个块，并将其作为单独的任务提交到进程池。可以通过将块大小设置为正整数来指定这些块的(近似)大小。

3. 继续map，文档建议对具有特定块的大型迭代对象使用imap以提高效率：

   请注意，这可能会导致非常长的可迭代对象内存使用率过高。考虑使用带有显式块大小选项的 imap() 或 imap_unordered() 以提高效率。

   从imap文档中：

   chunksize参数与 map() 方法使用的参数相同。对于非常长的可迭代对象，使用较大的块大小值可以使作业完成速度比使用默认值1快得多。

   因此，我们实际上可以提高效率并执行以下操作：

   chunksize = 100
   processes = 10
   with Pool(processes=processes) as pool:
   print(next(filter(None, pool.imap(query, data_list, chunksize=chunksize))))
   

   在这里，您可以使用chunksize甚至processes(从Pool回来)玩，看看哪种组合会产生最佳效果。

4. 如果您有兴趣，只需将 import 语句更改为：

   from multiprocessing.dummy import Pool
   

   正如文档所说：

   multiprocessing.dummy 复制了多处理的 API，但只不过是线程模块的包装器。

---

希望这以任何方式有所帮助