Hallo zusammen. Wir teilen mit Ihnen die Übersetzung eines Artikels, der speziell für die Studenten des Kurses erstellt wurde. Legen wir los!
Apache Beam und DataFlow für Echtzeit-Pipelines
Der heutige Beitrag basiert auf einer Aufgabe, die ich kürzlich bei der Arbeit bearbeitet habe. Ich war wirklich erfreut, sie umzusetzen und die geleistete Arbeit in Form eines Blogpostings zu beschreiben, da es mir die Möglichkeit gab, mich mit Data Engineering zu beschäftigen und etwas zu schaffen, das für mein Team sehr nützlich wäre. Neulich stellte ich fest, dass in unseren Systemen eine ziemlich große Nutzungsprotokolldatei gespeichert war, die mit einem unserer Datenprodukte verbunden war. Es stellte sich heraus, dass niemand diese Daten verwendet hatte, also war ich sofort daran interessiert, was wir herausfinden könnten, wenn wir sie regelmäßig analysieren würden. Allerdings gab es mehrere 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, sodass ich keines meiner bevorzugten Tools zur Datenanalyse verwenden konnte.
Ich stand vor der Herausforderung, den Zugang für uns zu erleichtern und einen Mehrwert zu schaffen, indem ich diese Datenquelle in einige unserer Nutzerinteraktionen integrierte. Nach einigem Nachdenken entschied ich mich, eine Pipeline zur Übertragung dieser Daten in eine Cloud-Datenbank zu konstruieren, damit mein Team und ich darauf zugreifen und erste Analysen durchführen konnten. Nachdem ich vor einiger Zeit den Kurs Data Engineering bei Coursera abgeschlossen hatte, brannte ich darauf, einige der Werkzeuge aus dem Kurs in mein Projekt zu integrieren.
Die Ablage der Daten in einer Cloud-Datenbank schien also eine sinnvolle Lösung für mein erstes Problem zu sein. Aber was konnte ich hinsichtlich des Problems Nummer 2 tun? Glücklicherweise gab es eine Möglichkeit, diese Daten in eine Umgebung zu übertragen, in der ich auf Werkzeuge wie Python und Google Cloud Platform (GCP) zugreifen konnte. Dies war jedoch ein langwieriger Prozess, weshalb ich etwas tun musste, das es mir erlaubte, während der Übertragung der Daten weiterzuarbeiten. Die Lösung, die ich fand, bestand darin, gefälschte Daten mit der Bibliothek zu erstellen. Faker in Python. Ich hatte diese Bibliothek noch nie zuvor verwendet, aber ich erkannte schnell, wie nützlich sie ist. Mit diesem Ansatz konnte ich beginnen, Code zu schreiben und die Pipeline ohne echte Daten zu testen.
Angesichts des Vorhergesagten werde ich in diesem Beitrag erläutern, wie ich die oben beschriebene Pipeline aufgebaut habe, unter Verwendung einiger der Technologien, die in GCP verfügbar sind. Insbesondere werde ich verwenden Apache Beam (Python-Version), Dataflow, Pub/Sub und BigQuery um Benutzerprotokolle zu sammeln, Daten zu transformieren und sie in eine Datenbank für weitere Analysen zu übertragen. In meinem Fall benötigte ich nur die batch-gestützte Funktionalität von Beam, da meine Daten nicht in Echtzeit eingingen, weshalb Pub/Sub nicht erforderlich war. Ich werde jedoch auch die Streaming-Version behandeln, da dies etwas ist, mit dem Sie in der Praxis konfrontiert werden könnten.
Einführung in GCP und Apache Beam
Google Cloud Platform bietet eine Reihe wirklich nützlicher Werkzeuge zur Verarbeitung großer Datenmengen. Hier sind einige der Werkzeuge, die ich verwenden werde:
- — ist ein Messaging-Dienst, der das Publisher-Subscriber-Modell verwendet und es uns ermöglicht, Daten in Echtzeit zu empfangen.
- — ist ein Dienst, der die Erstellung von Datenpipelines vereinfacht und automatisch Aufgaben wie die Skalierung der Infrastruktur löst, was bedeutet, dass wir uns nur auf das Schreiben des Codes für unsere Pipeline konzentrieren können.
- — ist ein cloudbasiertes Datenspeicher. Wenn Sie mit anderen SQL-Datenbanken vertraut sind, werden Sie sich schnell in BigQuery einarbeiten können.
- Und schließlich werden wir Apache Beam verwenden, insbesondere konzentrieren wir uns auf die Python-Version zur Erstellung unserer Pipeline. Mit diesem Tool können wir eine Pipeline für die Stream- oder Batch-Verarbeitung erstellen, die sich nahtlos in GCP integriert. Es ist besonders nützlich für parallele Verarbeitung und eignet sich für Aufgaben wie Extrahieren, Transformieren und Laden (ETL). Wenn wir also Daten von einem Ort an einen anderen verschieben müssen, während wir Transformationen oder Berechnungen durchführen, ist Beam eine ausgezeichnete Wahl.
Auf GCP steht eine große Vielzahl von Tools zur Verfügung, sodass es schwierig sein kann, den Überblick über sie alle und deren Zweck zu behalten, aber hier ist eine Zusammenfassung zu Ihrer Orientierung.
In GCP gibt es eine Vielzahl von Werkzeugen, sodass es schwierig sein kann, sie alle abzudecken, einschließlich ihres Zwecks. Dennoch hier eine kurze Zusammenfassung zu Ihrer Information.
Visualisierung unseres Pipelines
Lassen Sie uns die Komponenten unserer Pipeline auf Abbildung 1visualisieren. Auf hohem Niveau möchten wir Benutzerdaten in Echtzeit sammeln, sie verarbeiten und in BigQuery übertragen. Protokolle werden erstellt, wenn Benutzer mit dem Produkt interagieren und Anfragen an den Server senden, die dann protokolliert werden. Diese Daten können besonders nützlich sein, um zu verstehen, wie Nutzer mit unserem Produkt interagieren und ob es ordnungsgemäß funktioniert. Insgesamt wird die Pipeline die folgenden Phasen enthalten:
Beam macht diesen Prozess sehr einfach, unabhängig davon, ob wir einen Streaming-Datenquelle oder eine CSV-Datei haben und batchweise verarbeiten möchten. Später werden Sie sehen, dass im Code nur minimale Änderungen erforderlich sind, um zwischen diesen beiden zu wechseln. Das ist eines der Vorteile der Verwendung von Beam.
Abbildung 1: Hauptdatenpipeline: Quelle:
Erstellung von pseudodaten mit Faker
Wie bereits erwähnt, habe ich aufgrund des eingeschränkten Zugriffs auf Daten beschlossen, Pseudodaten im gleichen Format wie die tatsächlichen zu erstellen. Das war wirklich eine nützliche Übung, da ich Code schreiben und die Pipeline testen konnte, während ich auf die Daten wartete. Schauen wir uns Faker an, wenn Sie wissen möchten, was diese Bibliothek noch zu bieten hat. Unsere Benutzerdaten werden insgesamt ähnlich wie im folgenden Beispiel sein. Basierend auf diesem Format können wir zeilenweise Daten generieren, um Echtzeitdaten zu simulieren. Diese Protokolle bieten uns Informationen wie Datum, Anfrage-Typ, Server-Antwort, IP-Adresse usw.
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 den geschweiften Klammern unten. Wir werden diese auch als Variablennamen in unserem Tabellenschema später 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, auch wenn wir eine Reihe von Samples in einem bestimmten Zeitrahmen erstellen müssten. Um Faker zu verwenden, erstellen wir einfach ein Objekt und rufen die benötigten Methoden auf. In diesem Zusammenhang war Faker besonders nützlich, um IP-Adressen und 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 kommenden Tagen werden wir das Fortsetzung des Artikels mit Ihnen teilen, und wie gewohnt freuen wir uns auf Ihre Kommentare ;-).
Quelle: habr.com
