Ahoj všichni. Sdílíme překlad závěrečné části článku, který byl připraven speciálně pro studenty kurzu . S první částí se můžete seznámit .
Apache Beam a DataFlow pro real-time toky
Nastavení Google Cloud
Poznámka: Pro spuštění toku a publikaci dat uživatelského protokolu jsem použil Google Cloud Shell, protože jsem měl problémy se spuštěním toku na Pythonu 3. Google Cloud Shell používá Python 2, který lépe spolupracuje s Apache Beam.
Abychom spustili tok, musíme se trochu podívat do nastavení. Ti z vás, kteří dříve nepoužívali GCP, musí splnit následujících 6 kroků uvedených na této .
Poté budeme muset nahrát naše skripty do cloudového úložiště Google a zkopírovat je do našeho Google Cloud Shell. Nahrávání do cloudového úložiště je dost triviální (popis najdete ). Abyste zkopírovali naše soubory, můžeme otevřít Google Cloud Shell z panelu nástrojů kliknutím na první ikonu vlevo na obrázku 2 níže.
Obrázek 2
Příkazy, které potřebujeme k zkopírování souborů a nastavení potřebných knihoven, jsou uvedeny níže.
# 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>Vytvoření naší databáze a tabulky
Jakmile splníme všechny kroky spojené s nastavením, další věcí, kterou musíme udělat, je vytvoření datové sady a tabulky v BigQuery. Existuje několik způsobů, jak to udělat, ale ten nejjednodušší je použít konzoli Google Cloud a nejprve vytvořit datovou sadu. Můžete provést kroky uvedené na následující , abyste vytvořili tabulku se schématem. Naše tabulka bude mít 7 sloupců, které odpovídají komponentám každého uživatelského protokolu. Pro snadnost definujeme všechny sloupce jako řetězce (typ string), s výjimkou proměnné timelocal, a pojmenujeme je podle proměnných, které jsme vygenerovali dříve. Schéma naší tabulky by mělo vypadat jako na obrázku 3.
Obrázek 3. Schéma tabulky
Publikování dat uživatelského protokolu
Pub/Sub je kriticky důležitou součástí našeho konvejeru, protože umožňuje více nezávislým aplikacím vzájemně si komunikovat. Především funguje jako zprostředkovatel, který nám umožňuje odesílat a přijímat zprávy mezi aplikacemi. První, co musíme udělat, je vytvořit téma (topic). Je to dost jednoduché — stačí přejít do Pub/Sub v konzoli a stisknout VYTVOŘIT TÉMA.
Následující kód vyvolává náš skript pro generování dat logu, jak bylo uvedeno výše, a poté se připojuje a odesílá logy do Pub/Sub. Jediné, co musíme udělat, je vytvořit objekt PublisherClient, určit cestu k tématu pomocí metody topic_path a zavolat funkci publish s topic_path s daty. Všimněte si, že importujeme generate_log_line z našeho skriptu stream_logs, proto se ujistěte, že tyto soubory jsou ve stejné složce, jinak dostanete chybu importu. Pak to můžeme spustit přes naši google konzoli pomocí:
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):
# Když není časový limit určen, metoda exception čeká neomezeně.
if message_future.exception(timeout=30):
print('Publikování zprávy na {} vyvolalo výjimku {}.'.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)Jakmile se soubor spustí, budeme moci vidět výstup dat logu na konzoli, jak je znázorněno na obrázku níže. Tento skript bude fungovat, dokud nepoužijeme CTRL+C, abychom ho ukončili.
Obrázek 4. Výstup publish_logs.py
Napsání kódu našeho konvejeru
Nyní, když jsme všechno připravili, můžeme se pustit do té nejzajímavější části — psaní kódu našeho konvejeru pomocí Beam a Pythonu. K vytvoření Beam konvejeru musíme vytvořit objekt konvejeru (p). Jakmile vytvoříme objekt konvejeru, můžeme použít několik funkcí jednu za druhou pomocí operátoru pipe (|). Celkově pracovní tok vypadá jako na obrázku níže.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) Ve našem kódu vytvoříme dvě uživatelské funkce. Funkci regex_clean, která prochází data a extrahuje odpovídající řetězec na základě seznamu PATTERNS pomocí funkce re.search. Funkce vrací řetězec oddělený čárkami. Pokud nejste expertem na regulární výrazy, doporučuji vám se seznámit s tímto a vyzkoušet si to v textovém editoru, abyste otestovali kód. Poté definujeme uživatelskou funkci ParDo s názvem Split, která je variantou Beam transformace pro paralelní zpracování. V Pythonu se to dělá zvláštním způsobem — musíme vytvořit třídu, která dědí od třídy DoFn Beam. Funkce Split přijímá zpracovaný řetězec z předchozí funkce a vrací seznam slovníků s klíči odpovídajícími názvům sloupců v naší tabulce BigQuery. Je dobré si uvědomit, že jsem musel importovat datetime uvnitř funkce, aby fungovala. Dostával jsem chybovou zprávu při importu na začátku souboru, což bylo podivné. Tento seznam se poté předává funkci WriteToBigQuery, která jednoduše přidává naše data do tabulky. Kód pro Batch DataFlow Job a Streaming DataFlow Job je uveden níže. Jediný rozdíl mezi dávkovým a streamovacím kódem je v tom, že v dávkovém zpracování čteme CSV z src_path, pomocí funkce ReadFromText z Beam.
Batch DataFlow Job (zpracování dávky)
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 (zpracování proudu)
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()
Spuštění kanálového zpracování
Kanál můžeme spouštět několika různými způsoby. Pokud bychom chtěli, mohli bychom ho jednoduše spustit lokálně z terminálu nebo vzdáleně přes GCP.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingBudeme ho však spouštět pomocí DataFlow. To můžeme udělat pomocí níže uvedeného příkazu, přičemž nastavíme následující povinné parametry.
projekt— ID vašeho projektu GCP.runner— nástroj pro spuštění kanálu, který analyzuje váš program a sestaví váš kanál. Pro provedení v cloudu musíte uvést DataflowRunner.staging_location— cesta k cloudovému úložišti Cloud Dataflow pro indexaci balíčků kódu nezbytných pro zpracovatele prací.temp_location— cesta k cloudovému úložišti Cloud Dataflow pro umístění dočasných souborů úloh vytvořených během běhu kanálu.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Dokud je tento příkaz prováděn, můžeme přejít na kartu DataFlow v Google konzole a zkontrolovat náš konvej. Kliknutím na konvej bychom měli vidět něco podobného obrázku 4. Pro ladění může být velmi užitečné přejít do protokolů a poté do Stackdriveru pro zobrazení podrobných protokolů. To mi pomohlo vyřešit problémy s konvejem v několika případech.
Obrázek 4: Beam konvej
Přístup k našim datům v BigQuery
Takže by nám již měl běžet konvej, do kterého vstupují data do naší tabulky. Chcete-li to zkontrolovat, můžeme přejít do BigQuery a podívat se na data. Po použití následujícího příkazu byste měli vidět prvních několik řádků datového souboru. Nyní, když máme data uložená v BigQuery, můžeme provádět další analýzy a sdílet data se svými kolegy a začít odpovídat na obchodní otázky.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Obrázek 5: BigQuery
Závěr
Doufáme, že tento příspěvek poslouží jako užitečný příklad pro vytváření streamovacího datového konveje a také pro hledání způsobů, jak učinit data přístupnějšími. Uložení dat v tomto formátu nám přináší mnoho výhod. Nyní můžeme začít odpovídat na důležité otázky, například kolik lidí používá náš produkt? Roste v průběhu času naše uživatelská základna? S jakými aspekty produktu lidé nejvíce interagují? A jsou někde chyby, kde by neměly být? To jsou otázky, které budou pro organizaci zajímavé. Na základě myšlenek vyplývajících z odpovědí na tyto otázky můžeme produkt vylepšit a zvýšit zájem uživatelů.
Beam je skutečně užitečný pro tento typ cvičení a má také řadu dalších zajímavých využití. Například můžete analyzovat data o burzovních tickách v reálném čase a provádět obchody na základě analýzy, možná máte data ze senzorů, které přicházejí z vozidel, a chcete vypočítat úroveň provozu. Můžete také být například herní společností, která shromažďuje uživatelská data a používá je k vytváření informačních panelů pro sledování klíčových ukazatelů. Dobře, dámy a pánové, tohle je téma už pro jiný příspěvek, děkuji za přečtení, a pro ty, kteří chtějí vidět celý kód, níže je odkaz na můj GitHub.
A to je všechno. .
Zdroj: habr.com
