Hallo zusammen. Freunde, wir teilen mit euch die Übersetzung eines Artikels, der speziell für die Studenten des Kurses vorbereitet wurde . Lassen Sie uns beginnen!
Apache Beam und DataFlow für Echtzeit-Pipelines
Der heutige Beitrag basiert auf einer Aufgabe, die ich kürzlich bei der Arbeit gehabt habe. Ich war wirklich begeistert, sie umzusetzen und die geleistete Arbeit in Form eines Blogbeitrags zu beschreiben, da mir dies die Möglichkeit gab, mich mit Data Engineering zu beschäftigen und gleichzeitig etwas zu tun, das für mein Team sehr nützlich wäre. Vor nicht allzu langer Zeit stellte ich fest, dass in unseren Systemen genügend große Benutzerprotokolle gespeichert waren, die mit einem unserer Produkte zur Datenverarbeitung zusammenhingen. Es stellte sich heraus, dass niemand diese Daten nutzte, also war ich sofort daran interessiert, was wir lernen könnten, wenn wir anfangen würden, sie regelmäßig zu analysieren. Allerdings gab es einige Probleme auf dem Weg. Das erste Problem war, dass die Daten in vielen verschiedenen Textdateien gespeichert waren, die nicht für eine sofortige Analyse zugänglich waren. Das zweite Problem war, dass sie in einem geschlossenen System gespeichert waren, weshalb ich keines meiner Lieblingswerkzeuge zur Datenanalyse verwenden konnte.
Ich musste entscheiden, wie wir den Zugang einfacher gestalten und wenigstens einen gewissen Wert schaffen konnten, indem ich diese Datenquelle in einige unserer Benutzerinteraktionslösungen integrierte. Nachdem ich eine Weile nachgedacht hatte, beschloss ich, eine Pipeline zu bauen, um diese Daten in eine Cloud-Datenbank zu übertragen, damit mein Team und ich darauf zugreifen und beginnen konnten, einige Erkenntnisse zu gewinnen. Nachdem ich vor einiger Zeit den Data Engineering-Kurs bei Coursera abgeschlossen hatte, brannte ich darauf, einige der Werkzeuge aus dem Kurs in meinem Projekt zu verwenden.
Das Unterbringen der Daten in einer Cloud-Datenbank schien also eine sinnvolle Lösung für mein erstes Problem zu sein, aber was konnte ich bezüglich Problem Nummer 2 tun? Glücklicherweise gab es einen Weg, diese Daten in eine Umgebung zu übertragen, wo ich auf Werkzeuge wie Python und Google Cloud Platform (GCP) zugreifen konnte. Das war jedoch ein langwieriger Prozess, also musste ich etwas tun, das es mir erlaubte, mit der Entwicklung fortzufahren, während ich auf den Abschluss der Datenübertragung wartete. Die Lösung, auf die ich kam, bestand darin, gefälschte Daten mit der Bibliothek Faker in Python. Ich habe diese Bibliothek zuvor nie verwendet, aber schnell erkannt, wie nützlich sie ist. Die Verwendung dieses Ansatzes ermöglichte es mir, mit dem Schreiben von Code und dem Testen der Pipeline ohne tatsächliche Daten zu beginnen.
In Anbetracht des bereits Gesagten werde ich in diesem Beitrag erläutern, wie ich die oben beschriebene Pipeline erstellt habe, unter Verwendung einiger der in GCP verfügbaren Technologien. Insbesondere werde ich Apache Beam (Python-Version), Dataflow, Pub/Sub und BigQuery verwenden, um Benutzerprotokolle zu sammeln, Daten zu transformieren und sie in eine Datenbank zur weiteren Analyse zu übermitteln. In meinem Fall brauchte ich nur die Batch-Funktionalität von Beam, da meine Daten nicht in Echtzeit eingingen, weshalb Pub/Sub nicht erforderlich war. Ich werde jedoch auf die Streaming-Version eingehen, da dies das ist, womit Sie in der Praxis konfrontiert werden könnten.
Einführung in GCP und Apache Beam
Die Google Cloud Platform bietet eine Reihe wirklich nützlicher Werkzeuge zur Verarbeitung großer Datenmengen. Hier sind einige der Tools, die ich verwenden werde:
- ist ein Messaging-Dienst, der das Publisher-Subscriber-Muster verwendet, das es uns ermöglicht, Daten in Echtzeit zu empfangen.
- ist ein Service, der die Erstellung von Datenpipelines vereinfacht und automatisch Aufgaben wie die Skalierung der Infrastruktur löst, was bedeutet, dass wir uns nur darauf konzentrieren können, Code für unsere Pipeline zu schreiben.
- ist ein Cloud-Datenspeicher. Wenn Sie mit anderen SQL-Datenbanken vertraut sind, werden Sie sich bei BigQuery nicht lange einarbeiten müssen.
- Und schließlich werden wir Apache Beam verwenden, und zwar konzentrieren wir uns auf die Python-Version, um unsere Pipeline zu erstellen. Dieses Tool ermöglicht es uns, eine Pipeline für die Streaming- oder Batch-Verarbeitung zu erstellen, die sich in GCP integriert. Es ist besonders nützlich für parallele Verarbeitung und eignet sich für Aufgaben wie Extraktion, Transformation und Laden (ETL). Wenn wir also Daten von einem Ort zum anderen bewegen müssen, während wir Transformationen oder Berechnungen durchführen, ist Beam eine gute Wahl.
Es gibt eine Vielzahl von tools, die in GCP verfügbar sind, daher kann es schwierig sein, sie alle im Blick zu behalten und ihren Zweck zu erkennen. Hier ist jedoch eine Zusammenfassung für Ihre Referenz.
In GCP gibt es eine große Anzahl von Werkzeugen, sodass es schwierig sein kann, alle abzudecken, einschließlich ihrer Funktionen, dennoch Kurze Zusammenfassung zur Information.
Visualisierung unserer Pipeline
Lassen Sie uns die Komponenten unserer Pipeline auf Abbildung 1. Auf hoher Ebene möchten wir Nutzerdaten in Echtzeit erfassen, sie verarbeiten und an BigQuery übermitteln. Protokolle werden erstellt, wenn Benutzer mit dem Produkt interagieren, indem sie Anfragen an den Server senden, die dann protokolliert werden. Diese Daten können besonders nützlich sein, um zu verstehen, wie Benutzer mit unserem Produkt interagieren und ob es korrekt funktioniert. Insgesamt wird die Pipeline folgende Schritte enthalten:
Beam macht diesen Prozess sehr einfach, unabhängig davon, ob wir eine Streaming-Datenquelle oder eine CSV-Datei haben und eine Batch-Verarbeitung durchführen möchten. Später werden Sie sehen, dass im Code nur minimale Änderungen erforderlich sind, um zwischen ihnen umzuschalten. Dies ist einer der Vorteile von Beam.
Abbildung 1: Hauptdatenpipeline: Quelle:
Erzeugung von Pseudodaten mit Faker
Wie ich bereits erwähnt habe, habe ich aufgrund des eingeschränkten Zugriffs auf Daten beschlossen, Pseudodaten im gleichen Format wie die tatsächlichen zu erstellen. Dies war eine wirklich nützliche Übung, da ich den Code schreiben und die Pipeline testen konnte, während ich auf Daten wartete. Werfen wir einen Blick auf Faker, wenn Sie erfahren möchten, was diese Bibliothek noch zu bieten hat. Unsere Benutzerdaten werden insgesamt ähnlich wie das folgende Beispiel sein. Basierend auf diesem Format können wir zeilenweise Daten generieren, um Daten in Echtzeit zu simulieren. Diese Protokolle geben uns Informationen wie Datum, Anfrage-Art, Serverantwort, IP-Adresse usw. an.
192.52.197.161 - - [30/Apr/2019:21:11:42] "PUT /tag/category/tag HTTP/1.1" [401] 155 "https://harris-lopez.com/categories/about/" "Mozilla/5.0 (Macintosh; PPC Mac OS X 10_11_2) AppleWebKit/5312 (KHTML, like Gecko) Chrome/34.0.855.0 Safari/5312"
Basierend auf der obigen Zeile möchten wir unsere Variable LINE, unter Verwendung von 7 Variablen in geschweiften Klammern unten. Wir werden sie auch später als Variablennamen in unserem Tabellenschema verwenden.
ZEILE = """
{remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"
"""
Wenn wir eine Batch-Verarbeitung durchführen würden, wäre der Code sehr ähnlich, obwohl wir eine Reihe von Proben in einem bestimmten Zeitrahmen erstellen müssten. Um Faker zu verwenden, erstellen wir einfach ein Objekt und rufen die benötigten Methoden auf. Insbesondere war Faker nützlich, um IP-Adressen sowie Webseiten zu generieren. Ich habe die folgenden Methoden verwendet:
falsche.ipv4()
falsche.uri_pfad()
falsche.uri()
falsche.benutzer_agent()
from faker import Faker
import time
import random
import os
import numpy as np
from datetime import datetime, timedelta
LINE = """
{remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"
"""
def generate_log_line():
fake = Faker()
now = datetime.now()
remote_addr = fake.ipv4()
time_local = now.strftime('%d/%b/%Y:%H:%M:%S')
request_type = random.choice(["GET", "POST", "PUT"])
request_path = "\/" + fake.uri_path()
status = np.random.choice([200, 401, 404], p = [0.9, 0.05, 0.05])
body_bytes_sent = random.choice(range(5, 1000, 1))
http_referer = fake.uri()
http_user_agent = fake.user_agent()
log_line = LINE.format(
remote_addr=remote_addr,
time_local=time_local,
request_type=request_type,
request_path=request_path,
status=status,
body_bytes_sent=body_bytes_sent,
http_referer=http_referer,
http_user_agent=http_user_agent
)
return log_lineEnde des ersten Teils.
In den nächsten Tagen werden wir die Fortsetzung des Artikels mit Ihnen teilen, und wie gewohnt freuen wir uns über Ihre Kommentare ;-).
Quelle: habr.com
