ჩვენ ვქმნით ნაკადის მონაცემთა დამუშავების მილსადენს. Ნაწილი 2

Სალამი ყველას. გთავაზობთ სტატიის ბოლო ნაწილის თარგმანს, რომელიც სპეციალურად მომზადებულია კურსის სტუდენტებისთვის. მონაცემთა ინჟინერი. შეგიძლიათ წაიკითხოთ პირველი ნაწილი აქ.

Apache Beam და DataFlow რეალურ დროში მილსადენებისთვის

Google Cloud-ის დაყენება

შენიშვნა: მე გამოვიყენე Google Cloud Shell მილსადენის გასაშვებად და მორგებული ჟურნალის მონაცემების გამოქვეყნების გამო, რადგან მიჭირდა მილსადენის გაშვება Python 3-ში. Google Cloud Shell იყენებს Python 2-ს, რომელიც უფრო შეესაბამება Apache Beam-ს.

მილსადენის დასაწყებად, ჩვენ ცოტა უნდა ჩავუღრმავდეთ პარამეტრებს. მათთვის, ვისაც აქამდე არ გამოუყენებია GCP, მოგიწევთ დაიცვას ქვემოთ მოცემული 6 ნაბიჯი გვერდზე.

ამის შემდეგ, ჩვენ უნდა ავტვირთოთ ჩვენი სკრიპტები Google Cloud Storage-ში და დააკოპიროთ ისინი ჩვენს Google Cloud Shel-ში. ღრუბლოვან საცავში ატვირთვა საკმაოდ ტრივიალურია (აღწერილობა შეგიძლიათ ნახოთ აქ). ჩვენი ფაილების კოპირებისთვის, ჩვენ შეგვიძლია გავხსნათ Google Cloud Shel ხელსაწყოთა ზოლიდან, მარცხნივ პირველ ხატზე დაწკაპუნებით ქვემოთ, სურათზე 2.

ფიგურა 2

ბრძანებები, რომლებიც გვჭირდება ფაილების კოპირებისთვის და საჭირო ბიბლიოთეკების დასაყენებლად, ქვემოთ მოცემულია.

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

ჩვენი მონაცემთა ბაზის და ცხრილის შექმნა

მას შემდეგ რაც დავასრულებთ დაყენების ყველა საფეხურს, შემდეგი რაც უნდა გავაკეთოთ არის მონაცემთა და ცხრილის შექმნა BigQuery-ში. ამის გაკეთების რამდენიმე გზა არსებობს, მაგრამ უმარტივესი არის Google Cloud კონსოლის გამოყენება, პირველად მონაცემთა ნაკრების შექმნით. შეგიძლიათ მიჰყვეთ ქვემოთ მოცემულ ნაბიჯებს ლინკებისქემით ცხრილის შექმნა. ჩვენს მაგიდას ექნება 7 სვეტი, რომელიც შეესაბამება თითოეული მომხმარებლის ჟურნალის კომპონენტებს. მოხერხებულობისთვის, ჩვენ განვსაზღვრავთ ყველა სვეტს, როგორც სტრიქონებს, გარდა timelocal ცვლადისა და დავასახელებთ მათ ადრე შექმნილ ცვლადების მიხედვით. ჩვენი ცხრილის განლაგება უნდა გამოიყურებოდეს სურათი 3-ში.

სურათი 3. მაგიდის განლაგება

მომხმარებლის ჟურნალის მონაცემების გამოქვეყნება

Pub/Sub არის ჩვენი მილსადენის მნიშვნელოვანი კომპონენტი, რადგან ის საშუალებას აძლევს მრავალ დამოუკიდებელ აპლიკაციას დაუკავშირდნენ ერთმანეთს. კერძოდ, ის მუშაობს როგორც შუამავალი, რომელიც საშუალებას გვაძლევს გავაგზავნოთ და მივიღოთ შეტყობინებები აპლიკაციებს შორის. პირველი რაც უნდა გავაკეთოთ არის თემის შექმნა. უბრალოდ გადადით Pub/Sub-ზე კონსოლში და დააჭირეთ CREATE TOPIC.

ქვემოთ მოცემული კოდი მოუწოდებს ჩვენს სკრიპტს, რომ შექმნას ზემოთ განსაზღვრული ჟურნალის მონაცემები და შემდეგ აკავშირებს და აგზავნის ჟურნალებს Pub/Sub-ში. ერთადერთი რაც უნდა გავაკეთოთ არის ობიექტის შექმნა PublisherClient, მეთოდის გამოყენებით მიუთითეთ თემისკენ მიმავალი გზა topic_path და გამოიძახეთ ფუნქცია publish с topic_path და მონაცემები. გთხოვთ გაითვალისწინოთ, რომ ჩვენ იმპორტს generate_log_line ჩვენი სცენარიდან stream_logs, ამიტომ დარწმუნდით, რომ ეს ფაილები იმავე საქაღალდეშია, წინააღმდეგ შემთხვევაში თქვენ მიიღებთ იმპორტის შეცდომას. ამის შემდეგ ჩვენ შეგვიძლია გავატაროთ ეს ჩვენი Google კონსოლის მეშვეობით:

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):
    # When timeout is unspecified, the exception method waits indefinitely.
    if message_future.exception(timeout=30):
        print('Publishing message on {} threw an Exception {}.'.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)

