Ciao a tutti. Amici, condividiamo con voi la traduzione di un articolo preparato specificamente per gli studenti del corso . Andiamo!
Apache Beam e DataFlow per pipeline in tempo reale
Il post di oggi si basa su un compito su cui ho recentemente lavorato. Ero davvero contento di realizzarlo e di descrivere il lavoro svolto in un formato da blog, poiché mi ha dato l'opportunità di impegnarmi nell'ingegneria dei dati e di creare qualcosa che sarebbe stato molto utile per il mio team. Recentemente ho scoperto che nei nostri sistemi era conservato un registro utenti piuttosto consistente, relativo a uno dei nostri prodotti per la gestione dei dati. Si è rivelato che nessuno stava utilizzando questi dati, quindi mi sono subito interessato a ciò che avremmo potuto scoprire se avessimo iniziato ad analizzarli regolarmente. Tuttavia, c'erano alcuni ostacoli lungo il cammino. Il primo problema era che i dati erano archiviati in molti file di testo diversi, che non erano accessibili per un'analisi immediata. Il secondo problema era che erano salvati in un sistema chiuso, quindi non potevo usare alcuno dei miei strumenti preferiti per l'analisi dei dati.
Dovevo decidere come semplificare l'accesso per noi e come portare un valore aggiunto integrando questa fonte di dati in alcune delle nostre soluzioni di interazione con gli utenti. Dopo aver riflettuto per un po', ho deciso di costruire una pipeline per trasferire questi dati in un database cloud, in modo che io e il mio team potessimo accedervi e iniziare a generare alcune conclusioni. Dopo aver completato la specializzazione in Data Engineering su Coursera qualche tempo fa, avevo il grande desiderio di utilizzare nel progetto alcuni strumenti del corso.
Quindi, posizionare i dati in un database cloud sembrava un modo ragionevole per risolvere il mio primo problema, ma cosa potevo fare riguardo al problema numero 2? Fortunatamente, c'era un modo per trasferire questi dati in un ambiente dove potessi accedere a strumenti come Python e Google Cloud Platform (GCP). Tuttavia, si trattava di un processo lungo, quindi avevo bisogno di fare qualcosa che mi permettesse di continuare lo sviluppo mentre aspettavo il completamento del trasferimento dei dati. La soluzione a cui sono arrivato è stata quella di creare dati fittizi utilizzando la libreria Faker in Python. Non avevo mai usato questa libreria prima d'ora, ma ho rapidamente capito quanto fosse utile. L'uso di questo approccio mi ha permesso di iniziare a scrivere codice e testare la pipeline senza dati effettivi.
Detto ciò, in questo post racconterò come ho costruito la pipeline descritta sopra, utilizzando alcune delle tecnologie disponibili in GCP. In particolare, utilizzerò Apache Beam (versione per Python), Dataflow, Pub/Sub e Big Query per raccogliere i log utente, trasformare i dati e trasferirli in un database per ulteriori analisi. Nel mio caso, avevo solo bisogno della funzionalità batch di Beam, poiché i miei dati non arrivavano in tempo reale, quindi Pub/Sub non era necessario. Tuttavia, parlerò della versione stream, poiché è quella con cui potreste avere a che fare nella pratica.
Introduzione a GCP e Apache Beam
Google Cloud Platform offre un insieme di strumenti veramente utili per l'elaborazione di grandi dati. Ecco alcuni degli strumenti che utilizzerò:
- è un servizio di messaggistica che utilizza il modello Publisher-Subscriber, che ci consente di ricevere dati in tempo reale.
- è un servizio che semplifica la creazione di pipeline per i dati e risolve automaticamente compiti come il ridimensionamento dell'infrastruttura, il che significa che possiamo concentrarci solo sulla scrittura del codice per la nostra pipeline.
- è un archivio dati cloud. Se sei familiare con altri database SQL, non dovrai spendere molto tempo a capire BigQuery.
- E infine, utilizzeremo Apache Beam, concentrandoci sulla versione Python per creare la nostra pipeline. Questo strumento ci permetterà di creare una pipeline per l'elaborazione streaming o batch, che si integra con GCP. È particolarmente utile per l'elaborazione parallela e adatta per compiti di estrazione, trasformazione e caricamento (ETL), quindi, se dobbiamo spostare dati da un luogo all'altro con trasformazioni o calcoli, Beam è una buona scelta.
C'è una vasta gamma di strumenti disponibili su GCP, quindi può essere difficile tenerne traccia e capire il loro scopo, ma ecco un riassunto per riferimento.
Su GCP è disponibile un'ampia varietà di strumenti, quindi può essere difficile coprire tutto, compresi i loro scopi, tuttavia sintesi per riferimento.
Visualizzazione della nostra pipeline
Visualizziamo i componenti della nostra pipeline in figura 1. A un livello alto, vogliamo raccogliere dati degli utenti in tempo reale, elaborarli e trasferirli in BigQuery. I log vengono creati quando gli utenti interagiscono con il prodotto, inviando richieste al server, che vengono poi registrate. Questi dati possono essere particolarmente utili per capire come gli utenti interagiscono con il nostro prodotto e se funzionano correttamente. In generale, la pipeline conterrà i seguenti passaggi:
Beam rende questo processo molto semplice, indipendentemente dal fatto che abbiamo una fonte di dati in streaming o un file CSV, e desideriamo eseguire un'elaborazione batch. Più avanti vedrete che nel codice ci sono solo modifiche minime necessarie per passare da uno all'altro. Questo è uno dei vantaggi dell'utilizzo di Beam.
Figura 1: Pipeline dati principale: Origine:
Creazione di dati fittizi con Faker
Come ho già accennato, a causa dell'accesso limitato ai dati ho deciso di creare dati fittizi nello stesso formato degli effettivi. È stato un esercizio davvero utile, poiché ho potuto scrivere codice e testare la pipeline mentre aspettavo i dati. Vi invito a dare un'occhiata a Faker, se desiderate scoprire cos'altro offre questa libreria. I nostri dati degli utenti saranno in linea generale simili all'esempio qui sotto. Sulla base di questo formato possiamo generare dati riga per riga per simulare dati in tempo reale. Questi log ci forniscono informazioni come la data, il tipo di richiesta, la risposta del server, l'indirizzo IP ecc.
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"
Basandoci sulla riga sopra, vogliamo creare la nostra variabile LINE, utilizzando 7 variabili tra parentesi graffe qui sotto. Le utilizzeremo anche come nomi delle variabili nella nostra schema di tabelle poco dopo.
LINE = """
{remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"
"""
Se avessimo eseguito un'elaborazione batch, il codice sarebbe stato molto simile, anche se avremmo dovuto creare un insieme di campioni in un certo intervallo di tempo. Per utilizzare Faker, creiamo semplicemente un oggetto e chiamiamo i metodi di cui abbiamo bisogno. In particolare, Faker è stato utile per generare indirizzi IP e siti web. Ho utilizzato i seguenti metodi:
finto.ipv4()
finto.uri_path()
finto.uri()
finto.user_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_lineFine della prima parte.
Condivideremo con voi il seguito dell'articolo nei prossimi giorni, nel frattempo, come sempre, attendiamo i vostri commenti ;-).
Fonte: habr.com
