Hallo allemaal. We delen de vertaling van het laatste deel van een artikel, speciaal voorbereid voor studenten van de cursus . Je kunt het eerste deel bekijken .
Apache Beam en DataFlow voor real-time pijplijnen
Google Cloud Instellen
Opmerking: Voor het uitvoeren van de pijplijn en het publiceren van gegevens van het gebruikerslogboek heb ik Google Cloud Shell gebruikt, omdat ik problemen had met het uitvoeren van de pijplijn op Python 3. Google Cloud Shell gebruikt Python 2, wat beter samenwerkt met Apache Beam.
Om de pijplijn uit te voeren, moeten we wat in de instellingen graven. Voor degenen onder jullie die GCP nog niet eerder hebben gebruikt, moeten de volgende 6 stappen worden uitgevoerd, zoals aangegeven in deze .
Daarna moeten we onze scripts uploaden naar Google Cloud Storage en ze kopiƫren naar onze Google Cloud Shell. Uploaden naar de cloudopslag is vrij eenvoudig (de beschrijving is te vinden ). Om onze bestanden te kopiƫren, kunnen we Google Cloud Shell openen vanuit het hulpmiddelenpaneel door op het eerste pictogram links te klikken in afbeelding 2 hieronder.
Figuur 2
De commando's die we nodig hebben om bestanden te kopiƫren en de benodigde bibliotheken te installeren, zijn hieronder weergegeven.
# 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>Onze database en tabel maken
Nadat we alle stappen met betrekking tot de configuratie hebben doorlopen, is de volgende stap die we moeten nemen, het maken van een dataset en een tabel in BigQuery. Er zijn verschillende manieren om dit te doen, maar de eenvoudigste is om de Google Cloud-console te gebruiken, eerst een dataset aan te maken. Je kunt de stappen volgen die zijn aangegeven in de volgende , om een tabel met een schema te maken. Onze tabel zal hebben 7 kolommen, die overeenkomen met de componenten van elk gebruikerslogboek. Voor de duidelijkheid definiƫren we alle kolommen als tekenreeksen (type string), met uitzondering van de variabele timelocal, en noemen we ze naar de variabelen die we eerder hebben gegenereerd. Het schema van onze tabel moet eruitzien zoals in afbeelding 3.
Afbeelding 3. Tabelschema
Publicatie van gegevens van gebruikerslogboeken
Pub/Sub is een cruciaal onderdeel van onze pijplijn, omdat het meerdere onafhankelijke toepassingen in staat stelt om met elkaar te communiceren. In het bijzonder fungeert het als een tussenpersoon, waardoor we berichten tussen applicaties kunnen verzenden en ontvangen. Het eerste wat we moeten doen, is een onderwerp (topic) aanmaken. Het is eenvoudig genoeg om naar Pub/Sub in de console te gaan en op CREĆER ONDERWERP te klikken.
De onderstaande code roept ons script aan voor het genereren van loggegevens zoals hierboven gedefinieerd, en maakt vervolgens verbinding om de logs naar Pub/Sub te verzenden. Het enige wat we moeten doen, is een object creƫren PublisherClient, het pad naar het onderwerp opgeven met de methode topic_path en de functie aanroepen publish met topic_path met de gegevens. Let op, we importeren generate_log_line uit ons script stream_logs, zorg ervoor dat deze bestanden in dezelfde map staan, anders krijg je een importfout. Vervolgens kunnen we dit via onze Google-console uitvoeren met:
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):
# Wanneer de time-out niet gespecificeerd is, wacht de exceptiemethode onbeperkt.
if message_future.exception(timeout=30):
print('Het publiceren van een bericht op {} gaf een uitzondering {}.'.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)Zodra het bestand wordt uitgevoerd, kunnen we de uitvoer van de loggegevens op de console zien, zoals weergegeven in de onderstaande afbeelding. Dit script blijft werken totdat we CTRL+C, om het te beƫindigen.
Figuur 4. Uitvoer publish_logs.py
Het schrijven van de code voor onze pipeline
Nu we alles hebben voorbereid, kunnen we beginnen met het meest interessante deel - het schrijven van de code voor onze pipeline met gebruik van Beam en Python. Om een Beam-pipeline te maken, moeten we een pipeline-object (p) creƫren. Zodra we het pipeline-object hebben gemaakt, kunnen we meerdere functies achtereenvolgens toepassen met behulp van de operator pipe (|). In het algemeen ziet de workflow eruit zoals weergegeven in de onderstaande afbeelding.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform]) In onze code zullen we twee gebruikersdefinities functies creĆ«ren. De functie regex_clean, die de gegevens scant en de relevante string extraheren op basis van de lijst PATTERNS, met gebruik van de functie re.search. De functie retourneert een door komma's gescheiden string. Als je geen expert bent in reguliere expressies, raad ik aan om deze en oefenen in het notitieblok om de code te testen. Vervolgens definiĆ«ren we een aangepaste ParDo-functie met de naam Split, die een variant is van de Beam-transformatie voor parallelle verwerking. In Python doen we dit op een bijzondere manier ā we moeten een klasse maken die erft van de DoFn-klasse van Beam. De functie Split neemt een geparsed string van de eerdere functie en retourneert een lijst van dictionaries met sleutels die overeenkomen met de kolomnamen in onze BigQuery-tabel. Er is iets dat opvalt aan deze functie: ik moest importeren datetime binnen de functie, zodat deze werkte. Ik ontving een foutmelding bij het importeren aan het begin van het bestand, wat vreemd was. Deze lijst wordt vervolgens doorgegeven aan de functie WriteToBigQuery, die simpelweg onze gegevens aan de tabel toevoegt. De code voor de Batch DataFlow Job en de Streaming DataFlow Job staat hieronder. Het enige verschil tussen de batch- en streamingcode is dat we in de batchverwerking CSV lezen van src_path, met behulp van de functie ReadFromText van Beam.
Batch DataFlow Job (batchverwerking)
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 (streamverwerking)
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()
Pipeline starten
Wij kunnen de pipeline op verschillende manieren starten. Als we zouden willen, konden we deze eenvoudig lokaal vanaf de terminal starten door op afstand in te loggen in GCP.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingEchter, we gaan deze starten met DataFlow. We kunnen dit doen met het onderstaande commando door de volgende vereiste parameters in te stellen.
projectā ID van uw GCP-project.runnerā het hulpmiddel dat de pipeline analyseert en uw pipeline bouwt. Voor uitvoering in de cloud moet u DataflowRunner opgeven.staging_locationā pad naar Cloud Dataflow-cloudopslag om de benodigde codepakketten voor de uitvoerende verwerkingsprogramma's te indexeren.temp_locationā pad naar Cloud Dataflow-cloudopslag voor het opslaan van tijdelijke bestanden die tijdens de uitvoering van de pipeline worden gemaakt.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
Terwijl dit commando wordt uitgevoerd, kunnen we naar het tabblad DataFlow in de Google-console gaan en ons pijplijn bekijken. Als we op de pijplijn klikken, zouden we iets moeten zien wat lijkt op afbeelding 4. Voor debugging kan het zeer nuttig zijn om naar de logs te gaan en vervolgens naar Stackdriver voor gedetailleerde logs. Dit heeft me in verschillende gevallen geholpen om problemen met de pijplijn op te lossen.
Afbeelding 4: Beam-pijplijn
Toegang tot onze gegevens in BigQuery
Dus, we zouden al een pijplijn moeten hebben die draait met gegevens die naar onze tabel worden gestuurd. Om dit te controleren, kunnen we naar BigQuery gaan en de gegevens bekijken. Na het uitvoeren van het onderstaande commando, zou je de eerste paar rijen van de dataset moeten zien. Nu we gegevens hebben opgeslagen in BigQuery, kunnen we verdergaande analyses uitvoeren, gegevens delen met collega's en beginnen met het beantwoorden van zakelijke vragen.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 
Afbeelding 5: BigQuery
Conclusie
We hopen dat deze post een nuttig voorbeeld zal zijn van het creƫren van een data-streamingpijplijn en het vinden van manieren om gegevens toegankelijker te maken. Het opslaan van gegevens in dit formaat biedt ons veel voordelen. Nu kunnen we beginnen met het beantwoorden van belangrijke vragen, zoals hoeveel mensen ons product gebruiken? Groeit de klantenbasis in de loop van de tijd? Met welke aspecten van het product interageren mensen het meest? En zijn er fouten waar ze niet zouden moeten zijn? Dit zijn de vragen die belangrijk zijn voor de organisatie. Op basis van de inzichten die voortkomen uit de antwoorden op deze vragen, kunnen we ons product verbeteren en de betrokkenheid van gebruikers vergroten.
Beam is inderdaad nuttig voor dit soort oefeningen en heeft ook een aantal andere interessante toepassingen. U kunt bijvoorbeeld gegevens van beurs-tickers in realtime analyseren en transacties uitvoeren op basis van die analyses. Misschien heeft u gegevens van sensoren die van voertuigen binnenkomen en wilt u het verkeersniveau berekenen. U kunt ook een gamingbedrijf zijn dat gegevens over gebruikers verzamelt en deze gebruikt voor het maken van dashboards om belangrijke prestatie-indicatoren te volgen. Goed, dames en heren, dit onderwerp is al voor een andere post, bedankt voor het lezen, en voor degenen die de volledige code willen zien, hieronder is de link naar mijn GitHub.
Dat is alles. .
Bron: habr.com
