Loome andmevoogude töötlemise toru. Osa 1

Tere, sõbrad. Jagame teiega artikli tõlget, mis on spetsiaalselt ette valmistatud kursuse tudengitele „Andmeinsener“. Alustame!

Apache Beam ja DataFlow reaalaja töötluse jaoks

Tänane postitus põhineb ülesandel, millega ma hiljuti tööl tegelesin. Olin tõeliselt elevil, et sai selle ellu viia ja oma tööprotsessi kirja panna blogipostituse vormis, kuna see andis mulle võimaluse tegeleda andmeinseneeria valdkonnaga ning luua midagi, mis oleks minu meeskonnale tõeliselt kasulik. Hiljuti avastasin, et meie süsteemides hoitakse üsna suurt kasutajate logi, mis on seotud ühe meie andmetöötlustootega. Selgus, et neid andmeid keegi ei kasutanud, seega huvitas mind kohe see, mida me võiksime õppida, kui hakkaksime neid regulaarselt analüüsima. Kuid teel olid mitmed probleemid. Esimene probleem oli see, et andmed olid salvestatud paljudesse erinevatesse tekstifailidesse, mis ei olnud kohe analüüsitavad. Teine probleem seisnes selles, et need olid salvestatud suletud süsteemis, mistõttu ei saanud ma kasutada ühtegi oma lemmikanalüüsi tööriista.

Pidin välja mõtlema, kuidas meie juurdepääsu lihtsamaks muuta ja mingit väärtust lisada, integreerides selle andmeallika mõnelle meie kasutajate suhtlemise lahendusele. Pärast mõningast mõtlemist otsustasin ehitada toru, et edastada need andmed pilveandmebaasi, et mina ja mu meeskond saaksime neile juurde pääseda ja hakata mingisuguseid järeldusi genereerima. Pärast seda, kui lõpetasin Data Engineeringi erialakursuse Courseras mõni aeg tagasi, olin põnevil, et saaksin projektis kasutada kursuse tööriistu.

Seega tundus andmete paigutamine pilveandmebaasi mõistlik viis minu esimese probleemi lahendamiseks, kuid mida ma saaksin teha teise probleemi osas? Õnneks oli võimalus neid andmeid viia keskkonda, kus ma saaksin kasutada selliseid tööriistu nagu Python ja Google Cloud Platform (GCP). Kuid see oli pikk protsess, seega pidin midagi välja mõtlema, mis lubaks mul edasi töötada, kuni andmete edastamine lõppes. Lahendus, millega ma välja tulin, oli valeandmete loomine raamatukogu abil Faker Pythonis. Ma pole varem selle raamatukoguga töötanud, kuid mõistsin kiiresti, kui kasulik see on. Selle lähenemise kasutamine võimaldas mul alustada koodi kirjutamist ja katsetada juhtmehhanismi ilma tegelike andmeteta.

Arvestades juba öeldut, räägin selles postituses, kuidas ehitasin eelnevalt kirjeldatud juhtmehhanismi, kasutades mõningaid GCP-s saadaval olevaid tehnoloogiaid. Eelkõige kasutan Apache Beam (Python versioon), Dataflow, Pub/Sub ja BigQuery kasutajate logide kogumiseks, andmete töötlemiseks ja nende edastamiseks andmebaasi edasiseks analüüsiks. Minu puhul vajas ainult pakettide funktsionaalsust Beam, kuna minu andmed ei tulnud reaalajas, seega Pub/Sub ei olnud vajalik. Siiski peatun voogedastuse versioonil, kuna see on see, millega võiksite praktikas kokku puutuda.

Sissejuhatus GCP-sse ja Apache Beam'i

Google Cloud Platform pakub suurepäraseid tööriistu suurte andmete töötlemiseks. Siin on mõned tööriistad, mida kasutan:

