Pozdrav svima. Dijelimo prijevod završnog dijela članka, pripremljenog posebno za studente tečaja . S prvim dijelom možete se upoznati .
Apache Beam i DataFlow za cjevovode u stvarnom vremenu
Postavke Google Clouda
Napomena: Za pokretanje cjevovoda i objavu podataka korisničkog loga koristio sam Google Cloud Shell, jer sam imao problema s pokretanjem cjevovoda na Pythonu 3. Google Cloud Shell koristi Python 2, koji je bolje usklađen s Apache Beamom.
Da bismo pokrenuli cjevovod, moramo malo istražiti postavke. Oni od vas koji prije nisu koristili GCP, moraju slijediti sljedećih 6 koraka koje možete pronaći na ovoj .
Nakon toga trebamo prenijeti naše skripte u Google Cloud skladište i kopirati ih u naš Google Cloud Shell. Prenos u Google Cloud skladište je prilično trivijalan (opis možete pronaći ). Da bismo kopirali naše datoteke, možemo otvoriti Google Cloud Shell iz alatne trake, kliknuvši na prvu ikonu s lijeve strane na slici 2 ispod.
Slika 2
Komande koje trebamo za kopiranje datoteka i instalaciju potrebnih biblioteka su navedene u nastavku.
# 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>Kreiranje naše baze podataka i tablice
Nakon što smo izvršili sve korake vezane uz postavke, sljedeće što trebamo učiniti je stvoriti skup podataka i tablicu u BigQueryju. Postoji nekoliko načina kako to učiniti, ali najjednostavnije je koristiti konzolu Google Clouda, prvo stvarajući skup podataka. Možete izvršiti radnje navedene u sljedećoj , kako biste stvorili tablicu s shemom. Naša će tablica imati 7 stupaca, koji odgovaraju komponentama svakog korisničkog loga. Radi jednostavnosti, odredit ćemo sve stupce kao nizove (tip string), osim varijable timelocal, i nazvat ćemo ih prema varijablama koje smo generirali ranije. Shema naše tablice trebala bi izgledati kao na slici 3.
Slika 3. Shema tablice
Objava podataka korisničkog loga
Pub/Sub je ključna komponenta našeg sustava, jer omogućava više neovisnim aplikacijama da komuniciraju među sobom. Konkretno, djeluje kao posrednik koji nam omogućuje slanje i primanje poruka između aplikacija. Prvo što trebamo učiniti je kreirati temu (topic). Dovoljno je otići u Pub/Sub u konzoli i pritisnuti CREATE TOPIC.
Kod u nastavku poziva naš skript za generiranje podataka o logovima, definiranim ranije, a zatim se povezuje i šalje logove u Pub/Sub. Jedino što trebamo učiniti je kreirati objekt PublisherClient, odrediti putanju do teme koristeći metodu topic_path i pozvati funkciju publish s topic_path s podacima. Imajte na umu da uvozimo generate_log_line iz našeg skripta stream_logs, stoga se pobrinite da se ti datoteke nalaze u istoj mapi, inače ćete dobiti pogrešku pri uvozu. Zatim to možemo pokrenuti putem naše Google konzole, koristeći:
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):
# Kada vrijeme trajanja nije specificirano, metoda za iznimke čeka neodređeno.
if message_future.exception(timeout=30):
print('Objava poruke na {} izazvala je iznimku {}.'.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)Kada se datoteka pokrene, moći ćemo promatrati izlaz podataka o logu na konzoli, kako je prikazano na slici u nastavku. Ovaj skript će raditi sve dok ne pritisnemo CTRL+C, kako bismo ga prekinuli.
Slika 4. Izlaz publish_logs.py
Pisanje koda našeg sustava
Sada kada smo sve pripremili, možemo preći na najzanimljiviji dio — pisanje koda našeg sustava koristeći Beam i Python. Da bismo stvorili Beam sustav, trebamo kreirati objekt sustava (p). Nakon što smo kreirali objekt sustava, možemo primijeniti nekoliko funkcija jednu za drugom koristeći operator pipe (|). Općenito, radni tok izgleda kao na slici u nastavku.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) U našem kodu ćemo stvoriti dvije korisničke funkcije. Funkciju regex_clean, koja skenira podatke i izvlači odgovarajući niz na temelju liste PATTERNS, koristeći funkciju re.search. Funkcija vraća niz odvojen zarezima. Ako niste ekspert za regularne izraze, preporučujem da se upoznate s ovim i vježbati u notepad-u kako biste testirali kod. Nakon toga definiramo korisničku ParDo funkciju pod imenom Split, koja je varijacija Beam transformacije za paralelnu obradu. U Pythonu se to radi na poseban način — moramo stvoriti klasu koja nasljeđuje klasu DoFn Beam. Funkcija Split prihvaća obrađeni niz iz prethodne funkcije i vraća listu rječnika s ključevima koji odgovaraju imenima stupaca u našoj BigQuery tablici. Ima nešto o čemu treba napomenuti u vezi ove funkcije: morao sam importirati datetime unutar funkcije, da bi radila. Dobivao sam poruku o grešci prilikom importiranja na početku datoteke, što je bilo čudno. Ova lista se zatim prosljeđuje funkciji WriteToBigQuery, koja jednostavno dodaje naše podatke u tablicu. Kod za Batch DataFlow Job i Streaming DataFlow Job je naveden u nastavku. Jedina razlika između batch i stream koda je u tome što u batch obradi čitamo CSV iz src_path, koristeći funkciju ReadFromText iz Beama.
Batch DataFlow Job (obrada paketa)
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 (obrada toka)
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()
Pokretanje cevovoda
Cevovod možemo pokrenuti na nekoliko različitih načina. Da smo htjeli, mogli bismo ga jednostavno pokrenuti lokalno iz terminala, prijavivši se na GCP.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingMeđutim, namjeravamo ga pokrenuti putem DataFlow-a. To možemo učiniti pomoću sljedeće naredbe, postavljajući sljedeće obavezne parametre.
projekt— ID vašeg GCP projekta.pokretač— pokretač cevovoda koji će analizirati vaš program i konstruirati vaš cevovod. Za izvršavanje u oblaku morate navesti DataflowRunner.staging_location— put do Cloud Dataflow skladišta za indeksiranje paketnih kodova potrebnih izvođačima koji izvršavaju zadatak.temp_location— put do Cloud Dataflow skladišta za smještaj privremenih datoteka zadataka koje se stvaraju tijekom izvršavanja cevovoda.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Dok dok izvršavanje ove naredbe, možemo prijeći na DataFlow karticu u Google konzoli i pregledati našu cijev. Kliknuvši na cijev, trebali bismo vidjeti nešto slično slici 4. U svrhu debagiranja može biti vrlo korisno otići u zapise, a zatim u Stackdriver za pregled detaljnih zapisa. To mi je pomoglo da razriješim probleme s cijevi u nekoliko slučajeva.
Slika 4: Beam cijev
Pristup našim podacima u BigQuery
Dakle, trebali bismo već imati pokrenutu cijev s podacima koji pristižu u našu tablicu. Da bismo to provjerili, možemo otići do BigQuery i pregledati podatke. Nakon korištenja dolje navedene naredbe trebali biste vidjeti prvih nekoliko redaka skupa podataka. Sada kada imamo podatke pohranjene u BigQuery, možemo izvršiti daljnje analize, dijeliti podatke s kolegama i početi odgovarati na poslovna pitanja.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Slika 5: BigQuery
Zaključak
Nadamo se da će ovaj post poslužiti kao koristan primjer za kreiranje stream cijevi podataka, kao i pronalaženje načina da učinimo podatke dostupnijima. Pohranjivanje podataka u ovom formatu donosi nam mnoge prednosti. Sada možemo početi odgovarati na važna pitanja, poput: koliko ljudi koristi naš proizvod? Raste li s vremenom baza korisnika? S kojim aspektima proizvoda ljudi najviše komuniciraju? I ima li grešaka tamo gdje ih ne bi trebalo biti? To su pitanja koja će biti zanimljiva za organizaciju. Na osnovu ideja proizašlih iz odgovora na ova pitanja, možemo unaprijediti proizvod i povećati angažman korisnika.
Beam je zaista koristan za ovu vrstu vježbi, a također ima niz drugih zanimljivih primjena. Na primjer, možete analizirati podatke o burzovnim tikovima u stvarnom vremenu i izvršavati transakcije na temelju analize, ili možda imate podatke sa senzora koji dolaze s vozila i želite izračunati razinu prometa. Također, možete biti igraća kompanija koja prikuplja podatke o korisnicima i koristi ih za izradu informativnih panela za praćenje ključnih pokazatelja. Dobro, gospodo, to je tema za drugi post, hvala na čitanju, a za one koji žele vidjeti puni kod, ispod je poveznica na moj GitHub.
To je to. .
Izvor: habr.com