როგორც კი ფაილი გაიშვება, ჩვენ შევძლებთ დავინახოთ ჟურნალის მონაცემების გამომავალი კონსოლში, როგორც ეს ნაჩვენებია ქვემოთ მოცემულ ფიგურაში. ეს სკრიპტი იმუშავებს მანამ, სანამ ჩვენ არ ვიყენებთ CTRL + Cმის დასასრულებლად.

სურათი 4. გამომავალი publish_logs.py

ჩვენი მილსადენის კოდის დაწერა

ახლა, როდესაც ყველაფერი მომზადებული გვაქვს, შეგვიძლია დავიწყოთ სახალისო ნაწილი - ჩვენი მილსადენის კოდირება Beam-ისა და Python-ის გამოყენებით. Beam მილსადენის შესაქმნელად, ჩვენ უნდა შევქმნათ მილსადენის ობიექტი (p). მას შემდეგ რაც შევქმნით მილსადენის ობიექტს, ჩვენ შეგვიძლია გამოვიყენოთ რამდენიმე ფუნქცია ერთმანეთის მიყოლებით ოპერატორის გამოყენებით pipe (|). ზოგადად, სამუშაო პროცესი ჰგავს ქვემოთ მოცემულ სურათს.

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

ჩვენს კოდში ჩვენ შევქმნით ორ მორგებულ ფუნქციას. ფუნქცია regex_clean, რომელიც სკანირებს მონაცემებს და იღებს შესაბამის მწკრივს PATTERNS სიის საფუძველზე ფუნქციის გამოყენებით re.search. ფუნქცია აბრუნებს მძიმით გამოყოფილი სტრიქონს. თუ თქვენ არ ხართ რეგულარული გამოხატვის ექსპერტი, გირჩევთ შეამოწმოთ ეს სახელმძღვანელო და ივარჯიშეთ ნოუთბუქში კოდის შესამოწმებლად. ამის შემდეგ ჩვენ განვსაზღვრავთ მორგებულ ParDo ფუნქციას, რომელსაც ეძახიან Split, რომელიც წარმოადგენს სხივის ტრანსფორმაციის ვარიაციას პარალელური დამუშავებისთვის. პითონში ეს კეთდება სპეციალური გზით - უნდა შევქმნათ კლასი, რომელიც მემკვიდრეობით მიიღებს DoFn Beam კლასის. Split ფუნქცია იღებს გაანალიზებულ მწკრივს წინა ფუნქციიდან და აბრუნებს ლექსიკონების სიას ჩვენს BigQuery ცხრილის სვეტების სახელების შესაბამისი კლავიშებით. არის რაღაც აღსანიშნავი ამ ფუნქციასთან დაკავშირებით: მომიწია იმპორტი datetime ფუნქციის შიგნით, რათა ის იმუშაოს. იმპორტის შეცდომას ვიღებდი ფაილის დასაწყისში, რაც უცნაური იყო. ეს სია შემდეგ გადაეცემა ფუნქციას WriteToBigQuery, რომელიც უბრალოდ ამატებს ჩვენს მონაცემებს ცხრილში. Batch DataFlow Job-ისა და Streaming DataFlow Job-ის კოდი მოცემულია ქვემოთ. ერთადერთი განსხვავება სერიულ და სტრიმინგ კოდს შორის არის ის, რომ პარტიაში ვკითხულობთ CSV-ს src_pathფუნქციის გამოყენებით ReadFromText სხივიდან.

Batch DataFlow Job (სამეფო დამუშავება)

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("There was an error with the regex search")
    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()

მონაცემთა ნაკადის სამუშაო (ნაკადის დამუშავება)

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"(?<=[).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("There was an error with the regex search")
    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],
            'body_bytes_sent': element[3],
            'status': element[4],
            'http_referer': element[5],
            'http_user_agent': element[6]
    
        }]

