Concurrent.futures:线程与进程

Question

虽然我发现了一些与这个问题相关的假设和理论文章，但我发现的最接近的是这里，发布的答案与我认为我正在寻求帮助的内容正好相反(以防万一这个链接对其他人有帮助)。

我从Github，这里上的wiki中获得了以下代码。它的实现看起来非常简单，但是，我无法以它的本机形式使用它。

下面是我使用的“流程”代码：

import dask.dataframe as dd

from concurrent.futures import ProcessPoolExecutor
import pandas as pd
import gdelt

gd = gdelt.gdelt(version=2)

e = ProcessPoolExecutor()

def getter(x):
    try:
        date = x.strftime('%Y%m%d')
        d = gd.Search(date, coverage=True)
        d.to_csv("{}_gdeltdata.csv".format(date),encoding='utf-8',index=False)
    except:
        pass

results = list(e.map(getter,pd.date_range('2015 Apr 21','2018 Apr 21')))

以下是完整的错误：

BrokenProcessPool                         Traceback (most recent call last)
<ipython-input-1-874f937ce512> in <module>()
     21 
     22 # now pull the data; this will take a long time
---> 23 results = list(e.map(getter,pd.date_range('2015 Apr 21','2018 Apr 21')))
     24 
     25 

C:\Anaconda3\lib\concurrent\futures\process.py in_chain_from_iterable_of_lists(iterable)
    364     careful not to keep references to yielded objects.
    365     """
--> 366     for element in iterable:
    367         element.reverse()
    368         while element:

C:\Anaconda3\lib\concurrent\futures\_base.py in result_iterator()
    584                     # Careful not to keep a reference to the popped future
    585                     if timeout is None:
--> 586                         yield fs.pop().result()
    587                     else:
    588                         yield fs.pop().result(end_time - time.time())

C:\Anaconda3\lib\concurrent\futures\_base.py in result(self, timeout)
    430                 raise CancelledError()
    431             elif self._state == FINISHED:
--> 432                 return self.__get_result()
    433             else:
    434                 raise TimeoutError()

C:\Anaconda3\lib\concurrent\futures\_base.py in __get_result(self)
    382     def __get_result(self):
    383         if self._exception:
--> 384             raise self._exception
    385         else:
    386             return self._result

*BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.*

关于如何解决这个错误有什么想法吗？--我知道如果我将ProcessPoolExecutor改为ThreadPoolExecutor，这个问题似乎已经解决了(尽管我没有一路运行数据集，所以我不能完全确定)，但是，我相信如果我使用ProcessPoolExecutor，我会得到一个更快的结果。

最终，我将使用dask来处理Pandas中的数据。提前谢谢。

Answer 1

文档中的示例始终显示if __name__ == '__main__'子句中的执行情况。希望这个mcve能准确地模拟您的用例。

def gd(s):
    return s*3

def getter(w):
    return gd(w)

data = list('abcdefg')

def main():
    with ProcessPoolExecutor(max_workers=4) as executor:
        for thing in executor.map(getter, data):
            print(thing)

像这样执行，

#main()
if __name__ == '__main__':
    main()

但是像这样执行并不会引发BrokenProcessPool错误

main()
if __name__ == '__main__':
    #main()

尝试确保行results = list(e.map(getter,pd.date_range(...)))在*__main__*进程中运行。