Hallo zusammen. Wir teilen die Übersetzung des abschließenden Teils des Artikels, der speziell für die Studierenden des Kurses vorbereitet wurde . Den ersten Teil finden Sie hier .
Apache Beam und DataFlow für Echtzeit-Pipelines
Einrichtung von Google Cloud
Hinweis: Um die Pipeline auszuführen und die Daten des benutzerdefinierten Logs zu veröffentlichen, habe ich Google Cloud Shell verwendet, da ich Probleme hatte, die Pipeline mit Python 3 auszuführen. Google Cloud Shell verwendet Python 2, was besser mit Apache Beam harmoniert.
Um die Pipeline zu starten, müssen wir einige Einstellungen durchsehen. Denjenigen von euch, die GCP noch nie verwendet haben, empfehle ich, die folgenden 6 Schritte auf dieser .
Seite auszuführen. Danach müssen wir unsere Skripte in den Google Cloud Storage hochladen und sie in unsere Google Cloud Shell kopieren. Das Hochladen in den Cloud Storage ist ziemlich einfach (eine Beschreibung finden Sie hier ). Um unsere Dateien zu kopieren, können wir Google Cloud Shell aus der Werkzeugleiste öffnen, indem wir auf das erste Symbol links im Bild 2 unten klicken.
Abbildung 2
Die Befehle, die wir zum Kopieren von Dateien und zur Installation der erforderlichen Bibliotheken benötigen, sind unten aufgeführt.
# Copy file from cloud storage
gsutil cp gs://<YOUR-BUCKET>/ * .
sudo pip install apache-beam[gcp] oauth2client==3.0.0
sudo pip install -U pip
sudo pip install Faker==1.0.2
# Environment variables
BUCKET=<YOUR-BUCKET>
PROJECT=<YOUR-PROJECT>Erstellen unserer Datenbank und Tabelle
Nachdem wir alle Schritte zur Einrichtung abgeschlossen haben, müssen wir als Nächstes in BigQuery einen Datensatz und eine Tabelle erstellen. Es gibt mehrere Möglichkeiten, dies zu tun, aber die einfachste ist die Verwendung der Google Cloud-Konsole, indem wir zunächst einen Datensatz erstellen. Sie können die folgenden Schritte ausführen, , um eine Tabelle mit dem Schema zu erstellen. Unsere Tabelle wird 7 Spaltenhaben, die den Komponenten jedes Benutzerprotokolls entsprechen. Zur Vereinfachung definieren wir alle Spalten als Strings (Datentyp string), mit Ausnahme der Variable timelocal, und benennen sie nach den Variablen, die wir zuvor erstellt haben. Das Schema unserer Tabelle sollte wie in Abbildung 3 aussehen.
Abbildung 3. Schema der Tabelle
Veröffentlichung der Benutzerdatenprotokolle
Pub/Sub ist ein kritischer Bestandteil unserer Pipeline, da es mehreren unabhängigen Anwendungen ermöglicht, miteinander zu kommunizieren. Insbesondere fungiert es als Vermittler, der es uns erlaubt, Nachrichten zwischen den Anwendungen zu senden und zu empfangen. Zuerst müssen wir ein Thema (Topic) erstellen. Dazu gehen wir einfach zu Pub/Sub in der Konsole und klicken auf THEMA ERSTELLEN.
Der folgende Code ruft unser Skript zur Generierung der oben definierten Log-Daten auf, verbindet sich und sendet die Logs an Pub/Sub. Das Einzige, was wir tun müssen, ist, ein Objekt zu erstellen. PublisherClient, den Pfad zum Thema mit der Methode topic_path anzugeben und die Funktion publish mit topic_path mit den Daten aufzurufen. Beachten Sie, dass wir generate_log_line aus unserem Skript importieren, stream_logs, daher stellen Sie sicher, dass sich diese Dateien im gleichen Ordner befinden, andernfalls erhalten Sie einen Importfehler. Dann können wir dies über unsere Google-Konsole ausführen, indem wir:
python publish.pyfrom stream_logs import generate_log_line
import logging
from google.cloud import pubsub_v1
import random
import time
PROJECT_ID="user-logs-237110"
TOPIC = "userlogs"
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(PROJECT_ID, TOPIC)
def publish(publisher, topic, message):
data = message.encode('utf-8')
return publisher.publish(topic_path, data = data)
def callback(message_future):
# Wenn kein Timeout angegeben ist, wartet die Ausnahme-Methode unbegrenzt.
if message_future.exception(timeout=30):
print('Beim Veröffentlichen der Nachricht auf {} trat eine Ausnahme {} auf.'.format(
topic_name, message_future.exception()))
else:
print(message_future.result())
if __name__ == '__main__':
while True:
line = generate_log_line()
print(line)
message_future = publish(publisher, topic_path, line)
message_future.add_done_callback(callback)
sleep_time = random.choice(range(1, 3, 1))
time.sleep(sleep_time)Sobald die Datei ausgeführt wird, können wir die Protokolldaten auf der Konsole sehen, wie im folgenden Bild gezeigt. Dieses Skript wird laufen, bis wir CTRL+C, um es zu beenden.
Abbildung 4. Ausgabe publish_logs.py
Schreiben des Codes für unsere Pipeline
Jetzt, da wir alles vorbereitet haben, können wir mit dem spannendsten Teil beginnen – dem Schreiben des Codes für unseren Pipeline unter Verwendung von Beam und Python. Um eine Beam-Pipeline zu erstellen, müssen wir ein Pipeline-Objekt (p) erstellen. Nachdem wir das Pipeline-Objekt erstellt haben, können wir mehrere Funktionen nacheinander anwenden, indem wir den Operator pipe (|). Im Großen und Ganzen sieht der Arbeitsablauf wie im Bild unten aus.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) In unserem Code werden wir zwei benutzerdefinierte Funktionen erstellen. Die Funktion regex_clean, die die Daten scannt und die entsprechende Zeile basierend auf der Liste der PATTERNS extrahiert, wobei die Funktion re.searchverwendet wird. Die Funktion gibt eine durch Kommas getrennte Zeichenfolge zurück. Wenn Sie kein Experte für reguläre Ausdrücke sind, empfehle ich, sich dieses anzusehen und in einem Notebook zu üben, um den Code zu überprüfen. Danach definieren wir die benutzerdefinierte ParDo-Funktion mit dem Namen Split, die eine Variante der Beam-Transformation für die parallele Verarbeitung ist. In Python geschieht das auf besondere Weise – wir müssen eine Klasse erstellen, die von der DoFn-Klasse von Beam erbt. Die Funktion Split nimmt den analysierten String aus der vorherigen Funktion und gibt eine Liste von Dictionaries zurück, wobei die Schlüssel den Namen der Spalten in unserer BigQuery-Tabelle entsprechen. Es gibt etwas, das ich zu dieser Funktion anmerken muss: Ich musste importieren datetime innerhalb der Funktion, damit sie funktioniert. Ich erhielt eine Fehlermeldung, als ich am Anfang der Datei importierte, was merkwürdig war. Diese Liste wird dann an die Funktion WriteToBigQuery, die einfach unsere Daten in die Tabelle hinzufügt. Der Code für den Batch DataFlow Job und den Streaming DataFlow Job ist unten angegeben. Der einzige Unterschied zwischen dem Batch- und dem Streaming-Code besteht darin, dass wir in der Batch-Verarbeitung CSV aus src_path, unter Verwendung der Funktion ReadFromText von Beam.
Batch DataFlow Job (Batch-Verarbeitung)
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import bigquery
import re
import logging
import sys
PROJECT='user-logs-237110'
schema = 'remote_addr:STRING, timelocal:STRING, request_type:STRING, status:STRING, body_bytes_sent:STRING, http_referer:STRING, http_user_agent:STRING'
src_path = "user_log_fileC.txt"
def regex_clean(data):
PATTERNS = [r'(^S+.[S+.]+S+)s',r'(?<=[).+?(?=])',
r'"(S+)s(S+)s*(S*)"',r's(d+)s',r"(?> beam.io.textio.ReadFromText(src_path)
| "adresse reinigen" >> beam.Map(regex_clean)
| 'ParseCSV' >> beam.ParDo(Split())
| 'WriteToBigQuery' >> beam.io.WriteToBigQuery('{0}:userlogs.logdata'.format(PROJECT), schema=schema,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
)
p.run()
if __name__ == '__main__':
logger = logging.getLogger().setLevel(logging.INFO)
main()Streaming DataFlow Job (Datenstromverarbeitung)
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import pubsub_v1
from google.cloud import bigquery
import apache_beam as beam
import logging
import argparse
import sys
import re
PROJECT="user-logs-237110"
schema = 'remote_addr:STRING, timelocal:STRING, request_type:STRING, status:STRING, body_bytes_sent:STRING, http_referer:STRING, http_user_agent:STRING'
TOPIC = "projects/user-logs-237110/topics/userlogs"
def regex_clean(data):
PATTERNS = [r'(^S+.[S+.]+S+)s',r'(?<=[).+?(?=])',
r'"(S+)s(S+)s*(S*)"',r's(d+)s',r"(?> beam.io.ReadFromPubSub(topic=TOPIC).with_output_types(bytes)
| "Decode" >> beam.Map(lambda x: x.decode('utf-8'))
| "Clean Data" >> beam.Map(regex_clean)
| 'ParseCSV' >> beam.ParDo(Split())
| 'WriteToBigQuery' >> beam.io.WriteToBigQuery('{0}:userlogs.logdata'.format(PROJECT), schema=schema,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
)
result = p.run()
result.wait_until_finish()
if __name__ == '__main__':
logger = logging.getLogger().setLevel(logging.INFO)
main()
Start der Pipeline
Wir können die Pipeline auf verschiedene Weise starten. Wenn wir wollten, könnten wir sie einfach lokal über das Terminal starten, indem wir uns remote in GCP einloggen.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingWir werden jedoch vorerst den DataFlow verwenden. Dazu können wir den folgenden Befehl nutzen und die erforderlichen Parameter festlegen.
project— die ID Ihres GCP-Projekts.runner— das Tool zum Starten der Pipeline, das Ihr Programm analysiert und die Pipeline erstellt. Für die Ausführung in der Cloud müssen Sie DataflowRunner angeben.staging_location— der Pfad zum Cloud-Speicher von Cloud Dataflow zur Indizierung der Code-Pakete, die für die ausführenden Worker erforderlich sind.temp_location— der Pfad zum Cloud-Speicher von Cloud Dataflow zur Ablage temporärer Dateien der Jobs, die während der Ausführung der Pipeline erstellt werden.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Während dieser Befehl ausgeführt wird, können wir zum Tab DataFlow in der Google-Konsole wechseln und unseren Pipeline ansehen. Wenn wir auf die Pipeline klicken, sollten wir etwas Ähnliches wie Abbildung 4 sehen. Zum Debuggen kann es sehr hilfreich sein, zu den Logs und dann zu Stackdriver zu wechseln, um die detaillierten Logs zu überprüfen. Das hat mir in mehreren Fällen geholfen, Probleme mit der Pipeline zu lösen.
Abbildung 4: Beam-Pipeline
Zugriff auf unsere Daten in BigQuery
Wir sollten bereits eine Pipeline mit den Daten, die in unsere Tabelle fließen, aktiv haben. Um dies zu überprüfen, können wir zu BigQuery gehen und die Daten ansehen. Nach der Ausführung des folgenden Befehls sollten Sie die ersten paar Zeilen des Datensatzes sehen. Jetzt, da wir Daten in BigQuery gespeichert haben, können wir eine weitere Analyse durchführen, die Daten mit Kollegen teilen und beginnen, geschäftliche Fragen zu beantworten.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Abbildung 5: BigQuery
Fazit
Wir hoffen, dass dieser Beitrag ein nützliches Beispiel für die Erstellung eines Datenstrom-Pipelines darstellt und Möglichkeiten aufzeigt, Daten zugänglicher zu machen. Die Speicherung von Daten in einem solchen Format bietet uns viele Vorteile. Nun können wir wichtige Fragen beantworten, wie: Wie viele Menschen nutzen unser Produkt? Wächst die Nutzerbasis im Laufe der Zeit? Mit welchen Aspekten des Produkts interagieren die Nutzer am meisten? Und gibt es Fehler, wo sie nicht sein sollten? Dies sind die Fragen, die für eine Organisation von Interesse sind. Basierend auf den Ideen, die aus den Antworten auf diese Fragen entstehen, können wir das Produkt verbessern und das Nutzerengagement steigern.
Beam ist tatsächlich nützlich für diese Art von Übungen und bietet zudem eine Reihe anderer interessanter Anwendungsmöglichkeiten. Zum Beispiel können Sie Echtzeit-Daten von Börsenticks analysieren und auf dieser Grundlage Handelsentscheidungen treffen. Möglicherweise haben Sie Sensordaten von Fahrzeugen, mit denen Sie den Verkehrsfluss berechnen möchten. Oder Sie könnten ein Spieleunternehmen sein, das Benutzerdaten sammelt, um Dashboards zur Überwachung wichtiger Leistungsindikatoren zu erstellen. Das ist jedoch ein Thema für einen anderen Beitrag. Vielen Dank fürs Lesen! Für alle, die den vollständigen Code sehen möchten, finden Sie unten den Link zu meinem GitHub.
Das ist alles. .
Quelle: habr.com
