Wie kann ich schneller kratzen

16

Die Arbeit hier besteht darin, eine API einer Site zu kratzen, die von https://xxx.xxx.xxx/xxx/1.jsonbis beginnt , https://xxx.xxx.xxx/xxx/1417749.jsonund sie genau in mongodb zu schreiben. Dafür habe ich folgenden Code:

client = pymongo.MongoClient("mongodb://127.0.0.1:27017")
db = client["thread1"]
com = db["threadcol"]
start_time = time.time()
write_log = open("logging.log", "a")
min = 1
max = 1417749
for n in range(min, max):
    response = requests.get("https:/xx.xxx.xxx/{}.json".format(str(n)))
    if response.status_code == 200:
        parsed = json.loads(response.text)
        inserted = com.insert_one(parsed)
        write_log.write(str(n) + "\t" + str(inserted) + "\n")
        print(str(n) + "\t" + str(inserted) + "\n")
write_log.close()

Aber es braucht viel Zeit, um die Aufgabe zu erledigen. Die Frage hier ist, wie ich diesen Prozess beschleunigen kann.

Tek Nath
quelle
Haben Sie zuerst versucht zu bewerten, wie lange es dauert, einen einzelnen JSON zu verarbeiten? Angenommen, es dauert 300 ms pro Datensatz, können Sie alle diese Datensätze nacheinander in etwa 5 Tagen verarbeiten.
Tuxdna

Antworten:

5

asyncio ist auch eine Lösung, wenn Sie kein Multithreading verwenden möchten

import time
import pymongo
import json
import asyncio
from aiohttp import ClientSession


async def get_url(url, session):
    async with session.get(url) as response:
        if response.status == 200:
            return await response.text()


async def create_task(sem, url, session):
    async with sem:
        response = await get_url(url, session)
        if response:
            parsed = json.loads(response)
            n = url.rsplit('/', 1)[1]
            inserted = com.insert_one(parsed)
            write_log.write(str(n) + "\t" + str(inserted) + "\n")
            print(str(n) + "\t" + str(inserted) + "\n")


async def run(minimum, maximum):
    url = 'https:/xx.xxx.xxx/{}.json'
    tasks = []
    sem = asyncio.Semaphore(1000)   # Maximize the concurrent sessions to 1000, stay below the max open sockets allowed
    async with ClientSession() as session:
        for n in range(minimum, maximum):
            task = asyncio.ensure_future(create_task(sem, url.format(n), session))
            tasks.append(task)
        responses = asyncio.gather(*tasks)
        await responses


client = pymongo.MongoClient("mongodb://127.0.0.1:27017")
db = client["thread1"]
com = db["threadcol"]
start_time = time.time()
write_log = open("logging.log", "a")
min_item = 1
max_item = 100

loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run(min_item, max_item))
loop.run_until_complete(future)
write_log.close()
Frans
quelle
1
Die Verwendung von Async funktionierte schneller als Multithreading.
Tek Nath
Danke für die Rückmeldung. Interessantes Ergebnis.
Frans
10

Sie können verschiedene Dinge tun:

  1. Verbindung wiederverwenden. Laut der folgenden Benchmark ist es ungefähr dreimal schneller
  2. Sie können mehrere Prozesse parallel abkratzen

Parallelcode von hier

from threading import Thread
from Queue import Queue
q = Queue(concurrent * 2)
for i in range(concurrent):
    t = Thread(target=doWork)
    t.daemon = True
    t.start()
try:
    for url in open('urllist.txt'):
        q.put(url.strip())
    q.join()
except KeyboardInterrupt:
    sys.exit(1)

Timings aus dieser Frage für wiederverwendbare Verbindung

>>> timeit.timeit('_ = requests.get("https://www.wikipedia.org")', 'import requests', number=100)
Starting new HTTPS connection (1): www.wikipedia.org
Starting new HTTPS connection (1): www.wikipedia.org
Starting new HTTPS connection (1): www.wikipedia.org
...
Starting new HTTPS connection (1): www.wikipedia.org
Starting new HTTPS connection (1): www.wikipedia.org
Starting new HTTPS connection (1): www.wikipedia.org
52.74904417991638
>>> timeit.timeit('_ = session.get("https://www.wikipedia.org")', 'import requests; session = requests.Session()', number=100)
Starting new HTTPS connection (1): www.wikipedia.org
15.770191192626953
keiv.fly
quelle
4

Was Sie wahrscheinlich suchen, ist asynchrones Scraping. Ich würde Ihnen empfehlen, einige Stapel von URLs zu erstellen, dh 5 URLs (versuchen Sie, die Website nicht zu verdrängen) und sie asynchron zu kratzen. Wenn Sie nicht viel über Async wissen, googeln Sie für die Bibliothek asyncio. Ich hoffe, ich konnte dir helfen :)

