Распараллеливать обработку огромного списка

Question

Мне нужно сделать много веб-изображений из доменов, хранящихся в TXT-файле (размер около 50 МБ).

Я хочу сделать это многопоточным. Поэтому я загружаю несколько записей в список Python и обрабатываю каждый поток.

Пример:

biglist = ['google.com','facebook.com','apple.com'] 
threads = [threading.Thread(target=fetch_url, args=(chuck,)) 
      for domain in biglist] 
for thread in threads: 
    thread.start() 

for thread in threads: 
    thread.join() 

Это работает, но мне кажется, что это не очень эффективно, так как есть много использования памяти, и это занимает много времени, чтобы закончить.

Какие лучшие способы достичь того, что я делаю?

Answer 1

Вы используете неправильную библиотеку. threading.Thread действительно не выигрывает от нескольких процессоров, поскольку он заблокирован Global Interpreter Lock.

Из документации модуля threading (c.f.):

CPython деталь реализации: В CPython, в связи с Global Interpreter Блокировка, только один поток может выполнять код Python сразу (даже хотя некоторые ориентированные на производительность библиотеки могут преодолеть это ограничение). Если вы хотите, чтобы ваше приложение лучше использовало вычислительные ресурсы многоядерных машин, вам рекомендуется использовать multiprocessing. Однако threading по-прежнему является подходящей моделью, если вы хотите одновременно запускать несколько задач с привязкой к I/O.

Я предлагаю вам использовать process pool из multiprocessing модуля и map() для вычисления результатов параллельно. Не имеет смысла использовать больше процессов, чем тогда есть процессоры.

Из документации модуля multiprocessing (c.f.):

multiprocessing представляет собой пакет, который поддерживает процессы нерестилища с использованием API, похожий на threading модуля. Пакет multiprocessing предлагает как локальный, так и удаленный параллелизм, эффективно пошаговый Global Interpreter Lock с использованием подпроцессов вместо потоков. Благодаря этому модуль multiprocessing позволяет программисту полностью использовать несколько процессоров на данной машине. Он работает как в Unix, так и в Windows.

Пример:

from multiprocessing import Pool 

number_of_processors = 3 

data = range(10) 

def func(x): 
    print "processing", x 
    return x*x 

pool = Pool(number_of_processors) 
ret = pool.map(func, data) 

Выход:

$ python test.py 
processing 0 
processing 1 
processing 3 
processing 2 
processing 4 
processing 5 
processing 6 
processing 7 
processing 8 
processing 9 
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81] 

Answer 2

Не следует использовать списки/темы, но очереди/процессы вместо этого.
Если вы знаете Redis, я предлагаю RQ (http://python-rq.org) - Я делаю то же самое и прекрасно работаю.

Answer 3

Добавить в moooeeeeps ответ. Существует еще один способ обработки множества соединений в одном потоке, не порождая дорогостоящих процессов. Gevent имеет аналогичную api для многопроцессорности/многопоточности.

Документы и учебные пособия:

• http://www.gevent.org/intro.html
• http://sdiehl.github.io/gevent-tutorial/

Также есть рамки Python для скребковых URLs: http://scrapy.org/

Пример из документации asyc/синхронизации выборки URL-адресов: импорт gevent.monkey gevent.monkey.patch_socket()

import gevent 
import urllib2 
import simplejson as json 

def fetch(pid): 
    response = urllib2.urlopen('http://json-time.appspot.com/time.json') 
    result = response.read() 
    json_result = json.loads(result) 
    datetime = json_result['datetime'] 

    print('Process %s: %s' % (pid, datetime)) 
    return json_result['datetime'] 

def synchronous(): 
    for i in range(1,10): 
     fetch(i) 

def asynchronous(): 
    threads = [] 
    for i in range(1,10): 
     threads.append(gevent.spawn(fetch, i)) 
    gevent.joinall(threads) 

print('Synchronous:') 
synchronous() 

print('Asynchronous:') 
asynchronous()