Skapar en datastreamingpipeline. Del 2

Hej alla. Vi delar med oss av översättningen av den avslutande delen av artikeln, som är speciellt framtagen för studenter på kursen „Data Engineer“. Du kan bekanta dig med den första delen ). Ett typiskt exempel på användning:.

Apache Beam och DataFlow för realtidsrörledningar

Inställning av Google Cloud

Observera: För att köra pipelinen och publicera data fra användarloggen använde jag Google Cloud Shell, eftersom jag hade problem med att köra pipelinen på Python 3. Google Cloud Shell använder Python 2, vilket fungerar bättre med Apache Beam.

För att köra pipelinen behöver vi gräva lite i inställningarna. De av er som aldrig har använt GCP tidigare måste följa dessa 6 steg som anges på denna sidan.

Därefter behöver vi ladda upp våra skript till Google Cloud Storage och kopiera dem till vår Google Cloud Shell. Att ladda upp till Google Cloud Storage är ganska enkelt (beskrivning finns att hitta här). För att kopiera våra filer kan vi öppna Google Cloud Shell från verktygsfältet genom att klicka på den första ikonen till vänster i bild 2 nedan.

Figur 2

Kommandona som vi behöver för att kopiera filer och installera nödvändiga bibliotek listas nedan.

# 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>

Skapande av vår databas och tabell

När vi har genomfört alla steg relaterade till inställning är nästa steg att skapa en dataset och tabell i BigQuery. Det finns flera sätt att göra detta, men det enklaste är att använda Google Cloud-konsolen, först skapa dataset. Du kan följa stegen som anges nedan länken, för att skapa en tabell med schema. Vår tabell kommer att ha 7 kolumner, motsvarande komponenterna i varje användarlogg. För enkelhetens skull definierar vi alla kolumner som strängar (typ string), förutom variabeln timelocal, och namnger dem enligt de variabler vi genererade tidigare. Schemat för vår tabell bör se ut som på bild 3.

Bild 3. Tabellschema

Publicering av data från användarloggen

Pub/Sub är en kritisk komponent i vår pipeline eftersom den möjliggör interaktion mellan flera oberoende applikationer. Specifikt fungerar det som en medlare, vilket låter oss skicka och ta emot meddelanden mellan applikationerna. Det första vi behöver göra är att skapa ett ämne (topic). Det är enkelt att bara gå till Pub/Sub i konsolen och klicka på SKAPA ÄMNE.

Koden nedan anropar vårt skript för att generera loggdata som definierades ovan och kopplar sedan upp sig för att skicka loggar till Pub/Sub. Det enda vi behöver göra är att skapa ett objekt PublisherClient, ange sökvägen till ämnet med metoden topic_path och anropa funktionen publish med topic_path med data. Observera att vi importerar generate_log_line från vårt skript stream_logs, så se till att dessa filer ligger i samma mapp, annars får du ett importfel. Därefter kan vi köra detta via vår Google-konsol med:

python publish.py

from 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 inte anges, väntar exceptionsmetoden oändligt.
    if message_future.exception(timeout=30):
        print('Publicering av meddelande på {} kastade ett undantag {}.'.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)

När filen har körts kan vi se utdata från loggen på konsolen som visas på bilden nedan. Detta skript kommer att fungera tills vi använder CTRL+C, för att avsluta det.

Figur 4. Utdata publish_logs.py

Skriva koden för vår pipeline

Nu när vi har förberett allt kan vi gå vidare till den mest intressanta delen — skriva koden för vår pipeline med hjälp av Beam och Python. För att skapa en Beam-pipeline behöver vi skapa ett pipeline-objekt (p). När vi har skapat pipeline-objektet kan vi tillämpa flera funktioner en efter en med hjälp av operatorn pipe (|). Generellt sett ser arbetsflödet ut som bilden nedan.

[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
             | [Second Transform]
             | [Third Transform])

