Ciao a tutti. Condividiamo la traduzione della parte finale dell'articolo, preparato appositamente per gli studenti del corso . Puoi consultare la prima parte .
Apache Beam e DataFlow per pipeline in tempo reale
Configurazione di Google Cloud
Nota: Per eseguire la pipeline e pubblicare i log personalizzati, ho utilizzato Google Cloud Shell, poiché ho avuto problemi a eseguire la pipeline su Python 3. Google Cloud Shell utilizza Python 2, che è meglio compatibile con Apache Beam.
Per avviare la pipeline, dobbiamo immergerci nelle impostazioni. Quelli di voi che non hanno mai utilizzato GCP devono seguire questi 6 passaggi, riportati su questa .
Dopo di che, dovremo caricare i nostri script nello storage cloud di Google e copiarli nel nostro Google Cloud Shell. Il caricamento nel cloud storage è piuttosto semplice (puoi trovare la descrizione ). Per copiare i nostri file, possiamo aprire Google Cloud Shell dalla barra degli strumenti, cliccando sulla prima icona a sinistra nell'immagine 2 qui sotto.
Figura 2
I comandi di cui abbiamo bisogno per copiare i file e installare le librerie necessarie sono elencati di seguito.
# 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>Creazione del nostro database e della tabella
Dopo aver eseguito tutti i passaggi di configurazione, la prossima cosa da fare è creare un dataset e una tabella in BigQuery. Ci sono diversi modi per farlo, ma il più semplice è utilizzare la console di Google Cloud, iniziando con la creazione del dataset. Puoi eseguire i passaggi indicati di seguito , per creare una tabella con lo schema. La nostra tabella avrà 7 colonne, corrispondenti ai componenti di ciascun log personalizzato. Per comodità, definiremo tutte le colonne come stringhe (tipo string), eccetto la variabile timelocal, e le chiameremo in base alle variabili che abbiamo generato in precedenza. Lo schema della nostra tabella dovrebbe apparire come nell'immagine 3.
Figura 3. Schema della tabella
Pubblicazione dei dati del log personalizzato
Pub/Sub è un componente fondamentale della nostra pipeline, poiché consente a diverse applicazioni indipendenti di interagire tra loro. In particolare, funge da intermediario che ci permette di inviare e ricevere messaggi tra le applicazioni. La prima cosa da fare è creare un argomento (topic). È sufficiente andare in Pub/Sub nella console e fare clic su CREA ARGOMENTO.
Il codice riportato di seguito chiama il nostro script per generare i dati di log definiti in precedenza e poi si connette e invia i log a Pub/Sub. L'unica cosa che dobbiamo fare è creare un oggetto PublisherClient, specificare il percorso del tema utilizzando il metodo topic_path e chiamare la funzione publish con topic_path con i dati. Si noti che importiamo generate_log_line dal nostro script stream_logs, quindi assicurati che questi file siano nella stessa cartella, altrimenti otterrai un errore di importazione. Dopodiché, possiamo eseguire questo attraverso la nostra console Google, utilizzando:
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):
# Quando il timeout non è specificato, il metodo dell'eccezione aspetta indefinitamente.
if message_future.exception(timeout=30):
print('Pubblicare il messaggio su {} ha generato un'eccezione {}.'.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)Una volta avviato il file, saremo in grado di osservare l'output dei dati di log sulla console, come mostrato nell'immagine sottostante. Questo script continuerà a funzionare fino a quando non utilizziamo CTRL+C, per terminarlo.
Figura 4. Output publish_logs.py
Scrivere il codice del nostro pipeline
Ora che abbiamo preparato tutto, possiamo passare alla parte più interessante: scrivere il codice del nostro pipeline utilizzando Beam e Python. Per creare un pipeline Beam, dobbiamo creare un oggetto pipeline (p). Dopo aver creato l'oggetto pipeline, possiamo applicare diverse funzioni una dopo l'altra, utilizzando l'operatore pipe (|). In generale, il flusso di lavoro appare come nell'immagine sottostante.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) Nel nostro codice creeremo due funzioni personalizzate. La funzione regex_clean, che analizza i dati e estrae la stringa pertinente in base all'elenco di PATTERNS, utilizzando la funzione re.search. La funzione restituisce una stringa separata da virgole. Se non sei esperto di espressioni regolari, ti consiglio di dare un'occhiata a questo e fare pratica nel blocco note per controllare il codice. Dopo di che, definiamo una funzione ParDo personalizzata chiamata Split, che è una variazione della trasformazione Beam per l'elaborazione parallela. In Python, questo si fa in modo particolare: dobbiamo creare una classe che eredita dalla classe DoFn di Beam. La funzione Split prende una stringa analizzata dalla funzione precedente e restituisce un elenco di dizionari con chiavi corrispondenti ai nomi delle colonne nella nostra tabella BigQuery. C'è qualcosa che va notato su questa funzione: ho dovuto importare datetime all'interno della funzione, affinché funzionasse. Ricevevo un messaggio di errore quando importavo all'inizio del file, il che era strano. Questo elenco viene quindi passato alla funzione WriteToBigQuery, che aggiunge semplicemente i nostri dati alla tabella. Il codice per il Batch DataFlow Job e per il Streaming DataFlow Job è riportato qui sotto. L'unica differenza tra il codice batch e quello in streaming è che nella elaborazione batch leggiamo il CSV da src_path, utilizzando la funzione ReadFromText di Beam.
Batch DataFlow Job (elaborazione batch)
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)
| "clean address" > 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 (elaborazione in streaming)
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
PROGETTO="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(PROGETTO), 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()
Avvio della pipeline
Possiamo eseguire il pipeline in diversi modi. Se volessimo, potremmo semplicemente eseguirlo localmente dal terminale, accedendo da remoto a GCP.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingTuttavia, intendiamo eseguirlo utilizzando DataFlow. Possiamo farlo con il comando sottostante, impostando i seguenti parametri obbligatori.
progetto— ID del tuo progetto GCP.runner— strumento di esecuzione del pipeline che analizzerà il tuo programma e costruirà il tuo pipeline. Per l'esecuzione nel cloud, è necessario specificare DataflowRunner.staging_location— percorso per il Cloud Storage di Cloud Dataflow per indicizzare i pacchetti di codice necessari agli elaboratori che eseguono il lavoro.temp_location— percorso per il Cloud Storage di Cloud Dataflow per collocare i file temporanei dei job creati durante l'esecuzione del pipeline.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Mentre questo comando è in esecuzione, possiamo passare alla scheda DataFlow nella console di Google e visualizzare il nostro pipeline. Facendo clic sul pipeline, dovremmo vedere qualcosa di simile all'immagine 4. Per scopi di debug, può essere molto utile andare nei log e poi in Stackdriver per visualizzare i log dettagliati. Questo mi ha aiutato a risolvere problemi con il pipeline in diversi casi.
Immagine 4: Pipeline Beam
Accesso ai nostri dati in BigQuery
Quindi, dovremmo già avere un pipeline in esecuzione con i dati che fluiscono nella nostra tabella. Per verificarlo, possiamo andare su BigQuery e visualizzare i dati. Dopo aver usato il comando qui sotto, dovresti vedere le prime righe del set di dati. Ora che abbiamo dati memorizzati in BigQuery, possiamo effettuare ulteriori analisi e anche condividere i dati con i colleghi, iniziando a rispondere a domande aziendali.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Immagine 5: BigQuery
Conclusione
Speriamo che questo post possa servire come un utile esempio per la creazione di un pipeline di dati in streaming, oltre a trovare modi per rendere i dati più accessibili. Memorizzare i dati in questo formato ci offre molti vantaggi. Ora possiamo iniziare a rispondere a domande importanti, come quante persone utilizzano il nostro prodotto? La base utenti sta crescendo nel tempo? Con quali aspetti del prodotto interagiscono di più le persone? E ci sono errori dove non dovrebbero esserci? Queste sono le domande di interesse per l'organizzazione. Sulla base delle idee scaturite dalle risposte a queste domande, saremo in grado di migliorare il prodotto e aumentare l'engagement degli utenti.
Beam è davvero utile per questo tipo di esercizi e ha anche una serie di altri casi d'uso interessanti. Ad esempio, puoi analizzare i dati dei ticker di borsa in tempo reale e fare trading basandoti sull'analisi, magari hai dei dati dai sensori che provengono dai veicoli e vuoi calcolare il livello del traffico. Puoi anche, ad esempio, essere un'azienda di giochi che raccoglie dati sugli utenti e li utilizza per creare dashboard per monitorare i KPI. Bene, signori, questo è un argomento per un altro post, grazie per aver letto, e per coloro che vogliono vedere il codice completo, qui sotto c'è il link al mio GitHub.
Questo è tutto. .
Fonte: habr.com
