Запускайте HTTP-запросы с PySpark параллельно и асинхронно

Question

У меня есть текстовый файл, содержащий несколько миллионов URL, и мне нужно выполнить запрос POST для каждого из этих URL.Я пытался сделать это на своей машине, но это занимает вечность, поэтому я хотел бы использовать вместо этого свой кластер Spark.

Я написал этот код PySpark:

from pyspark.sql.types import StringType
import requests

url = ["http://myurltoping.com"]
list_urls = url * 1000 # The final code will just import my text file
list_urls_df = spark.createDataFrame(list_urls, StringType())

print 'number of partitions: {}'.format(list_urls_df.rdd.getNumPartitions())

def execute_requests(list_of_url):
    final_iterator = []
    for url in list_of_url:
        r = requests.post(url.value)
        final_iterator.append((r.status_code, r.text))
    return iter(final_iterator)

processed_urls_df = list_urls_df.rdd.mapPartitions(execute_requests)

, но он все еще принимаетмного времени, как я могу сделать функцию execute_requests более эффективной, например, асинхронно запускать запросы в каждом разделе?

Спасибо!

Answer 1

Использование пакета python grequests (устанавливается с pip install grequests) может быть простым решением вашей проблемы без использования spark.

Документация (можно найти здесь https://github.com/kennethreitz/grequests) даетпростой пример:

import grequests

urls = [
    'http://www.heroku.com',
    'http://python-tablib.org',
    'http://httpbin.org',
    'http://python-requests.org',
    'http://fakedomain/',
    'http://kennethreitz.com'
]

Создание набора неотправленных запросов:

>>> rs = (grequests.get(u) for u in urls)

Отправка их всех одновременно:

>>> grequests.map(rs)
[<Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>, None, <Response [200]>]

Я обнаружил, что использование gevent с foreach на искре Dataframe приводит к некоторым странным ошибкам и не работает. Кажется, что искра также полагается на gevent, который используется grequests ...