I vår kod kommer vi att skapa två användardefinierade funktioner. Funktionen regex_clean, som genomsöker data och extraherar den relevanta strängen baserat på listan PATTERNS, med hjälp av funktionen re.search. Funktionen returnerar en kommaseparerad sträng. Om du inte är expert på reguljära uttryck rekommenderar jag att du kollar in den här handledningen och öva i anteckningsblocket för att testa koden. Efter detta definierar vi en användardefinierad ParDo-funktion som heter Split, vilket är en variation av Beam-transformering för parallell bearbetning. I Python görs detta på ett speciellt sätt – vi måste skapa en klass som ärvts från Beam DoFn-klassen. Split-funktionen tar emot en parserad sträng från den föregående funktionen och returnerar en lista av ordböcker med nycklar som motsvarar kolumnnamnen i vår BigQuery-tabell. Det finns något viktigt att notera om denna funktion: jag var tvungen att importera datetime inne i funktionen för att den skulle fungera. Jag fick ett felmeddelande när jag importerade i början av filen, vilket var konstigt. Denna lista skickas sedan till funktionen WriteToBigQuery, som helt enkelt lägger till våra data i tabellen. Koden för Batch DataFlow Job och Streaming DataFlow Job är angiven nedan. Den enda skillnaden mellan batch- och streamkod är att vi i batchbearbetningen läser CSV från src_path, med hjälp av funktionen ReadFromText från Beam.

Batch DataFlow Job (batchbearbetning)

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 (streambearbetning)

från apache_beam.options.pipeline_options import PipelineOptions
från google.cloud import pubsub_v1
från google.cloud import bigquery
import apache_beam som 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()

Starta pipelinen

Vi kan starta pipelinen på flera olika sätt. Om vi ville, kunde vi helt enkelt köra den lokalt från terminalen, genom att logga in på GCP på distans.

python -m main_pipeline_stream.py 
 --input_topic "projects/user-logs-237110/topics/userlogs" 
 --streaming

Men vi kommer att köra det med hjälp av DataFlow. Vi kan göra detta med hjälp av kommandot nedan, genom att ange följande obligatoriska parametrar.

• projekt — ID för ditt GCP-projekt.
• runner — verktyget som kör pipelinen, som analyserar ditt program och konstruerar din pipeline. För att köra i molnet måste du ange DataflowRunner.
• staging_location — sökvägen till Cloud Dataflow-molnlagringen för indexering av kodpaket som behövs för att jobba med uppgifterna.
• temp_location — sökvägen till Cloud Dataflow-molnlagringen för att lagra temporära filuppgifter som skapas under genomförandet av pipelinen.
• streaming

python main_pipeline_stream.py 
--runner DataFlow 
--project $PROJECT 
--temp_location $BUCKET/tmp 
--staging_location $BUCKET/staging
--streaming

Medan detta kommando körs kan vi gå till DataFlow-fliken i Google-konsolen och granska vår pipeline. Genom att klicka på pipelinen bör vi se något som liknar bild 4. För felsökning kan det vara väldigt användbart att gå till loggarna och sedan till Stackdriver för att se detaljerade loggar. Detta har hjälpt mig att lösa problem med pipelinen i flera fall.

Bild 4: Beam-pipeline

Åtkomst till våra data i BigQuery

Så vi bör redan ha en pipeline igång med data som flödar in i vår tabell. För att kontrollera detta kan vi gå till BigQuery och titta på datan. Efter att ha kört kommandot nedan bör du se de första raderna av datasetet. Nu när vi har data lagrade i BigQuery kan vi göra djupare analyser, dela data med kollegor och börja besvara affärsfrågor.

SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10;

Bild 5: BigQuery

Slutsats

Vi hoppas att det här inlägget fungerar som ett nyttigt exempel på att skapa en strömmande datakälla och att hitta sätt att göra data mer tillgängliga. Att lagra data i detta format ger oss många fördelar. Nu kan vi börja besvara viktiga frågor, såsom hur många personer som använder vår produkt? Växer vår användarbas över tid? Vilka aspekter av produkten interagerar folk mest med? Och finns det fel där det inte borde finnas? Dessa frågor är av intresse för organisationen. Baserat på insikterna från svaren på dessa frågor kan vi förbättra produkten och öka användarengagemanget.

Beam är verkligen användbar för den här typen av övningar och har också en rad andra intressanta användningsområden. Till exempel kan du analysera data om aktietick i realtid och genomföra affärer baserat på analysen. Kanske har du sensordata som kommer från fordon och du vill beräkna trafiknivån. Du kan också, till exempel, vara ett spelföretag som samlar in användardata och använder den för att skapa instrumentpaneler för att spåra nyckeltal. Okej, mina damer och herrar, det här är en annan diskussion, tack för att du läste, och för dem som vill se hela koden, här är länken till min GitHub.

https://github.com/DFoly/User_log_pipeline

Det är allt. Läs första delen.

Källa: habr.com