Hej alle sammen. Vi deler oversættelsen af den afsluttende del af artiklen, der er forberedt specielt til studerende på kurset . Den første del kan læses .
Apache Beam og DataFlow til realtids pipelines
Konfiguration af Google Cloud
Bemærk: For at starte pipelinen og offentliggøre data fra loggen brugte jeg Google Cloud Shell, da jeg havde problemer med at starte pipelinen på Python 3. Google Cloud Shell bruger Python 2, som fungerer bedre med Apache Beam.
For at starte pipelinen skal vi dykke lidt ned i indstillingerne. De af jer, der ikke har brugt GCP før, skal følge de næste 6 trin, der er givet på denne .
Derefter skal vi uploade vores scripts til Google Cloud Storage og kopiere dem til vores Google Cloud Shell. Upload til cloud storage er ret ligetil (beskrivelsen kan findes ). For at kopiere vores filer kan vi åbne Google Cloud Shell fra værktøjslinjen ved at klikke på det første ikon til venstre på billede 2 nedenfor.
Billede 2
De kommandoer, vi har brug for til at kopiere filer og installere de nødvendige biblioteker, er anført nedenfor.
# 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>Oprettelse af vores database og tabel
Når vi har fulgt alle trinene i opsætningen, er det næste, vi skal gøre, at oprette et datasæt og en tabel i BigQuery. Der er flere måder at gøre dette på, men den nemmeste er at bruge Google Cloud-konsollen til først at oprette datasættet. Du kan udføre de handlinger, der er angivet på den næste , for at oprette en tabel med skema. Vores tabel vil have 7 kolonner, svarende til komponenterne i hver brugerlog. For nemheds skyld vil vi definere alle kolonner som strenge (type string), undtagen variablen timelocal, og navngive dem i henhold til de variabler, vi genererede tidligere. Skemaet for vores tabel skal se sådan ud som på billede 3.
Billede 3. Skema for tabel
Offentliggørelse af brugerlogdata
Pub/Sub er en kritisk komponent i vores pipeline, da den muliggør, at flere uafhængige applikationer interagerer med hinanden. Specifikt fungerer den som en mægler, der giver os mulighed for at sende og modtage beskeder mellem applikationer. Det første, vi skal gøre, er at oprette et emne (topic). Det er tilstrækkeligt at gå til Pub/Sub i konsollen og klikke på Opret emne.
Den nedenstående kode kalder vores script for at generere logdata, som beskrevet ovenfor, og tilslutter sig og sender logfiler til Pub/Sub. Det eneste, vi skal gøre, er at oprette et objekt PublisherClient, angive stien til emnet med metoden topic_path og kalde funktionen publish med topic_path og data. Vær opmærksom på, at vi importerer generate_log_line fra vores script stream_logs, så sørg for, at disse filer er i samme mappe, ellers får du en importfejl. Så kan vi køre dette gennem vores Google-konsol ved hjælp af:
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):
# Når timeout ikke er angivet, venter exceptionsmetoden ubegribentligt.
if message_future.exception(timeout=30):
print('Udgiver besked på {} udløste en undtagelse {}.'.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)Så snart filen kører, kan vi se uddataene fra loggen på konsollen, som vist på billedet nedenfor. Dette script vil fungere, indtil vi bruger CTRL+C, for at stoppe det.
Figur 4. Udgivelse publish_logs.py
Skrivningen af vores pipeline-kode
Nu hvor vi har forberedt alt, kan vi gå videre til den mest interessante del — at skrive koden til vores pipeline ved hjælp af Beam og Python. For at oprette en Beam-pipeline skal vi oprette et pipeline-objekt (p). Når vi har oprettet pipeline-objektet, kan vi anvende forskellige funktioner en efter en ved hjælp af operatoren pipe (|). Generelt ser arbejdsprocessen ud som på billedet nedenfor.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) I vores kode vil vi lave to brugerdefinerede funktioner. Funktionen regex_clean, som scanner data og trækker den relevante streng baseret på listen over PATTERNS ved hjælp af funktionen re.search. Funktionen returnerer en streng adskilt af kommaer. Hvis du ikke er ekspert i regulære udtryk, anbefaler jeg at du ser på denne og og øve i notesbogen for at teste koden. Derefter definerer vi en brugerdefineret ParDo-funktion kaldet Split, som er en variation af Beam-transformationen til parallel behandling. I Python gøres dette på en særlig måde — vi skal oprette en klasse, der arver fra Beam's DoFn-klasse. Split-funktionen tager en parseret streng fra den tidligere funktion og returnerer en liste af ordbøger med nøgler, der svarer til kolonnenavne i vores BigQuery-tabel. Der er noget, der bør bemærkes om denne funktion: jeg måtte importere datetime inden i funktionen for at få den til at fungere. Jeg fik en fejlmeddelelse, da jeg importerede i begyndelsen af filen, hvilket var underligt. Denne liste sendes derefter til funktionen WriteToBigQuery, som ganske enkelt tilføjer vores data til tabellen. Koden til Batch DataFlow Job og Streaming DataFlow Job er nedenfor. Den eneste forskel mellem batch- og streamingkoden er, at vi i batchbehandlingen læser CSV fra src_path, ved hjælp af funktionen ReadFromText fra Beam.
Batch DataFlow Job (batchbehandling)
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"(?<=[).d+(?=])",
r'"[A-Z][a-z]+', r'"(http|https):\/\/[a-z]+.[a-z]+.[a-z]+']
result = []
for match in PATTERNS:
try:
reg_match = re.search(match, data).group()
if reg_match:
result.append(reg_match)
else:
result.append(" ")
except:
print("Der var en fejl med regex-søgningen")
result = [x.strip() for x in result]
result = [x.replace('"', "") for x in result]
res = ','.join(result)
return res
class Split(beam.DoFn):
def process(self, element):
from datetime import datetime
element = element.split(",")
d = datetime.strptime(element[1], "%d\/ %b\/ %Y:%H:%M:%S")
date_string = d.strftime("%Y-%m-%d %H:%M:%S")
return [{
'remote_addr': element[0],
'timelocal': date_string,
'request_type': element[2],
'status': element[3],
'body_bytes_sent': element[4],
'http_referer': element[5],
'http_user_agent': element[6]
}]
def main():
p = beam.Pipeline(options=PipelineOptions())
(p
| 'ReadData' >> 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 (streambehandling)
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()
Kør pipeline
Vi kan køre pipelinen på flere forskellige måder. Hvis vi ønskede det, kunne vi blot køre den lokalt fra terminalen, ved at logge ind på GCP på afstand.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingMen vi vil køre den ved hjælp af DataFlow. Vi kan gøre dette med nedenstående kommando, og indstille de nødvendige parametre.
projekt— ID for dit GCP-projekt.runner— værktøjet der kører pipelinen, som analyserer dit program og konstruerer din pipeline. For at køre i skyen skal du angive DataflowRunner.staging_location— banen til Cloud Dataflow's cloud storage til indeksering af de nødvendige kodepakker til de job, der udfører arbejdet.temp_location— banen til Cloud Dataflow's cloud storage til placering af midlertidige jobfiler, der er oprettet under udførelsen af pipelinen.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Når denne kommando udføres, kan vi skifte til DataFlow fanen i Google Console og se vores pipeline. Ved at klikke på pipelinen bør vi se noget, der ligner billede 4. Til debug formål kan det være meget nyttigt at gå til loggene og derefter Stackdriver for at se detaljerede logs. Dette har hjulpet mig med at løse problemer med pipelinen i flere tilfælde.
Billede 4: Beam-pipeline
Adgang til vores data i BigQuery
Så vi skal allerede have en pipeline kørende med data, der strømmer ind i vores tabel. For at kontrollere dette kan vi gå til BigQuery og se på dataene. Efter at have kørt kommandoen nedenfor, bør du kunne se de første par rækker af datasættet. Nu, når vi har data gemt i BigQuery, kan vi foretage videre analyser, samt dele dataene med kolleger og begynde at besvare forretningsspørgsmål.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Billede 5: BigQuery
Konklusion
Vi håber, at dette indlæg vil være en nyttig eksempel på oprettelsen af en streaming dataløb, samt at finde måder at gøre dataene mere tilgængelige. At gemme data i et sådant format giver os mange fordele. Nu kan vi begynde at besvare vigtige spørgsmål, såsom hvor mange mennesker der bruger vores produkt? Vokser brugerdatabasen over tid? Hvilke aspekter af produktet interagerer folk mest med? Og er der fejl, hvor der ikke bør være? Dette er de spørgsmål, der vil være interessante for organisationen. Baseret på de indsigt, der kommer fra svarene på disse spørgsmål, vil vi kunne forbedre produktet og øge brugerengagement.
Beam er virkelig nyttig til denne type øvelser og har desuden en række andre interessante anvendelsestilfælde. For eksempel kan du analysere data om aktietickers i realtid og træffe handelsbeslutninger baseret på dette analyse. Måske har du sensor data fra køretøjer, og du vil beregne trafikniveauet. Du kan også være et spilfirma, der indsamler brugerdata og bruger det til at lave dashboards, der overvåger nøgletal. Nå, mine herrer, det er et emne for et andet indlæg, tak for læsningen, og for dem, der ønsker at se den fulde kode, herunder finder du et link til min GitHub.
Det var alt for nu. .
Kilde: habr.com
