Inhaltsverzeichnis
Bereinigen schmutziger Daten und Umgang mit Randfällen
Aus dem Internet extrahierte Daten sind häufig inkonsistent oder unvollständig, was bei den meisten Scraping-Projekten eine Herausforderung darstellt. In diesem Abschnitt werden Strategien zum Umgang mit solchen Sonderfällen erläutert, darunter fehlende Daten, unterschiedliche Datenformate und doppelte Einträge.
Hier sind einige Strategien zum Umgang mit Grenzfällen
- Versuchen/Außer – Nützlich, um Fehler elegant zu behandeln.
- Bedingtes Parsen – Implementieren Sie bedingte Logik, um Daten basierend auf ihrer Struktur unterschiedlich zu analysieren.
- Datenklassen – Verwenden Sie die Datenklassen von Python, um Ihre Daten zu strukturieren und so die Bereinigung und Bearbeitung zu erleichtern.
- Datenpipelines – Implementieren Sie eine Datenpipeline, um Daten vor der Speicherung zu bereinigen, zu validieren und zu transformieren.
- Bereinigen während der Datenanalyse – Führen Sie im Rahmen Ihres Datenanalyseprozesses eine Datenbereinigung durch.
Implementieren von Datenklassen für strukturierte Daten
In diesem Abschnitt verwenden wir Datenklassen, um unsere Scraping-Daten zu strukturieren und so Konsistenz und einfache Manipulation sicherzustellen.
Bevor wir die Daten bereinigen und verarbeiten, definieren wir die notwendigen Importe aus dem dataclasses Modul und richten Sie unser ein Product Datenklasse. Diese Klasse dient als Blaupause für die Strukturierung unserer gecrackten Produktdaten.
from dataclasses import dataclass, field, InitVar
@dataclass
class Product:
name: str = ""
price: InitVar(str) = ""
url: str = ""
price_gbp: float = field(init=False)
price_usd: float = field(init=False)
def __post_init__(self, price):
self.url = self.normalize_url()
self.price_gbp = self.clean_price(price)
self.price_usd = self.price_to_usd()
Hier importieren wir dataclass, fieldUnd InitVar von dem dataclasses Modul. Der @dataclass Der Dekorator fügt dem automatisch spezielle Methoden hinzu Product Klasse, wie z __init__ Und __repr__basierend auf den Klassenattributen.
Die Produktklasse wird mit mehreren Attributen definiert:
- Name: Eine Zeichenfolge, die den Produktnamen darstellt.
- Preis: Eine Initialisierungsvariable (
InitVar), die den Preis als String enthält. Dieses Attribut ist nicht in der Klasse enthalten__init__Methode und wird zur Verarbeitung verwendet, bevor sie verworfen wird. - URL (URL): Eine Zeichenfolge, die die Produkt-URL darstellt.
- Preis_GBP: Ein Float, der den Produktpreis in GBP darstellt. Dieses Attribut ist nicht in der Klasse enthalten
__init__-Methode und wird während der Initialisierung gesetzt (init=False). - Preis (USD: Ein Float, der den Produktpreis in USD darstellt. Wie
price_gbpdieses Attribut ist nicht in der Klasse enthalten__init__Methode und wird während der Initialisierung festgelegt.
Dieses Setup bietet eine strukturierte Möglichkeit zur Verwaltung von Produktdaten, einschließlich der Bereinigung und Umrechnung von Preisen, der Normalisierung von URLs und mehr.
Die nächsten Schritte umfassen die Implementierung von Methoden innerhalb der Product Klasse, um diese Operationen auszuführen.
Bereinigen Sie den Preis
Mit unserer Product Nachdem die Datenklasse definiert wurde, besteht der nächste Schritt darin, Methoden zur Bereinigung der Preisdaten zu implementieren. Wir bereinigen die Preiszeichenfolge, indem wir unnötige Zeichen wie das Währungssymbol („£“) und Verkaufspreisindikatoren („Sale price£“, „Sale priceFrom £“) entfernen.
Definieren Sie eine Methode clean_price innerhalb der Product Klasse, die eine Preiszeichenfolge annimmt, alle nicht numerischen Zeichen entfernt und den bereinigten Preis als Float zurückgibt.
def clean_price(self, price_string: str):
price_string = price_string.strip()
price_string = price_string.replace("Sale price£", "")
price_string = price_string.replace("Sale priceFrom £", "")
price_string = price_string.replace("£", "")
if price_string == "":
return 0.0
return float(price_string)
- Diese Methode entfernt zunächst alle führenden oder nachfolgenden Leerzeichen aus der Preiszeichenfolge
- Anschließend werden alle Instanzen von „Sale price£“ und „Sale priceFrom £“ aus der Zeichenfolge entfernt
- Danach wird das „£“-Symbol entfernt.
Wenn die resultierende Zeichenfolge leer ist, wird „0,0“ zurückgegeben, was darauf hinweist, dass der Preis fehlt oder nicht verfügbar ist. Andernfalls wird der bereinigte String in einen Float umgewandelt und zurückgegeben.
Konvertieren Sie den Preis
Nachdem wir den Preis bereinigt haben, müssen wir ihn von GBP in USD umrechnen, um die Währung in unserem gesamten Datensatz zu standardisieren, insbesondere beim Umgang mit internationalen Daten.
def price_to_usd(self):
return self.price_gbp * 1.28
Diese Methode multipliziert den bereinigten GBP-Preis mit dem Umrechnungskurs (1,28 in diesem Beispiel), um den Preis in USD zu berechnen. Dieser Umrechnungskurs kann basierend auf aktuellen Wechselkursen dynamisch aktualisiert werden.
Normalisieren Sie die URL
Ein weiterer häufiger Randfall bei Scraping-Daten sind inkonsistente URL-Formate. Einige URLs können relative Pfade sein, während andere absolute URLs sind. Durch die Normalisierung von URLs wird sichergestellt, dass sie konsistent formatiert sind, was die Arbeit mit ihnen erleichtert.
Wir werden a definieren normalize_url Methode innerhalb der Product Klasse, die prüft, ob die URL mit beginnt http:// oder https://. Wenn nicht, wird „vorangestellt“http://example.com“ zur URL.
def normalize_url(self):
if self.url == "":
return "missing"
if not self.url.startswith("http://") and not self.url.startswith("https://"):
return "http://example.com" + self.url
return self.url
- Diese Methode prüft zunächst, ob die URL leer ist. Wenn dies der Fall ist, wird „fehlend“ zurückgegeben.
- Anschließend wird geprüft, ob die URL nicht mit „http://“ oder „https://“ beginnt.
- In diesem Fall wird der URL „http://example.com“ vorangestellt, um sicherzustellen, dass sie ein gültiges Format hat.
- Wenn die URL bereits mit „http://“ oder „https://“ beginnt, wird die URL unverändert zurückgegeben.
Testen Sie die Produktdatenklasse
Testen Sie abschließend die Datenklasse „Product“ mit einigen Beispieldaten.
# Sample scraped data
scraped_products = (
{"name": "Delicious Chocolate", "price": "Sale priceFrom £1.99", "url": "/delicious-chocolate"},
{"name": "Yummy Cookies", "price": "£2.50", "url": "http://example.com/yummy-cookies"},
{"name": "Tasty Candy", "price": "Sale price£0.99", "url": "/apple-pies"}
)
# Process and clean the data
processed_products = (Product(name=product("name"), price=product("price"), url=product("url")) for product in scraped_products)
# Display the processed products
for product in processed_products:
print(f"Name: {product.name}, GBP_Price: £{product.price_gbp}, USD_Price: ${product.price_usd}, URL: {product.url}")
Dieser Code erstellt eine Liste von Wörterbüchern, die einige ausgelesene Produktdaten darstellen. Anschließend iteriert er über diese Liste und erstellt ein Product Instanz für jedes Wörterbuch und Anhängen an die processed_products Liste.
Schließlich iteriert es über die processed_products Liste auf und druckt den Namen, den GBP-Preis, den USD-Preis und die URL jedes Produkts aus.
Name: Delicious Chocolate, GBP_Price: £1.99, USD_Price: $2.5472,URL: http://example.com/delicious-chocolate
Name: Yummy Cookies, GBP_Price: £2.5, USD_Price: $3.2,URL: http://example.com/yummy-cookies
Name: Tasty Candy, GBP_Price: £0.99, USD_Price: $1.2672,URL: http://example.com/apple-pies
Dadurch wird überprüft, ob die Produktdatenklasse die gelöschten Daten korrekt bereinigt und verarbeitet.
Vollständiger Datenklassencode
Hier ist der vollständige Code für Product Beispiel für eine Datenklasse.
from dataclasses import dataclass, field, InitVar
@dataclass
class Product:
name: str = ""
price: InitVar(str) = ""
url: str = ""
price_gbp: float = field(init=False)
price_usd: float = field(init=False)
def __post_init__(self, price):
self.url = self.normalize_url()
self.price_gbp = self.clean_price(price)
self.price_usd = self.price_to_usd()
def clean_price(self, price_string: str):
price_string = price_string.strip()
price_string = price_string.replace("Sale price£", "")
price_string = price_string.replace("Sale priceFrom £", "")
price_string = price_string.replace("£", "")
if price_string == "":
return 0.0
return float(price_string)
def normalize_url(self):
if self.url == "":
return "missing"
if not self.url.startswith("http://") and not self.url.startswith("https://"):
return "http://example.com" + self.url
return self.url
def price_to_usd(self):
return self.price_gbp * 1.28
# Sample scraped data
sample_products = (
{'name': 'Artisanal Chocolate', 'price': 'Sale price£2.99', 'url': '/artisanal-chocolate'},
{'name': 'Gourmet Cookies', 'price': '£5.50', 'url': 'http://example.com/gourmet-cookies'},
{'name': 'Artisanal Chocolate', 'price': 'Sale price£2.99', 'url': '/artisanal-chocolate'},
{"name": "Delicious Chocolate", "price": "Sale priceFrom £1.99", "url": "/delicious-chocolate"},
{"name": "Yummy Cookies", "price": "£2.50", "url": "http://example.com/yummy-cookies"},
{"name": "Tasty Candy", "price": "Sale price£0.99", "url": "/apple-pies"}
)
# Process and clean the data
processed_products = ()
for product in scraped_products:
prod = Product(name=product("name"),
price=product("price"),
url=product("url"))
processed_products.append(prod)
# Display the processed products
for product in processed_products:
print(f"Name: {product.name}, GBP_Price: £{product.price_gbp}, USD_Price: ${product.price_usd},URL: {product.url}")
Verarbeiten und speichern Sie Scraped-Daten mit Datenpipelines
Nachdem wir unsere Daten mithilfe des strukturierten Ansatzes der Produktdatenklasse bereinigt haben, fahren wir mit dem nächsten entscheidenden Schritt fort: der effizienten Verarbeitung und Speicherung dieser Daten.
In dieser Phase spielen Datenpipelines eine zentrale Rolle, die es uns ermöglichen, die Daten systematisch zu verarbeiten, bevor wir sie speichern. Zu den Vorgängen innerhalb unserer Datenpipeline gehören:
- Duplikatsprüfung: Überprüfen Sie, ob ein Element bereits im Datensatz vorhanden ist, um Redundanz zu vermeiden.
- Datenwarteschlangenverwaltung: Verarbeitete Daten vor dem Speichern vorübergehend speichern und Fluss und Volumen verwalten.
- Periodische Datenspeicherung: Speichern Sie die Daten in regelmäßigen Abständen oder basierend auf bestimmten Auslösern in einer CSV-Datei.
Einrichten der ProductDataPipeline-Klasse
Definieren wir zunächst die Struktur unseres ProductDataPipeline Klasse, mit Schwerpunkt auf der Initialisierung und den grundlegenden Methoden, die die oben genannten Operationen unterstützen:
import csv
from dataclasses import asdict, fields
class ProductDataPipeline:
def __init__(self, csv_filename='product_data.csv', storage_queue_limit=10):
self.names_seen = set()
self.storage_queue = ()
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
pass
def clean_raw_product(self, scraped_data):
pass
def is_duplicate(self, product_data):
pass
def add_product(self, scraped_data):
pass
def close_pipeline(self):
pass
Der __init__ Die Methode richtet Anfangsbedingungen ein, einschließlich eines Satzes zum Verfolgen gesehener Produktnamen (um Duplikate zu überprüfen), einer Warteschlange zum vorübergehenden Speichern von Produkten und einer Konfiguration für die CSV-Dateiausgabe.
Rohproduktdaten bereinigen
Bevor ein Produkt zu unserer Verarbeitungswarteschlange hinzugefügt werden kann, muss es zunächst bereinigt und strukturiert werden. Der clean_raw_product Die Methode erreicht dies, indem sie rohe Scraped-Daten in eine Instanz von konvertiert Product Datenklasse, um sicherzustellen, dass unsere Daten der erwarteten Struktur und den erwarteten Typen entsprechen.
def clean_raw_product(self, scraped_data):
cleaned_data = {
"name": scraped_data.get("name", ""),
"price": scraped_data.get("price", ""),
"url": scraped_data.get("url", "")
}
return Product(**cleaned_data)
Sobald die Rohproduktdaten bereinigt sind, müssen sie auf Duplikate überprüft und bei Eindeutigkeit zur Verarbeitungswarteschlange hinzugefügt werden. Dies wird von der verwaltet add_product Und is_duplicate Methoden bzw.
Produkte hinzufügen und auf Duplikate prüfen
Der is_duplicate() Funktion prüft, ob sich ein Produkt bereits im befindet names_seen Liste. Wenn dies der Fall ist, wird eine Nachricht gedruckt und zurückgegeben True, was darauf hinweist, dass es sich bei dem Produkt um ein Duplikat handelt. Ist dies nicht der Fall, wird der Produktname hinzugefügt names_seen Liste und kehrt zurück False.
def is_duplicate(self, product_data):
if product_data.name in self.names_seen:
print(f"Duplicate item found: {product_data.name}. Item dropped.")
return True
self.names_seen.add(product_data.name)
return False
Der add_product Die Methode bereinigt zunächst die gelöschten Daten und erstellt eine Product Objekt. Anschließend prüft es, ob es sich bei dem Produkt um ein Duplikat handelt is_duplicate Methode. Wenn es sich nicht um ein Duplikat handelt, wird das Produkt dem hinzugefügt storage_queue. Schließlich prüft es, ob die Speicherwarteschlange ihr Limit erreicht hat, und ruft in diesem Fall das auf save_to_csv Methode zum Speichern der Produkte in der CSV-Datei.
def add_product(self, scraped_data):
product = self.clean_raw_product(scraped_data)
if not self.is_duplicate(product):
self.storage_queue.append(product)
if len(self.storage_queue) >= self.storage_queue_limit:
self.save_to_csv()
Regelmäßige Speicherung der Daten im CSV-Format
Der save_to_csv Die Methode wird entweder ausgelöst, wenn die Speicherwarteschlange ihr Limit erreicht oder wenn die Pipeline geschlossen wird, wodurch die Datenpersistenz sichergestellt wird.
def save_to_csv(self):
if not self.storage_queue:
return
self.csv_file_open = True
with open(self.csv_filename, mode='a', newline='', encoding='utf-8') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=(field.name for field in fields(self.storage_queue(0))))
if csvfile.tell() == 0:
writer.writeheader()
for product in self.storage_queue:
writer.writerow(asdict(product))
self.storage_queue.clear()
self.csv_file_open = False
Der save_to_csv Die Methode ist so konzipiert, dass sie ausgeführt wird, wenn die Speicherwarteschlange nicht leer ist. Es markiert die CSV-Datei als geöffnet (um den gleichzeitigen Zugriff zu verwalten), durchläuft dann jedes Produkt in der Warteschlange und serialisiert es in ein Wörterbuch (mit asdict) und in die CSV-Datei schreiben.
Die CSV-Datei wird im Anfügemodus geöffnet (a), wodurch Daten hinzugefügt werden können, ohne dass vorhandene Informationen überschrieben werden. Die Methode prüft, ob die Datei neu erstellt wurde und schreibt die Spaltenüberschriften entsprechend.
Nach dem Speichern der in der Warteschlange befindlichen Produkte wird die Warteschlange geleert und die CSV-Datei als geschlossen markiert, bereit für den nächsten Datenstapel.
Schließen der Pipeline
Es wird sichergestellt, dass keine Daten ungespeichert bleiben close_pipeline Die Methode übernimmt den endgültigen Datenfluss in die CSV-Datei.
def close_pipeline(self):
if self.storage_queue:
self.save_to_csv()
Testen der Datenpipeline
Um die Wirksamkeit unserer zu demonstrieren ProductDataPipelinesimulieren wir den Prozess des Hinzufügens mehrerer Produkte, einschließlich eines Duplikats, um zu sehen, wie unsere Pipeline die Datenbereinigung, Duplikaterkennung und CSV-Speicherung verwaltet.
data_pipeline = ProductDataPipeline(csv_filename='product_data.csv', storage_queue_limit=3)
# Sample scraped data to add to our pipeline
sample_products = (
{'name': 'Artisanal Chocolate', 'price': 'Sale price£2.99', 'url': '/artisanal-chocolate'},
{'name': 'Gourmet Cookies', 'price': '£5.50', 'url': 'http://example.com/gourmet-cookies'},
{'name': 'Artisanal Chocolate', 'price': 'Sale price£2.99', 'url': '/artisanal-chocolate'},
{"name": "Delicious Chocolate", "price": "Sale priceFrom £1.99", "url": "/delicious-chocolate"},
{"name": "Yummy Cookies", "price": "£2.50", "url": "http://example.com/yummy-cookies"},
{"name": "Tasty Candy", "price": "Sale price£0.99", "url": "/apple-pies"}
)
# Add each product to the pipeline
for product in sample_products:
data_pipeline.add_product(product)
# Ensure all remaining data in the pipeline gets saved to the CSV file
data_pipeline.close_pipeline()
print("Data processing complete. Check the product_data.csv file for output.")
Dieses Testskript initialisiert die ProductDataPipeline mit einem angegebenen CSV-Dateinamen und einem Speicherwarteschlangenlimit. Anschließend wird versucht, drei Produkte, einschließlich eines Duplikats, hinzuzufügen, um zu sehen, wie unsere Pipeline damit umgeht.
Der close_pipeline Die Methode wird zuletzt aufgerufen, um sicherzustellen, dass alle Daten in die CSV-Datei geschrieben werden. Dies demonstriert die Fähigkeit der Pipeline, Daten durchgängig zu verwalten.
Vollständiger Datenpipeline-Code
Hier ist der vollständige Code für ProductDataPipeline Klasse, die alle in diesem Artikel genannten Schritte integriert:
import csv
from dataclasses import asdict, fields
class ProductDataPipeline:
def __init__(self, csv_filename='product_data.csv', storage_queue_limit=10):
self.names_seen = set()
self.storage_queue = ()
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
if not self.storage_queue:
return
self.csv_file_open = True
with open(self.csv_filename, mode='a', newline='', encoding='utf-8') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=(field.name for field in fields(self.storage_queue(0))))
if csvfile.tell() == 0:
writer.writeheader()
for product in self.storage_queue:
writer.writerow(asdict(product))
self.storage_queue.clear()
self.csv_file_open = False
def clean_raw_product(self, scraped_data):
cleaned_data = {
"name": scraped_data.get("name", ""),
"price": scraped_data.get("price", ""),
"url": scraped_data.get("url", "")
}
return Product(**cleaned_data)
def is_duplicate(self, product_data):
if product_data.name in self.names_seen:
print(f"Duplicate item found: {product_data.name}. Item dropped.")
return True
self.names_seen.add(product_data.name)
return False
def add_product(self, scraped_data):
product = self.clean_raw_product(scraped_data)
if not self.is_duplicate(product):
self.storage_queue.append(product)
if len(self.storage_queue) >= self.storage_queue_limit:
self.save_to_csv()
def close_pipeline(self):
if self.storage_queue:
self.save_to_csv()
data_pipeline = ProductDataPipeline(csv_filename='product_data.csv', storage_queue_limit=3)
# Sample scraped data to add to our pipeline
sample_products = (
{'name': 'Artisanal Chocolate', 'price': 'Sale price£2.99', 'url': '/artisanal-chocolate'},
{'name': 'Gourmet Cookies', 'price': '£5.50', 'url': 'http://example.com/gourmet-cookies'},
{'name': 'Artisanal Chocolate', 'price': 'Sale price£2.99', 'url': '/artisanal-chocolate'},
{"name": "Delicious Chocolate", "price": "Sale priceFrom £1.99", "url": "/delicious-chocolate"},
{"name": "Yummy Cookies", "price": "£2.50", "url": "http://example.com/yummy-cookies"},
{"name": "Tasty Candy", "price": "Sale price£0.99", "url": "/apple-pies"}
)
# Add each product to the pipeline
for product in sample_products:
data_pipeline.add_product(product)
# Ensure all remaining data in the pipeline gets saved to the CSV file
data_pipeline.close_pipeline()
print("Data processing complete. Check the product_data.csv file for output.")
Lerne weiter
In diesem Artikel haben wir eine umfassende Anleitung zum Bereinigen und Strukturieren von Scraped-Daten mithilfe von Datenklassen und Datenpipelines bereitgestellt. Durch die Befolgung dieser Techniken können Sie sicherstellen, dass Ihre Scraping-Daten korrekt, konsistent und für die Analyse bereit sind.
Wenn Sie größere Datenmengen scrapen möchten, ohne blockiert zu werden, sollten Sie ScraperAPI verwenden. Es bietet eine einfache API, mit der Sie vollständig gerenderte HTML-Antworten abrufen können, einschließlich solcher von dynamischen Websites.
Bis zum nächsten Mal, viel Spaß beim Scrapen!
