【发布时间】：2020-09-14 00:40:45
【问题描述】：
我正在对 Dataframe 列应用一个函数，但我想让它更快，因为该函数在串行完成时需要大量处理时间。
df[df['codes']=='None']['q'][:1].apply(lambda x: clf(x,candidate_labels))
单行只需要 2.52 secs 运行，但是当使用多处理运行下面的代码时，它需要更长的时间 51.61 secs 并且我有大约 2500 rows 用于处理，所以它需要很多时间来运行函数。我希望至少能加快速度20%。
import multiprocessing
import pandas as pd
import numpy as np
def clf(x):
...
return list
def _apply_df(args):
df, func, kwargs = args
return df.apply(func, **kwargs)
def apply_by_multiprocessing(df, func, **kwargs):
workers = kwargs.pop('workers')
pool = multiprocessing.Pool(processes=workers)
result = pool.map(_apply_df, [(d, func, kwargs)
for d in np.array_split(df, workers)])
pool.close()
return pd.concat(list(result))
if __name__ == '__main__':
tart_time = time.time()
res=apply_by_multiprocessing(df[df['codes']=='None']['q'][:1],clf, workers=4)
print(res)
print("--- %s seconds ---" % (time.time() - start_time))
## run by 4 processors
我也为多处理尝试了不同的迭代，但似乎都没有加快流程，因为它们减慢了我的代码。
from pandarallel import pandarallel
import time
pandarallel.initialize(progress_bar=True)
start_time = time.time()
categories = df[df['codes']=='None']['q'][:10].parallel_apply(lambda x: clf(x,candidate_labels))
print("--- %s seconds ---" % (time.time() - start_time))
另一个实验：
import multiprocessing as mp
def clf:
...
return list
if __name__ == '__main__':
p = mp.Pool(processes=8)
pool_results = p.map(clf, df[df['codes']=='None']['q'][:1])
p.close()
p.join()
【问题讨论】：
-
clf在做什么？也许有矢量化的替代方案......此外，多进程需要时间来配置，并且它在规模上弥补了它，但如果没有明确的方法将函数拆分为线程，那么您将获得零时间并花更多时间配置调用 -
@RichieV 感谢您的回复，所以
clf实际上正在运行一个用于文本分类的预训练模型，我将其应用于 Dataframe 的每一行。每一行都是一个句子，需要分类到特定的主题中。 -
您使用的是哪种型号？我知道 sklearn 嵌入了多进程作为选项
-
否则，有没有办法将多行传递给您的模型？在创建多进程之前我会调查一下
-
@RichieV 我没有使用 sklearn 中的任何东西。它是 HuggingFace 的零样本文本分类模型。函数
clf只是在做candidate_labels=['a','b',..] cat=classifier(seq,candidate_labels) return cat
标签： pandas multithreading dataframe parallel-processing multiprocessing