并行化一个刮板程序，什么也不会发生

Question

我试着并行化一个刮板。不幸的是，当我执行这段代码时，它运行得异常长。直到我停下来。也不会生成输出。我是不是漏掉了什么？我使用os.system的问题是什么？

首先定义函数，然后加载数据池，然后将其输入到多进程中。

总而言之，我想这样做：

def cube(x):
    return x**3
pool = mp.Pool(processes=2)
results = pool.map(cube, range(1,7))
print(results)

但这个小计算现在已经运行了5分钟以上。所以我认为代码本身没有错误。而是我对多进程的理解

    from multiprocessing import Pool
    import os
    import json
    import datetime
    from dateutil.relativedelta import relativedelta
    import re
    os.chdir(r'C:\Users\final_tweets_de')
    
    p = Pool(5)
    
    import time
    
    
    
    def get_id(data_tweets):
        for i in range(len(data_tweets)):
    
            account = data_tweets[i]['user_screen_name']
            created = datetime.datetime.strptime(data_tweets[i]['date'], '%Y-%m-%d').date()
            until = created + relativedelta(days=10)
            id = data_tweets[i]['id']
            filename = re.search(r'(.*).json',file).group(1) + '_' + 'tweet_id_' +str(id)+ '_' + 'user_id_' + str(data_tweets[i]['user_id'])
    
    
            os.system('snscrape twitter-search "(to:'+account+') since:'+created.strftime("%Y-%m-%d")+' until:'+until.strftime("%Y-%m-%d")+' filter:replies" >C:\\Users\\test_'+filename)
           
        
    
    
    directory =r'C:\Users\final_tweets_de'
    path= r'C:\Users\final_tweets_de'
    for file in os.listdir(directory):
        fh = open(os.path.join(path, file),'r')
        print(file)
        
    
    
        with open(file, 'r', encoding='utf-8') as json_file:
            data_tweets = json.load(json_file)
            
            
        data_tweets = data_tweets[0:5]
        start = time.time()
        print("start")
        
       
        p.map(get_id, data_tweets)
        p.terminate()
        p.join()
        end = time.time()
        print(end - start)

更新

代码不能运行的原因首先是@Booboo解决的问题。另一个是当使用windows时，脚本必须通过cmd启动，在多处理的情况下。

就像这里：Python multiprocessing example not working

现在我看到键错误0。如果我运行代码。

import multiprocessing as mp
import os
import json
import datetime
from dateutil.relativedelta import relativedelta
import re
os.chdir(r'C:\Users\Paul\Documents\Uni\Masterarbeit\Datengewinnung\final_tweets_de')



import time


def get_id(data_tweets):
    for i in range(len(data_tweets)):
        print(i)

        account = data_tweets[i]['user_screen_name']
        created = datetime.datetime.strptime(data_tweets[i]['date'], '%Y-%m-%d').date()
        until = created + relativedelta(days=10)
        id = data_tweets[i]['id']
        filename = re.search(r'(.*).json',file).group(1) + '_' + 'tweet_id_' +str(id)+ '_' + 'user_id_' + str(data_tweets[i]['user_id'])
        
        try:
            os.system('snscrape twitter-search "(to:'+account+') since:'+created.strftime("%Y-%m-%d")+' until:'+until.strftime("%Y-%m-%d")+' filter:replies" >C:\\Users\\Paul\\Documents\\Uni\\Masterarbeit\\Datengewinnung\\Tweets_antworten\\Antworten\\test_'+filename)
        except:
            continue
    


    


directory =r'C:\Users\Paul\Documents\Uni\Masterarbeit\Datengewinnung\final_tweets_de'
path= r'C:\Users\Paul\Documents\Uni\Masterarbeit\Datengewinnung\final_tweets_de'

for file in os.listdir(directory):
    fh = open(os.path.join(path, file),'r')
    print(file)

    


    with open(file, 'r', encoding='utf-8') as json_file:
        data_tweets = json.load(json_file)
        
       
    data_tweets = data_tweets[0:2]
    start = time.time()
    print("start")
    if __name__ == '__main__': 
        pool = mp.Pool(processes=2)
        pool.map(get_id, data_tweets)
        

    end = time.time()
    print(end - start)
    del(data_tweets)

输出：

(NLP 2) C:\Users\Paul\Documents\Uni\Masterarbeit\Datengewinnung\Tweets_antworten>python scrape_id_antworten_parallel.py
corona.json
start
corona.json
corona.json
start
0.0009980201721191406
coronavirus.json
start
0.0
coronavirus.json
start
0.0
covid.json
start
0.0
SARS_CoV.json
start
0.0
0
0
multiprocessing.pool.RemoteTraceback:
"""
Traceback (most recent call last):
  File "C:\Users\Paul\Anaconda3\envs\NLP 2\lib\multiprocessing\pool.py", line 121, in worker
    result = (True, func(*args, **kwds))
  File "C:\Users\Paul\Anaconda3\envs\NLP 2\lib\multiprocessing\pool.py", line 44, in mapstar
    return list(map(*args))
  File "C:\Users\Paul\Documents\Uni\Masterarbeit\Datengewinnung\Tweets_antworten\scrape_id_antworten_parallel.py", line 25, in get_id
    account = data_tweets[i]['user_screen_name']
KeyError: 0
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "scrape_id_antworten_parallel.py", line 60, in <module>
    pool.map(get_id, data_tweets)
  File "C:\Users\Paul\Anaconda3\envs\NLP 2\lib\multiprocessing\pool.py", line 268, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "C:\Users\Paul\Anaconda3\envs\NLP 2\lib\multiprocessing\pool.py", line 657, in get
    raise self._value
KeyError: 0

Answer 1

我可以从path= r'C:\Users\final_tweets_de'上看到你的平台是Windows。在Windows下进行多进程处理时，创建子进程的代码必须绝对在一个块中执行，如下所示：

import multiprocessing as mp

def cube(x):
    return x**3

if __name__ == '__main__':
    pool = mp.Pool(processes=2)
    results = pool.map(cube, range(1,7))
    print(results)

否则，您将进入一个递归循环，其中子进程将尝试创建一个新的池和新的子进程ad-infinitum。修复此问题并重新测试。最简单的方法是将您的代码包装在一个函数中(例如，将其称为main )，然后添加：

if __name__ == '__main_':
    main()

另外，为什么在实际示例中只使用2个进程或5个进程。通过不为Pool构造函数指定参数，您将创建一个等于计算机上实际可用处理器数量的池大小。这不是一个糟糕的违约。