• Pub/Sub — see on sõnumite vahetamise teenus, mis kasutab väljaandja ja tellija (Publisher-Subscriber) mustrit, mis võimaldab meil saada andmeid reaalajas.
• DataFlow — see on teenus, mis lihtsustab andmejuhtmete loomist ja lahendab automaatselt selliseid ülesandeid nagu infrastruktuuri skaalamine, mis tähendab, et saame keskenduda ainult koodi kirjutamisele meie juhtmehhanismiga.
• BigQuery — see on pilveteenuste andmehoidla. Kui olete tuttav teiste SQL-baasidega, ei võta BigQuerys kaua harjumine.
• Ja lõpuks, kasutame Apache Beam'i, nimelt keskendume Python versioonile meie juhtmehhanismi loomiseks. See tööriist võimaldab meil luua juhtmehhanisme voog- või pakettide töötlemiseks, mis integreerub GCP-ga. See on eriti kasulik paralleelseks töötlemiseks ja sobib ETL (ekstraktsioon, transformatsioon ja laadimine) ülesannete jaoks, seega, kui peame andmeid ühest kohast teise liigutama koos transformatsioonide või arvutustega, on Beam hea valik.

GCP-s on saadaval suur hulk tööriistu, nii et nende jälgimine ja eri otstarvete tõlgendamine võib olla keeruline, kuid siin on nende kokkuvõte referentsiks.
GCP-s on saadaval suur hulk tööriistu, seega võib nende kõigi katmine sealhulgas nende määratus olla keeruline, kuid siiski. siin on lühikokkuvõte viitamiseks.

Meie toruvisualiseerimine

Kujundame meie toru komponente joonisel 1. Üksikasjalikult soovime koguda reaalajas kasutajateavet, töödelda see ja edastada BigQuery'sse. Logid genereeritakse, kui kasutajad suhelda tootega, saates päringud serverisse, mis seejärel logitakse. Need andmed võivad olla eriti kasulikud, et mõista, kuidas kasutajad meie tootega suhtlevad ja kas see töötab õigesti. Üldiselt sisaldab meil toru järgmisi etappe:

Beam muudab selle protsessi väga lihtsaks, sõltumata sellest, kas meil on voogedastusandmete allikas või CSV-fail, ja tahame teostada partiitöötlust. Hiljem näete, et koodis on vaid minimaalne muudatus, mis on vajalik nende vahel ülemiseks. See on üks Beam'i kasutamise eeliseid.

Joonis 1: Peamine andmestiku toru: Allikas:

Pseudandmete loomine Fakeriga

Nagu ma varem mainisin, otsustasin piiratud andmeedastuse tõttu luua pseudandmeid samas vormingus nagu tegelikud. See oli tõeliselt kasulik harjutus, kuna sain kirjutada koodi ja testida toru samal ajal, kui ootasin andmeid. Vaatame lähemalt dokumentatsioon Fakerit, kui soovite näha, mida see teek veel pakub. Meie kasutajate andmed sarnanevad üldiselt alloleva näitega. Selle vormingu põhjal saame ridade kaupa genereerida andmeid, et simuleerida reaalajas andmeid. 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"

Ükskõik, mis põhjal me loome meie muutuja LINE, kasutades allpool 7 muutujaid, millel on klesides. Kasutame neid samuti muutuja nimedena meie tabeli skeemis hiljem.

LINE = """
{remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"
"""

Kui me teeksime partiitöötlust, oleks kood väga sarnane, kuigi me peaksime looma proovide komplekti teatud ajavahemikus. Faker'i kasutamiseks loome lihtsalt objekti ja kutsume vajalikud meetodid. Eriti kasulik oli Faker IP-aadresside ja veebisaitide genereerimisel. Kasutasin järgmisi meetodeid:

fake.ipv4()
fake.uri_path()
fake.uri()
fake.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_line

Esimese osa lõpp.

Järgmiste päevade jooksul jagame teiega artikli jätku, kuid praegu ootame traditsiooniliselt teie kommentaare ;).

Allikas: habr.com