T Piper
quelle
1
Können Sie weitere Details hinzufügen?
Tek Nath
3

Versuchen Sie, die Anforderungen zu unterteilen und die MongoDB-Massenschreiboperation zu verwenden.

  • Gruppieren Sie die Anforderungen (100 Anforderungen pro Gruppe).
  • Durch die Gruppen iterieren
  • Verwenden Sie das asynchrone Anforderungsmodell, um die Daten abzurufen (URL in einer Gruppe).
  • Aktualisieren Sie die Datenbank nach Abschluss einer Gruppe (Massenschreibvorgang).

Dies kann auf folgende Weise viel Zeit sparen: * MongoDB-Schreiblatenz * Synchrone Netzwerkanruflatenz

Erhöhen Sie jedoch nicht die Anzahl der parallelen Anforderungen (Blockgröße). Dies erhöht die Netzwerklast des Servers, und der Server könnte dies als DDoS-Angriff betrachten.

  1. https://api.mongodb.com/python/current/examples/bulk.html
thuva4
quelle
1
Können Sie mit Code für die Gruppierung der Anfragen und Gruppenabruf helfen
Tek Nath
3

Unter der Annahme, dass Sie nicht von der API blockiert werden und dass es keine Ratenbeschränkungen gibt, sollte dieser Code den Prozess 50-mal beschleunigen (möglicherweise mehr, da alle Anforderungen jetzt mit derselben Sitzung gesendet werden).

import pymongo
import threading

client = pymongo.MongoClient("mongodb://127.0.0.1:27017")
db = client["thread1"]
com = db["threadcol"]
start_time = time.time()
logs=[]

number_of_json_objects=1417750
number_of_threads=50

session=requests.session()

def scrap_write_log(session,start,end):
    for n in range(start, end):
        response = session.get("https:/xx.xxx.xxx/{}.json".format(n))
        if response.status_code == 200:
            try:
                logs.append(str(n) + "\t" + str(com.insert_one(json.loads(response.text))) + "\n")
                print(str(n) + "\t" + str(inserted) + "\n")
            except:
                logs.append(str(n) + "\t" + "Failed to insert" + "\n")
                print(str(n) + "\t" + "Failed to insert" + "\n")

thread_ranges=[[x,x+number_of_json_objects//number_of_threads] for x in range(0,number_of_json_objects,number_of_json_objects//number_of_threads)]

threads=[threading.Thread(target=scrap_write_log, args=(session,start_and_end[0],start_and_end[1])) for start_and_end in thread_ranges]

for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

with open("logging.log", "a") as f:
    for line in logs:
        f.write(line)
Ibrahim Dar
quelle
2

Ich hatte vor vielen Jahren die gleiche Frage. Ich bin nie zufrieden mit Python-basierten Antworten, die ziemlich langsam oder zu kompliziert sind. Nachdem ich zu anderen ausgereiften Werkzeugen gewechselt bin, ist die Geschwindigkeit schnell und ich komme nie mehr zurück.

Kürzlich habe ich solche Schritte verwendet, um den Prozess wie folgt zu beschleunigen.

  1. Generieren Sie eine Reihe von URLs in txt
  2. Verwenden Sie aria2c -x16 -d ~/Downloads -i /path/to/urls.txtdiese Option, um diese Dateien herunterzuladen
  3. lokal analysieren

Dies ist der schnellste Prozess, den ich bisher gemacht habe.

In Bezug auf das Scraping von Webseiten lade ich sogar die erforderliche * .html herunter, anstatt die Seite einzeln zu besuchen, was eigentlich keinen Unterschied macht. Wenn schlagen Sie die Seite besuchen, mit Python - Tool wie requestsoder scrapyoder urllib, es Cache noch und Download für Sie die gesamten Web - Inhalte.

anonym
quelle
1

Erstellen Sie zuerst eine Liste aller Links, da alle gleich sind. Ändern Sie sie einfach und wiederholen Sie sie.

list_of_links=[]
for i in range(1,1417749):
    list_of_links.append("https:/xx.xxx.xxx/{}.json".format(str(i)))

t_no=2
for i in range(0, len(list_of_links), t_no):
    all_t = []
    twenty_links = list_of_links[i:i + t_no]
    for link in twenty_links:
        obj_new = Demo(link,)
        t = threading.Thread(target=obj_new.get_json)
        t.start()
        all_t.append(t)
    for t in all_t:
        t.join()

class Demo:
    def __init__(self, url):
        self.json_url = url

def get_json(self):
    try:
       your logic
    except Exception as e:
       print(e)

Durch einfaches Erhöhen oder Verringern von t_no können Sie die Anzahl der Threads ändern.

Mobin Alhassan
quelle