def main(argv=None):

   parser = argparse.ArgumentParser()
   parser.add_argument("--input_topic")
   parser.add_argument("--output")
   known_args = parser.parse_known_args(argv)


   p = beam.Pipeline(options=PipelineOptions())

   (p
      | 'ReadData' >> 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()

კონვეიერის გაშვება

ჩვენ შეგვიძლია გავატაროთ მილსადენი სხვადასხვა გზით. თუ გვინდოდა, შეგვეძლო მისი გაშვება ადგილობრივად ტერმინალიდან GCP-ში დისტანციურად შესვლისას.

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

თუმცა, ჩვენ ვაპირებთ მის გაშვებას DataFlow-ის გამოყენებით. ჩვენ შეგვიძლია ამის გაკეთება ქვემოთ მოცემული ბრძანების გამოყენებით შემდეგი საჭირო პარამეტრების დაყენებით.

• project — თქვენი GCP პროექტის ID.
• runner არის მილსადენის გამშვები, რომელიც გაანალიზებს თქვენს პროგრამას და ააშენებს თქვენს მილსადენს. ღრუბელში გასაშვებად, თქვენ უნდა მიუთითოთ DataflowRunner.
• staging_location — გზა Cloud Dataflow ღრუბლოვან საცავში, სამუშაოს შემსრულებელი პროცესორებისთვის საჭირო კოდის პაკეტების ინდექსაციისთვის.
• temp_location — გზა Cloud Dataflow ღრუბლოვანი საცავისკენ მილსადენის მუშაობის დროს შექმნილი სამუშაოების დროებითი ფაილების შესანახად.
• streaming

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

სანამ ეს ბრძანება მუშაობს, ჩვენ შეგვიძლია გადავიდეთ DataFlow ჩანართზე google კონსოლში და ვნახოთ ჩვენი მილსადენი. როდესაც ჩვენ ვაწკაპუნებთ მილსადენზე, ჩვენ უნდა დავინახოთ მსგავსი რამ, რაც ნახატ 4-ს. გამართვის მიზნებისთვის, შეიძლება ძალიან სასარგებლო იყოს Logs-ზე გადასვლა და შემდეგ Stackdriver-ზე დეტალური ჟურნალების სანახავად. ეს დამეხმარა მილსადენის საკითხების გადაჭრაში რიგ შემთხვევებში.

სურათი 4: სხივის კონვეიერი

ჩვენს მონაცემებზე წვდომა BigQuery-ში

ასე რომ, ჩვენ უკვე უნდა გვქონდეს მილსადენი, რომელიც მუშაობს ჩვენს ცხრილში მონაცემებით. ამის შესამოწმებლად შეგვიძლია გადავიდეთ BigQuery-ზე და შევხედოთ მონაცემებს. ქვემოთ მოცემული ბრძანების გამოყენების შემდეგ, თქვენ უნდა ნახოთ მონაცემთა ნაკრების პირველი რამდენიმე სტრიქონი. ახლა, როდესაც ჩვენ გვაქვს BigQuery-ში შენახული მონაცემები, შეგვიძლია ჩავატაროთ შემდგომი ანალიზი, ასევე გავაზიაროთ მონაცემები კოლეგებთან და დავიწყოთ ბიზნეს კითხვებზე პასუხის გაცემა.

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

სურათი 5: BigQuery

დასკვნა

ვიმედოვნებთ, რომ ეს პოსტი იქნება სასარგებლო მაგალითი ნაკადის მონაცემთა მილსადენის შესაქმნელად, ასევე მონაცემთა უფრო ხელმისაწვდომობის გასაუმჯობესებლად. მონაცემთა ამ ფორმატში შენახვა ბევრ უპირატესობას გვაძლევს. ახლა ჩვენ შეგვიძლია დავიწყოთ პასუხის გაცემა მნიშვნელოვან კითხვებზე, როგორიცაა რამდენი ადამიანი იყენებს ჩვენს პროდუქტს? იზრდება თქვენი მომხმარებლის ბაზა დროთა განმავლობაში? პროდუქტის რომელ ასპექტებთან ურთიერთობენ ადამიანები ყველაზე მეტად? და არის შეცდომები სადაც არ უნდა იყოს? ეს ის კითხვებია, რომლებიც ორგანიზაციისთვის საინტერესო იქნება. ამ კითხვებზე პასუხებიდან მიღებული შეხედულებების საფუძველზე, ჩვენ შეგვიძლია გავაუმჯობესოთ პროდუქტი და გავზარდოთ მომხმარებლის ჩართულობა.

Beam მართლაც სასარგებლოა ამ ტიპის ვარჯიშისთვის და აქვს სხვა საინტერესო გამოყენების შემთხვევებიც. მაგალითად, შეიძლება გსურთ რეალურ დროში გააანალიზოთ საფონდო ტკიპების მონაცემები და გააკეთოთ ვაჭრობა ანალიზის საფუძველზე, შესაძლოა, თქვენ გაქვთ სენსორის მონაცემები, რომლებიც მოდის მანქანებიდან და გსურთ გამოთვალოთ ტრაფიკის დონის გამოთვლები. თქვენ ასევე შეიძლება იყოთ, მაგალითად, სათამაშო კომპანია, რომელიც აგროვებს მომხმარებლის მონაცემებს და იყენებს მას საინფორმაციო დაფების შესაქმნელად ძირითადი მეტრიკის თვალყურის დევნებისთვის. კარგი, ბატონებო, ეს არის კიდევ ერთი პოსტის თემა, გმადლობთ, რომ კითხულობთ, ხოლო ვისაც სურს სრული კოდის ნახვა, ქვემოთ არის ჩემი GitHub-ის ბმული.

https://github.com/DFoly/User_log_pipeline

ეს ყველაფერი. წაიკითხეთ ნაწილი პირველი.

წყარო: www.habr.com