Hallo zusammen. Wir teilen die Übersetzung des letzten Teils des Artikels, der speziell für die Studierenden des Kurses vorbereitet wurde . Die erste Teil können Sie einsehen .
Apache Beam und DataFlow für Echtzeit-Pipelines
Google Cloud einrichten
Hinweis: Um die Pipeline und die Veröffentlichung der Benutzerdatenprotokolle zu starten, habe ich die Google Cloud Shell verwendet, da ich Probleme beim Ausführen der Pipeline mit Python 3 hatte. Die Google Cloud Shell verwendet Python 2, das besser mit Apache Beam kompatibel ist.
Um die Pipeline zu starten, müssen wir etwas in den Einstellungen herumschrauben. Für diejenigen unter Ihnen, die GCP noch nie verwendet haben, sind die folgenden 6 Schritte auf dieser .
Die nächsten Schritte bestehen darin, unsere Skripte in den Google Cloud-Speicher hochzuladen und sie in unser Google Cloud Shell zu kopieren. Das Hochladen in den Cloud-Speicher ist ziemlich trivial (eine Beschreibung finden Sie ). Um unsere Dateien zu kopieren, können wir die Google Cloud Shell über die Werkzeugleiste öffnen, indem wir auf das erste Symbol links in Abbildung 2 unten klicken.
Abbildung 2
Die Befehle, die wir zum Kopieren von Dateien und Installieren 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>Erstellung unserer Datenbank und Tabelle
Nachdem wir alle Schritte zur Einrichtung abgeschlossen haben, müssen wir als Nächstes einen Datensatz und eine Tabelle in BigQuery erstellen. Es gibt mehrere Möglichkeiten, dies zu tun, aber der einfachste Weg ist die Verwendung der Google Cloud-Konsole, indem Sie zunächst einen Datensatz erstellen. Sie können die Schritte, die Sie ausführen müssen, im Folgenden finden , um eine Tabelle mit einem Schema zu erstellen. Unsere Tabelle wird 7 Spalten, entsprechend den Komponenten jedes Benutzerprotokolls haben. Der Einfachheit halber definieren wir alle Spalten als Strings (Typ string), mit Ausnahme der Variablen timelocal, und benennen sie entsprechend den Variablen, die wir zuvor generiert haben. Das Schema unserer Tabelle sollte wie in Abbildung 3 aussehen.
Abbildung 3. Schema der Tabelle
Veröffentlichung der Benutzerdatenprotokolle
Pub/Sub ist ein entscheidender Bestandteil unserer Pipeline, da es mehreren unabhängigen Anwendungen ermöglicht, miteinander zu interagieren. Es fungiert insbesondere als Vermittler, der es uns ermöglicht, Nachrichten zwischen Anwendungen zu senden und zu empfangen. Das erste, was wir tun müssen, ist, ein Thema (topic) zu erstellen. Dazu müssen wir lediglich in der Konsole zu Pub/Sub gehen und auf CREATE TOPIC klicken.
Der nachfolgende Code ruft unser Skript zur Generierung von Logdaten auf, wie oben definiert, und verbindet sich dann 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 zu definieren topic_path und die Funktion aufzurufen publish c topic_path mit den Daten. Beachten Sie, dass wir generate_log_line aus unserem Skript importieren stream_logs, stellen Sie also sicher, dass diese Dateien im selben Ordner sind, andernfalls erhalten Sie einen Importfehler. Dann können wir das ü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('Das Veröffentlichen der Nachricht auf {} hat eine Ausnahme {} ausgelöst.'.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 läuft, können wir die Protokolldatenausgabe auf der Konsole beobachten, wie im unteren Bild gezeigt. Dieses Skript läuft, bis wir CTRL+C, um es zu beenden.
Abbildung 4. Ausgabe publish_logs.py
Schreiben des Codes für unsere Pipeline
Jetzt, wo wir alles vorbereitet haben, können wir mit dem spannendsten Teil beginnen — dem Schreiben des Codes für unsere 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 (|)verwenden. Insgesamt sieht der Arbeitsablauf wie im Bild unten aus.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) In unserem Code erstellen wir zwei benutzerdefinierte Funktionen. Die Funktion regex_clean, die Daten scannt und die entsprechende Zeichenfolge basierend auf der Liste von PATTERN extrahiert, unter Verwendung der Funktion re.search. Die Funktion gibt eine durch Kommas getrennte Zeichenfolge zurück. Wenn Sie kein Experte für reguläre Ausdrücke sind, empfehle ich Ihnen, sich mit diesem vertraut zu machen und in einem Notizbuch zu üben, um den Code zu testen. Danach definieren wir eine benutzerdefinierte ParDo-Funktion mit dem Namen Split, die eine Variante der Beam-Transformation für die parallele Verarbeitung ist. In Python wird dies auf besondere Weise durchgeführt — wir müssen eine Klasse erstellen, die von der DoFn-Beam-Klasse erbt. Die Funktion Split nimmt die geparste Zeichenfolge der vorherigen Funktion und gibt eine Liste von Dictionaries mit Schlüsseln zurück, die den Spaltennamen in unserer BigQuery-Tabelle entsprechen. Es gibt noch etwas, das über diese Funktion angemerkt werden sollte: Ich musste datetime innerhalb der Funktion importieren, damit sie funktioniert. Ich erhielt eine Fehlermeldung, als ich zu Beginn der Datei importierte, was seltsam war. Diese Liste wird dann an die Funktion WriteToBigQuery, die einfach unsere Daten in die Tabelle hinzufügt, übergeben. Der Code für Batch DataFlow Job und Streaming DataFlow Job ist unten aufgeführt. Der einzige Unterschied zwischen dem Batch- und dem Streaming-Code besteht darin, dass wir bei der Batchverarbeitung die CSV von src_path, unter Verwendung der Funktion ReadFromText aus Beam lesen.
Batch DataFlow Job (Batchverarbeitung)
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)
| "sauber Adresse" >> 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
PROJEKT="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'
THEMA = "projects/user-logs-237110/topics/userlogs"
def regex_clean(data):
MUSTER = [r'(^S+.[S+.]+S+)s',r'(?<=[).+?(?=])',
r'"(S+)s(S+)s*(S*)"',r's(d+)s',r"(?> beam.io.ReadFromPubSub(topic=THEMA).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(PROJEKT), 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()
Pipeline starten
Wir können die Pipeline auf verschiedene Arten starten. Wenn wir wollten, könnten wir sie einfach lokal über das Terminal ausführen, indem wir uns remote in GCP einloggen.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingWir werden sie jedoch über DataFlow starten. Das können wir mit dem folgenden Befehl tun, indem wir die folgenden erforderlichen Parameter angeben.
projekt— Ihre GCP-Projekt-ID.runner— Das Ausführungswerkzeug für die Pipeline, das Ihr Programm analysiert und Ihre Pipeline konstruiert. Für die Ausführung in der Cloud müssen Sie DataflowRunner angeben.staging_location— Der Pfad zum Cloud-Speicher von Cloud Dataflow für die Indizierung der erforderlichen Codepakete für die ausgeführten Aufgaben.temp_location— Der Pfad zum Cloud-Speicher von Cloud Dataflow zum Speichern temporärer Dateien von Aufgaben, 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 dieses Befehl ausgeführt wird, können wir zum Tab DataFlow in der Google-Konsole wechseln und unseren Pipeline einsehen. Wenn wir auf die Pipeline klicken, sollten wir etwas sehen, das ähnlich wie Abbildung 4 aussieht. Zur Fehlersuche kann es sehr nützlich sein, in die Logs zu wechseln und dann zu Stackdriver, um detaillierte Logs anzusehen. Dies hat mir in mehreren Fällen geholfen, Probleme mit der Pipeline zu lösen.
Abbildung 4: Beam-Pipeline
Zugriff auf unsere Daten in BigQuery
Also, wir sollten bereits eine Pipeline laufen haben, die Daten in unsere Tabelle speist. Um dies zu überprüfen, können wir zu BigQuery gehen und die Daten einsehen. Nach Ausführung des folgenden Befehls sollten Sie die ersten paar Zeilen des Datensatzes sehen. Jetzt, da unsere Daten in BigQuery gespeichert sind, können wir weitere Analysen durchführen sowie die Daten mit Kollegen teilen und anfangen, auf Geschäftsfragen zu antworten.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Abbildung 5: BigQuery
Fazit
Wir hoffen, dass dieser Beitrag als hilfreiches Beispiel für die Erstellung eines Streaming-Datenpipelines dient und auch Möglichkeiten aufzeigt, Daten zugänglicher zu machen. Die Speicherung von Daten in diesem Format bietet uns viele Vorteile. Jetzt können wir beginnen, auf wichtige Fragen zu antworten, wie zum Beispiel: 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 Fragen, die für die Organisation von Interesse sein werden. Basierend auf den Erkenntnissen aus den Antworten auf diese Fragen können wir das Produkt verbessern und das Engagement der Nutzer steigern.
Beam ist wirklich nützlich für diese Art von Übungen und hat auch eine Reihe anderer interessanter Anwendungsmöglichkeiten. Zum Beispiel können Sie in Echtzeit Börsendaten analysieren und auf der Grundlage dieser Analysen Handelsgeschäfte abschließen. Vielleicht haben Sie Sensordaten von Fahrzeugen und möchten das Verkehrsaufkommen berechnen. Sie können auch ein Gaming-Unternehmen sein, das Benutzerdaten sammelt und diese zur Erstellung von Dashboards zur Überwachung wichtiger Leistungskennzahlen verwendet. Gut, meine Damen und Herren, das ist ein Thema für einen anderen Beitrag. Vielen Dank fürs Lesen, und für diejenigen, die den vollständigen Code sehen möchten, finden Sie weiter unten den Link zu meinem GitHub.
Das ist alles. .
Quelle: habr.com
