Ciao a tutti. Condividiamo la traduzione della parte finale dell'articolo, preparato appositamente per gli studenti del corso . Puoi trovare la prima parte .
Apache Beam e DataFlow per pipeline in tempo reale
Configurazione di Google Cloud
Nota: Per avviare il pipeline e pubblicare i dati del log utente, ho utilizzato Google Cloud Shell, poiché ho avuto difficoltà ad avviare il pipeline su Python 3. Google Cloud Shell utilizza Python 2, che è più compatibile con Apache Beam.
Per avviare il pipeline, dobbiamo esplorare un po' nelle impostazioni. Per coloro di voi che non hanno mai utilizzato GCP, è necessario seguire i seguenti 6 passaggi indicati su questa .
Dopo di che, dovremo caricare i nostri script nello storage cloud di Google e copiarli nel nostro Google Cloud Shell. Il caricamento nello storage cloud è piuttosto semplice (puoi trovare la descrizione ). Per copiare i nostri file, possiamo aprire Google Cloud Shell dalla barra degli strumenti, facendo clic sulla prima icona a sinistra nella figura 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 completato tutti i passaggi relativi alla configurazione, il prossimo passo è creare un dataset e una tabella in BigQuery. Ci sono diversi modi per farlo, ma il più semplice è usare la Console di Google Cloud, iniziando con la creazione del dataset. Puoi seguire le istruzioni indicate di seguito , per creare una tabella con uno schema. La nostra tabella avrà 7 colonne, corrispondenti ai componenti di ogni log utente. 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 mostrato nella figura 3.
Figura 3. Schema della tabella
Pubblicazione dei dati del log utente
Pub/Sub è un componente fondamentale del nostro pipeline, poiché permette a diverse applicazioni indipendenti di interagire tra loro. In particolare, funge da intermediario, consentendoci di inviare e ricevere messaggi tra le applicazioni. La prima cosa da fare è creare un argomento (topic). È sufficiente andare su Pub/Sub nella console e fare clic su CREATE TOPIC.
Il codice seguente chiama il nostro script per generare i dati di log definiti sopra, quindi si connette e invia i log a Pub/Sub. L'unica cosa che dobbiamo fare è creare un oggetto PublisherClient, specificare il percorso dell'argomento utilizzando il metodo topic_path e chiamare la funzione publish con topic_path e i dati. Nota che stiamo importando generate_log_line dal nostro script stream_logs, assicurati che questi file siano nella stessa cartella, altrimenti riceverai un errore di importazione. Possiamo quindi eseguirlo tramite 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('Pubblicazione 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 che il file verrà eseguito, potremo osservare l'output dei dati di log sulla console, come mostrato nell'immagine qui sotto. Questo script continuerà a funzionare fino a quando non utilizzeremo CTRL+C, per terminarlo.
Figura 4. Output publish_logs.py
Scrittura del codice del nostro pipeline
Ora che abbiamo tutto pronto, possiamo passare alla parte più interessante: scrivere il codice della nostra pipeline utilizzando Beam e Python. Per creare una pipeline Beam, dobbiamo creare un oggetto pipeline (p). Una volta 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 ed estrae la stringa corrispondente in base all'elenco di PATTERNS, utilizzando la funzione re.search. La funzione restituisce una stringa separata da virgole. Se non sei un esperto di espressioni regolari, ti consiglio di dare un'occhiata a questo e fare pratica in un notebook per verificare il codice. Dopo di ciò, definiamo una funzione ParDo personalizzata chiamata Split, che è una variazione della trasformazione Beam per l'elaborazione parallela. In Python, questo viene fatto in un 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. È importante notare qualcosa su questa funzione: ho dovuto importare datetime all'interno della funzione affinché funzionasse. Ricevevo un messaggio di errore durante l'importazione all'inizio del file, il che era strano. Questa lista viene poi passata alla funzione WriteToBigQuery, che aggiunge semplicemente i nostri dati alla tabella. Il codice per il Batch DataFlow Job e lo Streaming DataFlow Job è riportato di seguito. L'unica differenza tra il codice batch e quello in streaming è che nel trattamento 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 del flusso)
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()
Avvio del pipeline
Possiamo avviare il pipeline in diversi modi. Se volessimo, potremmo semplicemente avviarlo localmente dal terminale, accedendo da remoto a GCP.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingTuttavia, intendiamo avviarlo utilizzando DataFlow. Possiamo farlo con il comando sottostante, specificando i seguenti parametri obbligatori.
project— ID del tuo progetto GCP.runner— strumento di avvio del pipeline che analizzerà il tuo programma e costruirà il tuo pipeline. Per l'esecuzione nel cloud, devi specificare DataflowRunner.staging_location— percorso nel Cloud Storage di Cloud Dataflow per indicizzare i pacchetti di codice necessari ai lavoratori per eseguire il compito.temp_location— percorso nel Cloud Storage di Cloud Dataflow per posizionare i file temporanei dei compiti 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 il team sta eseguendo questo comando, possiamo passare alla scheda DataFlow nella console di Google e visualizzare il nostro pipeline. Cliccando 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 vedere 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 dati che arrivano nella nostra tabella. Per verificare ciò, possiamo andare su BigQuery e visualizzare i dati. Dopo aver utilizzato il comando qui sotto, dovresti vedere le prime righe del set di dati. Ora che abbiamo i dati memorizzati in BigQuery, possiamo effettuare ulteriori analisi, condividere i dati con i colleghi e iniziare a rispondere alle domande di business.
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 flusso di lavoro per i dati, così come per trovare modi per rendere i dati più accessibili. Conservare 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 cresce nel tempo? Con quali aspetti del prodotto gli utenti interagiscono di più? E ci sono errori dove non dovrebbero essercene? Queste sono domande che saranno interessanti per l'organizzazione. Sulla base delle idee che emergono dalle risposte a queste domande, saremo in grado di migliorare il prodotto e aumentare il coinvolgimento degli utenti.
Beam è davvero utile per questo tipo di esercizi, e presenta anche diversi altri casi d'uso interessanti. Ad esempio, puoi analizzare i dati delle transazioni di borsa in tempo reale e effettuare operazioni basate su queste analisi. Forse hai dati dei sensori provenienti da veicoli e desideri calcolare il livello del traffico. Potresti anche essere un'azienda di giochi che raccoglie dati sugli utenti e li utilizza per creare dashboard per monitorare i principali indicatori. Bene, amici, questa è una questione per un altro post. Grazie per aver letto, e per coloro che desiderano vedere il codice completo, qui sotto trovate il link al mio GitHub.
Questo è tutto. .
Fonte: habr.com
