vendredi 17 juin 2016

Faster Scraping of JSON from API: Asynchronous or?

I need to scrape roughly 30GB of JSON data from a website API as quickly as possible. I don't need to parse it -- I just need to save everything that shows up on each API URL.

  • I can request quite a bit of data at a time -- say 1MB or even 50MB 'chunks' (API parameters are encoded in the URL and allow me to select how much data I want per request)
  • the API places a limit of 1 request per second.
  • I would like to accomplish this on a laptop and 100MB/sec internet connection

Currently, I am accomplishing this (synchronously & too slowly) by: -pre-computing all of the (encoded) URL's I want to scrape -using Python 3's requests library to request each URL and save the resulting JSON one-by-one in separate .txt files.

Basically, my synchronous, too-slow solution looks like this (simplified slightly):

#for each pre-computed encoded URL do:
    curr_url_request = requests.get(encoded_URL_i, timeout=timeout_secs)
    if curr_url_request.ok:
        with open('json_output.txt', 'w') as outfile:
            json.dump(curr_url_request.json(), outfile)

What would be a better/faster way to do this? Is there a straight-forward way to accomplish this asynchronously but respecting the 1-request-per-second threshold? I have read about grequests (no longer maintained?), twisted, asyncio, etc but do not have enough experience to know whether/if one of these is the right way to go.

EDIT Based on Kardaj's reply below, I decided to give async Tornado a try. Here's my current Tornado version (which is heavily based on one of the examples in their docs). It successfully limits concurrency.

The hangup is, how can I do an overall rate-limit of 1 request per second globally across all workers? (Kardaj, the async sleep makes a worker sleep before working, but does not check whether other workers 'wake up' and request at the same time).

from datetime import datetime
from datetime import timedelta
from tornado import httpclient, gen, ioloop, queues

URLS = ["https://baconipsum.com/api/?type=meat",
        "https://baconipsum.com/api/?type=filler",
        "https://baconipsum.com/api/?type=meat-and-filler",
        "https://baconipsum.com/api/?type=all-meat&paras=2&start-with-lorem=1"]

concurrency = 2


def handle_request(response):
    if response.code == 200:
        with open("FOO"+'.txt', "wb") as thisfile:#fix filenames to avoid overwrite
            thisfile.write(response.body) 

@gen.coroutine
def request_and_save_url(url):
    try:
        response = yield httpclient.AsyncHTTPClient().fetch(url, handle_request)
        print('fetched {0}'.format(url))
    except Exception as e:
        print('Exception: {0} {1}'.format(e, url))
        raise gen.Return([])


@gen.coroutine
def main():
    q = queues.Queue()
    tstart = datetime.now()
    fetching, fetched = set(), set()

    @gen.coroutine
    def fetch_url(worker_id):
        current_url = yield q.get()
        try:
            if current_url in fetching:
                return

            #print('fetching {0}'.format(current_url))
            print("Worker {0} starting, elapsed is {1}".format(worker_id, (datetime.now()-tstart).seconds ))
            fetching.add(current_url)
            yield request_and_save_url(current_url)
            fetched.add(current_url)

        finally:
            q.task_done()

    @gen.coroutine
    def worker(worker_id):
        while True:
            yield fetch_url(worker_id)

    # Fill a queue of URL's to scrape
    list = [q.put(url) for url in URLS] # this does not make a list...it just puts all the URLS into the Queue

    # Start workers, then wait for the work Queue to be empty.
    for ii in range(concurrency):
        worker(ii)
    yield q.join(timeout=timedelta(seconds=300))
    assert fetching == fetched
    print('Done in {0} seconds, fetched {1} URLs.'.format(
        datetime.now() - tstart, len(fetched)))


if __name__ == '__main__':
    import logging
    logging.basicConfig()
    io_loop = ioloop.IOLoop.current()
    io_loop.run_sync(main)

Aucun commentaire:

Enregistrer un commentaire