Pozdravljeni vsi. Delimo prevod zaključnega dela članka, pripravljenega posebej za študente tečaja . Z prvo delom se lahko seznanite .
Apache Beam in DataFlow za cevi v realnem času
Nastavitev Google Clouda
Opomba: Za zagon cevovoda in objavo podatkov dnevnika uporabnika sem uporabil Google Cloud Shell, ker sem imel težave pri zagonu cevovoda v Pythonu 3. Google Cloud Shell uporablja Python 2, ki se bolje ujema z Apache Beam.
Da bi zagnali cevovod, moramo malo raziskati nastavitve. Tisti izmed vas, ki še niste uporabljali GCP, morate izvesti naslednjih 6 korakov, ki so navedeni na tej .
Nato bomo morali naložiti naše skripte v Google Cloud Storage in jih kopirati v naš Google Cloud Shell. Nalaganje v Google Cloud Storage je dovolj preprosto (navodila najdete ). Da bi kopirali naše datoteke, lahko odprete Google Cloud Shell iz orodne vrstice, tako da kliknete na prvo ikono levo na sliki 2 spodaj.
Slika 2
Ukazi, ki jih potrebujemo za kopiranje datotek in namestitev potrebnih knjižnic, so navedeni spodaj.
# 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>Ustvarjanje naše baze podatkov in tabele
Ko smo izvedli vse korake, povezane z nastavitvijo, je naslednje, kar moramo storiti, ustvariti nabor podatkov in tabelo v BigQuery. Obstaja več načinov, kako to narediti, a najpreprostejši je, da uporabite konzolo Google Clouda in najprej ustvarite nabor podatkov. Izvedete lahko dejanja, navedena na naslednji , da ustvarite tabelo s shemo. Naša tabela bo imela 7 stolpcev, ki ustrezajo komponentam vsakega uporabniškega dnevnika. Za lažjo uporabo bomo vse stolpce določili kot niz (tip string), razen spremenljivke timelocal, in jih poimenovali v skladu s spremenljivkami, ki smo jih ustvarili prej. Shema naše tabele bi morala izgledati kot na sliki 3.
Slika 3. Shema tabele
Objava podatkov uporabniškega dnevnika
Pub/Sub je kritična komponenta našega cevovoda, saj omogoča več neodvisnim aplikacijam, da medsebojno sodelujejo. Zlasti deluje kot posrednik, ki nam omogoča pošiljanje in prejemanje sporočil med aplikacijami. Prvo, kar moramo storiti, je ustvariti temo (topic). Dovolj je, da preidemo v Pub/Sub v konzoli in kliknemo CREATE TOPIC.
Spodaj navedena koda pokliče našo skripto za generiranje podatkov o dnevniku, kot je bilo opisano zgoraj, in se nato poveže ter pošlje dnevnike v Pub/Sub. Edina stvar, ki jo moramo narediti, je ustvariti objekt PublisherClient, določiti pot do teme z metodo topic_path in poklicati funkcijo publish z topic_path s podatki. Upoštevajte, da uvozimo generate_log_line iz naše skripte stream_logs, zato se prepričajte, da so ti datoteki v isti mapi, sicer boste dobili napako pri uvozu. Potem lahko to zaženemo preko naše google konzole, tako da uporabimo:
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):
# Ko časovni omejitveni ni določen, metoda izjeme čaka neskončno.
if message_future.exception(timeout=30):
print('Objavljanje sporočila na {} je povzročilo izjemo {}.'.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)Ko se datoteka zažene, bomo lahko opazovali izhodne podatke dnevnika na konzoli, kot je prikazano na spodnji sliki. Ta skripta bo delovala, dokler ne uporabimo CTRL+C, da jo zaključimo.
Slika 4. Izhod publish_logs.py
Pisanje kode našega cevovoda
Zdaj, ko smo vse uredili, lahko preidemo k najbolj zanimivemu delu - pisanju kode našega cevovoda z uporabo Beama in Pythona. Da bi ustvarili Beam cevovod, moramo ustvariti objekt cevovoda (p). Ko enkrat ustvarimo objekt cevovoda, lahko zaporedno uporabimo več funkcij z operatorjem pipe (|). Na splošno postopek dela izgleda, kot je prikazano na spodnji sliki.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) V naši kodi bomo ustvarili dve uporabniški funkciji. Funkcijo regex_clean, ki skenira podatke in izvleče ustrezno vrstico na podlagi seznama PATTERNS, z uporabo funkcije re.search. Funkcija vrne niz, ločen z vejicami. Če niste strokovnjak za regularne izraze, priporočam, da se seznanite s tem in the notebook to test the code. After that, we define a custom ParDo function called Split, which is a variation of the Beam transformation for parallel processing. In Python, this is done in a special way – we need to create a class that inherits from the Beam DoFn class. The Split function takes the parsed string from the previous function and returns a list of dictionaries with keys corresponding to the column names in our BigQuery table. There is something to note about this function: I had to import datetime inside the function for it to work. I was getting an error message when importing at the beginning of the file, which was strange. This list is then passed to the function WriteToBigQuery, which simply adds our data to the table. The code for the Batch DataFlow Job and Streaming DataFlow Job is given below. The only difference between the batch and streaming code is that in batch processing, we read the CSV from src_path, using the function ReadFromText from Beam.
Batch DataFlow Job (batch processing)
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 (stream processing)
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()
Zagon cevi
Pipeline lahko zaženemo na več načinov. Če bi želeli, bi ga lahko preprosto zagnali lokalno z terminala, tako da bi se oddaljeno prijavili v GCP.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingVendar pa ga bomo zagnali z uporabo DataFlow. To lahko storimo s spodnjim ukazom, pri čemer navedemo naslednje obvezne parametre.
project— ID vašega GCP projekta.runner— način izvajanja kontejnerja, ki bo analiziral vašo programsko opremo in zgradil obrat. Za izvajanje v oblaku morate navesti DataflowRunner.staging_location— pot do oblaka Cloud Dataflow za indeksiranje paketov kode, potrebnih obdelovalnikom, ki izvajajo naloge.temp_location— pot do oblaka Cloud Dataflow za shranjevanje začasnih datotek nalog, ustvarjenih med izvajanjem kontejnerja.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Пока эта команда выполняется, мы можем перейти на вкладку DataFlow в Google Console и просмотреть наш конвейер. Кликнув по конвейеру, мы должны увидеть что-то похожее на рисунок 4. В целях отладки может быть очень полезно перейти в логи, а затем в Stackdriver для просмотра подробных логов. Это помогло мне разрешить проблемы с конвейером в ряде случаев.
Рисунок 4: Beam-конвейер
Доступ к нашим данным в BigQuery
Итак, у нас уже должен быть запущен конвейер с данными, поступающими в нашу таблицу. Чтобы проверить это, мы можем перейти к BigQuery и просмотреть данные. После использования команды ниже вы должны увидеть первые несколько строк набора данных. Теперь, когда у нас есть данные, хранящиеся в BigQuery, мы можем провести дальнейший анализ, а также поделиться данными с коллегами и начать отвечать на бизнес-вопросы.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Рисунок 5: BigQuery
Zaključek
Надеемся, что этот пост послужит полезным примером создания потокового конвейера данных, а также поиска способов сделать данные более доступными. Хранение данных в таком формате дает нам много преимуществ. Теперь мы можем начать отвечать на важные вопросы, например, сколько людей используют наш продукт? Растет ли со временем база пользователей? С какими аспектами продукта люди взаимодействуют больше всего? И есть ли ошибки, там, где их быть не должно? Это те вопросы, которые будут интересны для организации. На основе идей, вытекающих из ответов на эти вопросы, мы сможем усовершенствовать продукт и повысить заинтересованность пользователей.
Beam je res uporaben za ta tip vaj in ima tudi vrsto drugih zanimivih primerov uporabe. Na primer, lahko analizirate podatke o borznih тикih v realnem času in trgujete na podlagi analize; morda imate podatke senzorjev, ki prihajajo iz vozil, in želite izračunati raven prometa. Prav tako ste lahko na primer igralna družba, ki zbira podatke o uporabnikih in jih uporablja za ustvarjanje nadzornih plošč za spremljanje ključnih pokazateljev. No, gospodje, to je že tema za drug post, hvala za branje, tistim, ki želijo videti celotno kodo, pa je spodaj povezava do mojega GitHub-a.
S tem je vse. .
Izvor: habr.com
