Python：マルチプロセッシングを使用する場合の処理​​された要素の数を確認してください。

Question

multiprocessingモジュールを使用して、並行URLを取得しています。問題がある

pat = re.compile("(?P<url>https?://[^\s]+)") 
def resolve_url(text): 
    missing = 0 
    bad = 0 
    url = 'before' 
    long_url = 'after' 
    match = pat.search(text) ## a text looks like "I am at here. http:.....(a URL)" 
    if not match: 
     missing = 1 
    else: 
     url = match.group("url") 
     try: 
      long_url = urllib2.urlopen(url).url 
     except: 
      bad = 1 
    return (url, long_url, missing, bad) 

if __name__ == '__main__': 
    pool = multiprocessing.Pool(100) 
    resolved_urls = pool.map(resolve_url, checkin5) ## checkin5 is a list of texts 

、私のcheckin5リストは周り60万要素が含まれており、この並列作業は本当に時間がかかる：私のコードは次のようです。私は、どのように多くの要素が解決されたのかをチェックしたいと思う。ループのためのシンプルであれば、私はこのような操作を行うことができます。

resolved_urls = [] 
now = time.time() 
for i, element in enumerate(checkin5): 
    resolved_urls.append(resolve_url(element)) 
    if i%1000 == 0: 
     print("from %d to %d: %2.5f seconds" %(i-1000, i, time.time()-now)) 
     now = time.time() 

しかし、今、私は効率を上げるために必要なので、マルチプロセスが必要であるが、私は、この場合にはいずれかのプロセスを検査する方法がわかりませんアイディア？ところで

は、上記の方法は、この場合にも機能するかどうかをチェックするために、私はおもちゃのコードを試みた：

import multiprocessing 
import time 

def cal(x): 
    res = x*x 
    return res 

if __name__ == '__main__': 
    pool = multiprocessing.Pool(4) 

    t0 = time.time() 
    result_list = pool.map(cal,range(1000000)) 
    print(time.time()-t0) 

    t0 = time.time() 
    for i, result in enumerate(pool.map(cal, range(1000000))): 
     if i%100000 == 0: 
      print("%d elements have been calculated, %2.5f" %(i, time.time()-t0)) 
      t0 = time.time() 

をし、結果は以下のとおりです。

0.465271949768 
0 elements have been calculated, 0.45459 
100000 elements have been calculated, 0.02211 
200000 elements have been calculated, 0.02142 
300000 elements have been calculated, 0.02118 
400000 elements have been calculated, 0.01068 
500000 elements have been calculated, 0.01038 
600000 elements have been calculated, 0.01391 
700000 elements have been calculated, 0.01174 
800000 elements have been calculated, 0.01098 
900000 elements have been calculated, 0.01319 

結果から、Iここではシングルプロセスのメソッドが機能しないと考えています。の後にpool.mapが最初に呼び出され、が計算され、完全なリストが得られたら、enumerateが始まります。

Answer 1

結果の順序を気にするかどうかによって、Pool.imapまたはPool.imap_unorderedのいずれかでこれを行うことができます。それらは両方ともノンブロッキングです...

resolved_urls = [] 
pool = multiprocessing.Pool(100) 
res = pool.imap(resolve_url, checkin5) 

for x in res: 
    resolved_urls.append(x) 
    print 'finished one' 
    # ... whatever counting/tracking code you want here 

Answer 2

まず、@ danf1024は答えがあると思います。これは、pool.mapからpool.imapに切り替えるときの遅い問題に対処するためです。二乗がWebページをダウンロードして解析し、言う、と比較すると、各プロセッサは、入力の大きな途切れないチャンクを処理できる場合は、並列化が利益を持っています安く操作ですので

from multiprocessing import Pool 


def square(x): 
    return x * x 


N = 10 ** 4 
l = list(range(N)) 


def test_map(n=N): 
    list(Pool().map(square, l)) 

# In [3]: %timeit -n10 q.test_map() 
# 10 loops, best of 3: 14.2 ms per loop 


def test_imap(n=N): 
    list(Pool().imap(square, l)) 

# In [4]: %timeit -n10 q.test_imap() 
# 10 loops, best of 3: 232 ms per loop 


def test_imap1(n=N): 
    list(Pool(processes=1).imap(square, l)) 

# In [5]: %timeit -n10 q.test_imap1() 
# 10 loops, best of 3: 191 ms per loop 


def test_map_naive(n=N): 
    # cast map to list in python3 
    list(map(square, l)) 

# In [6]: %timeit -n10 q.test_map_naive() 
# 10 loops, best of 3: 1.2 ms per loop 

：ここ

は少し実験です。 imapの場合はそうではありません。これは4つのコアで非常に不十分です。面白いことに、プロセスの数を1に制限すると、競合条件が削除されるため、imapが速くなります。

しかし、より高価な操作に移行すると、とmapの差はますます小さくなります。