Ние создаваме цевковод за обработка на податоци за поток. Дел 2

Здраво на сите. Го споделуваме преводот на последниот дел од статијата, подготвен специјално за студентите на курсот. Инженер за податоци. Можете да го прочитате првиот дел тука.

Apache Beam и Data Flow за цевководи во реално време

Поставување на 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 pipeline, треба да создадеме објект на цевковод (p). Откако ќе создадеме објект на цевковод, можеме да примениме повеќе функции една по друга со помош на операторот pipe (|). Во принцип, работниот тек изгледа како сликата подолу.

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

Во нашиот код, ќе создадеме две сопствени функции. Функција regex_clean, кој ги скенира податоците и го враќа соодветниот ред врз основа на списокот PATTERNS користејќи ја функцијата re.search. Функцијата враќа низа одвоена со запирка. Ако не сте експерт за редовно изразување, препорачувам да го проверите ова упатство и вежбајте во бележник да го проверите кодот. По ова дефинираме прилагодена функција ParDo наречена Подели, што е варијација на Beam трансформацијата за паралелна обработка. Во Python, ова се прави на посебен начин - мора да создадеме класа што ќе наследи од класата DoFn Beam. Функцијата Split го зема анализираниот ред од претходната функција и враќа листа на речници со копчиња што одговараат на имињата на колоните во нашата табела BigQuery. Има нешто што треба да се забележи за оваа функција: морав да увезам datetime внатре во функција за да функционира. Добив грешка при увоз на почетокот на датотеката, што беше чудно. Оваа листа потоа се пренесува на функцијата WriteToBigQuery, што едноставно ги додава нашите податоци на табелата. Кодот за Batch DataFlow Job и Streaming DataFlow Job е даден подолу. Единствената разлика помеѓу серискиот и стриминг кодот е тоа што во серија го читаме CSV од src_pathкористејќи ја функцијата ReadFromText од Зрак.

Работа за сериски проток на податоци (сериска обработка)

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 — ID на вашиот GCP проект.
• 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