Tere kõigile. Sõbrad, jagame teiega artikli tõlket, mis on spetsiaalselt ette valmistatud kursuse õpilastele . Alustame!
Apache Beam ja DataFlow reaalajas konveierite jaoks
Tänane post põhineb ülesandel, millega ma hiljuti tööl tegelesin. Olin tõeliselt rõõmus, et sain selle ellu viia ja kirjeldada oma tehtud tööd blogipostituse vormis, kuna see andis mulle võimaluse tegeleda andmete inseneritööga ja teha midagi, mis oleks minu meeskonnale tõeliselt kasulik. Hiljuti avastasin, et meie süsteemides on piisavalt suurt kasutajalogi, mis on seotud ühe meie andmetöötlustoote kogumisega. Selgus, et keegi neid andmeid ei kasutanud, seega hakkasin kohe mõtlema, mida me võiksime teada saada, kui hakkaksime neid regulaarselt analüüsima. Kuid teel oli mitmeid probleeme. Esimene probleem oli see, et andmed olid salvestatud paljudesse erinevatesse tekstifailidesse, mis ei olnud kohese analüüsi jaoks kergesti kättesaadavad. Teine probleem seisnes selles, et need olid salvestatud suletud süsteemis, mistõttu ei saanud ma kasutada ühtegi oma lemmikandmeanalüüsi tööriista.
Mul tuli lahendada, kuidas muuta juurdepääsu meie jaoks lihtsamaks ja tuua vähemalt mingit väärtust, integreerides need andmeallikad mõnedesse meie kasutajaliidese lahendustesse. Mõnda aega järele mõeldes otsustasin luua andmete edastamise toru pilveandmebaasi, et mina ja mu meeskond saaksime neile juurde pääseda ja hakata genereerima mingisuguseid järeldusi. Pärast Courseras Data Engineering spetsialiseerumise lõpetamist olin ammuenud sooviga kasutada projektis mõningaid kursuse tööriistu.
Seega tundus pilveandmebaasi paigutamine mõistlik lahendus minu esimeses probleemis, kuid mida ma sain teha probleemiga number 2? Õnneks oli võimalus kanda need andmed keskkonda, kus sain juurde pääseda sellistele tööriistadele nagu Python ja Google Cloud Platform (GCP). Kuid see oli pikk protsess, seega pidi ma tegema midagi, mis võimaldaks mul arendust jätkata, samal ajal kui ootasin andmete edastamise lõppu. Lahendus, mille ma leidsin, oli valeandmete genereerimine, kasutades teeki Faker Pythonis. Ma ei ole kunagi varem seda teeki kasutanud, kuid mõistsin kiiresti, kui kasulik see on. Selle lähenemise kasutamine võimaldas mul hakata kirjutama koodi ja testima toru ilma tegelike andmeteta.
Nagu juba öeldud, räägin selles postituses, kuidas ma ehitasin eespool kirjeldatud toru, kasutades GCP-s saadaval olevaid tehnoloogiaid. Täpsemalt kasutan ma Apache Beam (Python versioon), Dataflow, Pub/Sub ja BigQuery kasutajate logide kogumiseks, andmete töötlemiseks ja nende edastamiseks andmebaasi edasiseks analüüsimiseks. Minu puhul vajas ainult Beam'i pakettfunktsionaalsust, kuna minu andmed ei olnud reaalajas, seega Pub/Sub ei olnud vajalik. Siiski peatun voogedastuse versioonil, kuna see on see, millega võite praktikas kokku puutuda.
Sissejuhatus GCP-sse ja Apache Beam'i
Google Cloud Platform pakub tõeliselt kasulikke tööriistu suurte andmete töötlemiseks. Siin on mõned tööriistad, mida kasutan:
- — see on sõnumiteenuse mudel, mis kasutab Publisher-Subscriber (väljastaja-iga) lähenemist ja võimaldab meil saada andmeid reaalajas.
- — see on teenus, mis lihtsustab andmevoogude loomist ja lahendab automaatselt selliseid ülesandeid nagu infrastruktuuri skaleerimine, mis tähendab, et saame keskenduda ainult oma voolu koodile.
- — see on pilveandmete salvestus. Kui olete tuttav muude SQL-põhiste andmebaasidega, ei tohiks BigQuery'ga kaua aega kuluda.
- Ja lõpuks kasutame Apache Beam'i, keskendudes täpsemalt Python'i versioonile oma voolu loomisel. See tööriist võimaldab meil luua voogu reaalajas või erinevate partiide töötlemiseks, mis integreerub GCP-ga. See on eriti kasulik paralleelseks töötlemiseks ja sobib ETL (väljatõmbamine, muutmine ja laadimine) tüüpsete ülesannete jaoks, seega, kui peame andmeid ühelt kohalt teisele liigutama koos muundamiste või arvutustega, siis Beam on hea valik.
GCP-l on saadaval lai valik tööriistu, seega võib kõiki neid jälgida olla keeruline, kuid siin on nende kokkuvõte viidatud.
GCP-s on saadaval palju tööriistu, mistõttu võib kõiki neid, sealhulgas nende otstarvet, olla keeruline katta, kuid siiski lühikokkuvõte teie jaoks.
Meie andmepipeli visualiseerimine
Visualiseerime meie andmepipeli komponente joonisel 1.Korktasemel tahame koguda kasutajaandmeid reaalajas, töödelda neid ja edastada BigQuery-sse. Logid genereeritakse, kui kasutajad suhtlevad tootega, esitades päringuid serverisse, mis siis logitakse. Need andmed võivad olla eriti kasulikud, et mõista, kuidas kasutajad meie tootega suhtlevad ja kas see töötab õigesti. Üldiselt sisaldab andmepipel järgmisi etappe:
Beam muudab selle protsessi väga lihtsaks, olenemata sellest, kas meil on voogandmete allikas või CSV-fail, ja soovime läbi viia partii töötlemise. Hiljem näete, et koodis on vaid minimaalne muudatus, mis on vajalik nende vahel vahetamiseks. See on üks Beam'i kasutamise eeliseid.
Joonis 1: Peamine andmepipel: Allikas:
Valeandmete loomine Fakeriga
Nagu juba varem mainisin, otsustasin andmete piiratud juurdepääsu tõttu luua valeandmed samas formaadis kui tegelikud. See oli tõeliselt kasulik harjutus, kuna sain kirjutada koodi ja testida torustikku, kuni andmed saabusid. Soovitan vaadata Faker, kui soovite teada, mida veel see raamatukogu pakub. Meie kasutajate andmed on üldjoontes sarnased allolevale näitele. Selle formaadi põhjal saame rida-realt genereerida andmeid reaalajas andmete simuleerimiseks. Need logid annavad meile sellist teavet nagu kuupäev, päringu tüüp, serveri vastus, IP-aadress jne.
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, nagu Gecko) Chrome/34.0.855.0 Safari/5312"
Eelneva rea põhjal soovime luua meie muutuja LINE, kasutades allpool 7 muutujaid sulgudes. Kasutame neid ka hiljem meie tabeli skeemi muutujate nimedena.
REAA = """
{remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"
"""
Kui me töötaksime pakettide töötlemise meetodil, oleks kood väga sarnane, kuigi me peaksime looma näidiste komplekti teatud ajavahemikus. Fakeri kasutamiseks loome lihtsalt objekti ja kutsume vajalikud meetodid. Eelkõige on Faker olnud kasulik IP-aadresside ja veebilehtede genereerimisel. Kasutasin järgmisi meetodeid:
vale.ipv4()
vale.uri_path()
vale.uri()
vale.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_lineEsimese osa lõpp.
Järgmiste päevade jooksul jagame teiega artikli jätku, kuid praegu ootame traditsiooniliselt teie kommentaare ;-).
Allikas: habr.com
