Практическое, готовое к использованию решение для запуска парсеров в качестве задач Celery

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

Поскольку парсинг страниц обычно занимает время, часто лучше запускать его в фоновом режиме и время от времени уведомлять пользователей о его прогрессе.

Если вы читаете это, вероятно, вы уже знакомы с Scrapy и/или Celery. В случае, если вы новичок в Scrapy, это фреймворк с открытым исходным кодом, который позволяет нам писать парсеры для извлечения структурированных данных с веб-сайтов.

Celery, вкратце, является широко используемой, хорошо поддерживаемой, распределенной очередью задач на Python, которая позволяет запускать долгосрочные задачи в фоновом режиме с помощью пула выделенных рабочих процессов. Подумайте о Celery как о системе обработки почты в вашем местном почтовом отделении. Местное отделение (Celery) берет ваши письма (задачи) и распределяет их сотрудникам (рабочим) для их доставки.

Что включает этот статья?

В этой статье я расскажу о следующем:

• Почему стоит запускать парсеры без использования генерируемого кода Scrapy и как это сделать?
• Чтение статистики после выполнения парсеров
• Запуск парсеров в задачах Celery
• Как предотвратить завершение долго выполняющейся задачи парсинга Celery
• Как избежать известного исключения ReactorNotRestartable и дополнительный раздел о технических деталях, связанных с ним
• Как парсить без раскрытия вашего IP-адреса и как предотвратить блокировку ваших парсеров

(Я буду использовать термины "краулер" и "парсер" взаимозаменяемо.)

Почему запускать Scrapy без предоставленного шаблона?

Если вы следуете руководству из официальной документации, вам будет предложено использовать шаблонный код, который генерирует Scrapy с помощью команды scrapy startproject my_project_name. Хотя это предоставляет некоторую структуру кода для создания ваших парсеров, не все дополнительные функции необходимы, если ваши парсеры предназначены для запуска в качестве задач Celery.

Если вы создаете веб-приложение, которое использует задачи Celery для сбора данных, как я это делал, структура проекта, сгенерированная Scrapy, может не подходить к текущей структуре вашего веб-приложения. Поэтому может быть лучше настроить ваш парсер для запуска в виде скрипта.

Запуск Scrapy из скрипта!

Альтернативой использованию базового проекта, предоставляемого Scrapy, является запуск его из скрипта с помощью Scrapy Crawler API.

В последней официальной документации показано, как запускать Scrapy-парсеры с использованием scrapy.crawler.CrawlerProcess:

CrawlerProcess предоставляет простой интерфейс для запуска парсеров внутри скрипта и также используется внутренне в Scrapy CLI (scrapy команда в вашем терминале). Тем не менее, приведенный пример кода довольно ограничен, если вы хотите получить доступ к собранным данным, собранным пауками.

Как получить доступ к статистике паука при его запуске в скрипте?

Пауки Scrapy собирают полезную статистику и создают информативные логи. Я уверен, что вы видели некоторые из логов выше. Обратите внимание на скриншот выше, вы найдете полезную информацию, включая количество извлеченных элементов и т. д.

В моем проекте мне нужно сообщить пользователям о количестве извлеченных элементов по завершении каждой задачи. Важно иметь возможность получить доступ к собранной статистике. Однако использование CrawlerProcess для запуска пауков может быть сложным для доступа к собранной статистике.

Поэтому я рекомендую использовать низкоуровневый API Crawler:

В приведенном примере кода run_scrapy() создает Crawler, начинает обход с использованием MySpider и регистрирует количество извлеченных элементов.

Запуск парсера в задаче Celery

Я рассмотрел тему о том, почему и как запускать парсеры в скрипте, используя Crawler, вместо CrawlerProcess, чтобы иметь возможность получить доступ к данным, собранным пауками.

Существует множество способов определения задачи Celery. Для простоты я определю ее следующим образом:

Возможно, вас интересует, почему я использую multiprocessing.Process() внутри задачи Celery. Это предотвращает возникновение исключения ReactorNotRestartable(), о котором я подробнее расскажу в следующем разделе.

Установка более длительных временных ограничений для долго выполняющихся задач

По умолчанию Celery прекращает выполнение задачи, если она не может быть завершена в заданный временной лимит. Чтобы предотвратить прекращение работы ваших парсеров Scrapy, вы должны увеличить временной лимит. Для этого просто передайте временные лимиты в качестве ключевых аргументов следующим образом:

Если вы не знакомы с разницей между мягким и жестким лимитом, не стесняйтесь ознакомиться с этим разделом официальной документации.

Устранение исключения ReactorNotRestartable

В одном из предыдущих разделов я использовал multiprocessing.Process() для запуска парсера в отдельном процессе. Это было сделано для избежания исключения ReactorNotRestartable.

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

Чтобы исправить это, нам нужно изменить модель конкурентности, которую использует Celery. По умолчанию Celery использует модель предварительного форка при создании рабочих процессов. Мы должны изменить ее на многопоточность.

Для этого, при запуске celery workers, добавьте аргумент -P threads, чтобы запустить рабочие процессы в режиме многопоточности:

celery -A my_project worker -P threads

Запуск рабочих процессов Celery в режиме многопоточности, в сочетании с multiprocess.Process(), проблема с ReactorNotRestartable больше не должна возникать.

[Optional] Технические детали ReactorNotRestartable

Необязательно углубляться в детали, которые вызывают ReactorNotRestartable и обоснование решения.

ReactorNotRestartable является побочным эффектом следующих факторов:

Факт того, что Twisted хранит свое состояние глобально и сохраняет его после каждой задачи, так как рабочие процессы повторно используются. После остановки Reactor нет способа его перезапустить, если не создать новый рабочий процесс.

Мой подход к этой проблеме заключается в том, чтобы обеспечить создание нового Reactor для каждой задачи, поэтому я использую multiprocessing.Process() для создания отдельного процесса для запуска парсеров. (Новый процесс = новый Reactor)

Однако невозможно создать дочерний процесс из рабочего процесса, если сам рабочий процесс является дочерним процессом (поскольку он создается в режиме предварительной форки). Вот почему я должен указать Celery создавать рабочие процессы в виде отдельных потоков, а не процессов (-P threads).

Конечно, это также не является панацеей, если вы знакомы с ограничениями потоков из-за печально известного GIL, но в моем случае это более чем достаточно.

[Бонус] Предотвращение блокировки ваших парсеров

Современные веб-сайты используют техники для идентификации парсеров и немедленной блокировки их действий. Очень вероятно, что ваш IP-адрес будет навсегда заблокирован сайтами после обнаружения вас как робота. Чтобы обойти это, я лично использую ScraperAPI. Это сервис прокси с большим пулом IP-адресов из различных местоположений. Он поставляется с пакетом pip, который можно легко установить и использовать. Мой опыт работы с ним был довольно приятным. Если вы решили попробовать его, не стесняйтесь использовать промокод: SCRAPE1933074

Summary

В этой статье я рассмотрел, как запускать парсеры Scrapy в задачах Celery и предложил решения для некоторых проблем, с которыми вы можете столкнуться.

Проект теперь завершен, и мои клиенты уже несколько месяцев успешно используют его без проблем. Надеюсь, вы найдете эту статью полезной.

Буду благодарен за любые комментарии и предложения. 😃