Üdv mindenkinek. Megosztjuk a cikk záró részének fordítását, amely kifejezetten a kurzus hallgatói számára készült. Az első rész megismerhető. .
Apache Beam és DataFlow valós idejű adatfolyamokhoz
Google Cloud beállítása
Megjegyzés: A feladat végrehajtásához és a felhasználói naplóadatok közzétételéhez a Google Cloud Shell-t használtam, mivel problémáim adódtak a folyamat Python 3-ban való futtatásával. A Google Cloud Shell Python 2-t használ, amely jobban kompatibilis az Apache Beam-mel.
A folyamat elindításához egy kicsit mélyebbre kell ásnunk a beállításokban. Azok számára, akik még soha nem használták a GCP-t, el kell végezniük az alábbi 6 lépést, amelyeket ezen a .
Ezután fel kell töltenünk a szkriptjeinket a Google Cloud tárolóba, és másolnunk kell őket a Google Cloud Shell-be. A Google Cloud tárolóba való feltöltés elég egyszerű (az útmutatásokat megtalálhatják ). A fájlok másolásához megnyithatjuk a Google Cloud Shell-t az eszköztáron, kattintva az alábbi 2. ábrán látható bal oldali első ikonra.
2. ábra
Azok a parancsok, amelyekre szükségünk van a fájlok másolásához és a szükséges könyvtárak telepítéséhez, az alábbiakban találhatók.
# 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>Adatbázisunk és táblánk létrehozása
Miután végrehajtottuk az összes beállítással kapcsolatos lépést, a következő dolog, amit meg kell tennünk, hogy létrehozzunk egy adathalmazon és táblát a BigQuery-ban. Számos módja van ennek, de a legegyszerűbb az, ha a Google Cloud konzolt használjuk, először létrehozva az adathalmazt. Az alábbiak szerint hajthatjuk végre a lépéseket , hogy táblázatot hozzunk létre egy sémával. A táblázatunknak 7 oszlopa lesz, amelyek megfelelnek a felhasználói napló minden elemének. A kényelem kedvéért minden oszlopot karakterláncként (string) definiálunk, kivéve a timelocal változót, és elnevezzük őket a korábban generált változók szerint. A táblázatunk sémája a 3. ábrához hasonlóan kell kinézzen.
3. ábra. A táblázat sémája
Felhasználói naplóadatok közzététele
A Pub/Sub kritikus fontosságú összetevője a folyamatunknak, mivel lehetővé teszi, hogy több független alkalmazás kölcsönhatásba lépjen egymással. Különösen úgy működik, mint egy közvetítő, amely lehetővé teszi számunkra, hogy üzeneteket küldjünk és fogadjunk az alkalmazások között. Az első lépés, hogy létrehozzunk egy témát (topic). Elég, ha a konzolban a Pub/Sub-ra lépünk és rákattintunk a CREATE TOPIC-ra.
Az alábbi kód meghívja a szkriptünket a fent definiált naplóadatok generálásához, majd csatlakozik és elküldi a naplókat a Pub/Sub-ra. Egyetlen dolog, amit tennünk kell, az az, hogy létrehozzuk az objektumot. PublisherClient, megadva a téma útvonalát a topic_path metódussal, és meghívjuk a publish a topic_path funkciót az adatokkal. Vegye figyelembe, hogy importáljuk a generate_log_line től a szkriptünkből stream_logs, ezért győződjön meg arról, hogy ezek a fájlok ugyanabban a mappában vannak, különben importálási hibát kap. Ezután ezt a Google Konzolon keresztül futtathatjuk a következő használatával:
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):
# Ha a timeout nincs megadva, a kivétel kezelő végtelen ideig vár.
if message_future.exception(timeout=30):
print('A(z) {} üzenet közzététele kivételt dobott: {}.'.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)Miután a fájl elindul, megfigyelhetjük a naplóadatok kimenetét a konzolon, ahogy az az alábbi képen látható. Ez a szkript addig működik, amíg meg nem nyomjuk a CTRL+C, hogy leállítsuk azt.
4. ábra. Kimenet publish_logs.py
A csövünk kódjának megírása
Most, hogy mindent előkészítettünk, át tudunk térni a legérdekesebb részre — a csövünk kódjának megírására a Beam és a Python segítségével. A Beam-cső létrehozásához létre kell hoznunk egy csőobjektumot (p). Miután létrehoztuk a csőobjektumot, alkalmazhatunk pár funkciót egymás után a pipe (|). Általában a munkafolyamat az alábbiak szerint néz ki.
[Végső Output PCollection] = ([Kezdeti Input PCollection] | [Első Átalakítás]
| [Második Átalakítás]
| [Harmadik Átalakítás]) A kódunkban két egyedi funkciót hozunk létre. Az egyik funkció a regex_clean, amely beolvassa az adatokat, és kinyeri a megfelelő sort a PATTERNS lista alapján, a re.search. A funkció egy vesszővel elválasztott sort ad vissza. Ha nem vagy tapasztalt a reguláris kifejezésekben, javaslom, hogy nézd meg ezt a és gyakorolj a jegyzetfüzetben, hogy ellenőrizd a kódot. Ezután definiáljuk a ParDo egyéni funkcióját, amelynek neve Split, amely a Beam-transzformáció egy változata párhuzamos feldolgozásra. Pythonban ezt különleges módon végezzük — létre kell hoznunk egy osztályt, amely öröklődik a Beam DoFn osztályból. A Split funkció egy előző funkcióból származó, párhuzamosított sztringet fogad el, és egy szótárak listáját adja vissza, ahol a kulcsok megfelelnek a BigQuery táblázat oszlopainak neveinek. Van egy dolog, amit érdemes megjegyezni ezzel a funkcióval kapcsolatban: importálnom kellett datetime a funkción belül, hogy működjön. Hibaüzenetet kaptam az elején a fájl importálásakor, ami furcsa volt. Ezt a listát ezután átadjuk a WriteToBigQuery, amely egyszerűen hozzáadja az adatainkat a táblázathoz. Az Batch DataFlow Job és Streaming DataFlow Job kódja az alábbiakban található. Az egyetlen különbség a batch és a stream kód között az, hogy a batch feldolgozás során CSV-t olvasunk src_path, a ReadFromText funkciót használva a Beam-ből.
Batch DataFlow Job (tömegfeldolgozás)
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)
| "cím tisztítása" >> 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 (folyamfeldolgozás)
az apache_beam.options.pipeline_options import PipelineOptions
az google.cloud import pubsub_v1
az 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()
A csővezeték indítása
A csővezeték többféleképpen is elindítható. Ha szeretnénk, egyszerűen futtathatjuk a helyi terminálról, távolról belépve a GCP-be.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingAzonban a DataFlow segítségével fogjuk futtatni. Ezt az alábbi parancs segítségével tehetjük meg, a következő kötelező paraméterek beállításával.
project— A GCP projekt azonosítója.runner— A csővezeték indítóeszköze, amely elemzi a programodat, és elkészíti a csővezetéket. A felhőben történő végrehajtásért meg kell adnod a DataflowRunner-t.staging_location— A Cloud Dataflowhoz tartozó, a munkákat végző processzorok számára szükséges kódcsomagok indexelésére szolgáló útvonal az objektumtárolóban.temp_location— A Cloud Dataflowhoz tartozó, a csővezeték működése során létrehozott munkatárolásra szolgáló ideiglenes fájlokat tároló útvonal.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Miközben ez a parancs fut, átválthatunk a DataFlow fülre a Google Console-ban, és megtekinthetjük a csatornát. A csatornára kattintva valami hasonlót kell látnunk, mint a 4. ábrán. Hibaelhárítási célokból nagyon hasznos lehet áttérni a naplókra, majd a Stackdriverbe a részletes naplók megtekintéséhez. Ez több esetben is segített nekem a csatornával kapcsolatos problémák megoldásában.
4. ábra: Beam-csatorna
Hozzáférés az adatainkhoz a BigQuery-ban
Tehát már el kell indítanunk a csatornát, amely adatokat juttat a táblázatunkba. Ennek ellenőrzésére lépjünk a BigQueryra, és tekintsük meg az adatokat. Az alábbi parancs használata után látnunk kell az adatmodell első néhány sorát. Most, hogy az adatok a BigQueryban tárolva vannak, további elemzéseket végezhetünk, megoszthatjuk az adatokat a kollégákkal, és elkezdhetünk válaszolni az üzleti kérdésekre.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
5. ábra: BigQuery
Zárás
Reméljük, hogy ez a bejegyzés hasznos példaként szolgál a folyamatban lévő adatsorozat létrehozásához, valamint az adatok hozzáférhetőségének javítására. Az adatok ilyen formátumban való tárolása számos előnyt nyújt számunkra. Most már kezdhetünk válaszolni fontos kérdésekre, például hány ember használja a termékünket? Növekszik-e az idő múlásával a felhasználói bázis? Mely termékelemekkel lépnek kapcsolatba a felhasználók leggyakrabban? És előfordulnak-e hibák ott, ahol nem kellene? Ezek olyan kérdések, amelyek érdekelhetik a szervezetet. Az ezekre a kérdésekre adható ötletek alapján javíthatjuk a terméket és növelhetjük a felhasználói érdeklődést.
A Beam valóban hasznos az ilyen típusú gyakorlatokhoz, és számos más érdekes felhasználási eset is van. Például elemezheti a tőzsdei ticker adatokat valós időben, és kereskedhet az elemzés alapján. Lehet, hogy vannak érzékelő adatai, amelyek járművekből érkeznek, és szeretné kiszámítani a forgalom szintjét. Ön például lehet egy játékcég, amely felhasználói adatokat gyűjt, és ezeket információs panelek létrehozásához használja a kulcsfontosságú teljesítménymutatók nyomon követésére. Nos, uraim, ez már egy másik poszt témája, köszönöm a figyelmet, és azok számára, akik látni szeretnék a teljes kódot, itt van a link a GitHubomra.
Ezzel véget ért. .
Forrás: habr.com
