Създаваме конвейер за обработка на данни. Част 2

Здравейте на всички. Споделяме превод на заключителната част от статията, подготвена специално за студенти на курса „Data Engineer“. С първата част можете да се запознаете тук.

Apache Beam и DataFlow за реалновременни потоци

Настройка на Google Cloud

Забележка: За да стартирам конвейера и да публикувам данните на потребителския лог, използвах Google Cloud Shell, тъй като имах проблеми с пускането на конвейера на Python 3. Google Cloud Shell използва Python 2, който е по-съвместим с Apache Beam.

За да стартираме конвейера, трябва да се задълбочим в настройките. На тези от вас, които не са използвали GCP преди, е нужно да изпълнят следните 6 стъпки, посочени на тази страница.

След това ще трябва да качим нашите скриптове в облачното хранилище на Google и да ги копираме в нашия Google Cloud Shell. Качването в облачното хранилище е доста тривиално (можете да намерите описанието тук). За да копираме нашите файлове, можем да отворим Google Cloud Shell от панела с инструменти, като щракнем върху първата икона вляво на изображение 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 колонки, съответстващи на компонентите на всеки потребителски лог. За удобство ще определим всички колонки като низове (тип string), с изключение на променливата timelocal, и ще ги наречем в съответствие с променливите, които генерирахме по-рано. Схемата на нашата таблица трябва да изглежда като на изображение 3.

Изображение 3. Схема на таблицата

Публикуване на данни на потребителския лог

Pub/Sub е критично важен компонент на нашия конвейер, тъй като позволява на няколко независими приложения да взаимодействат помежду си. Конкретно, той функционира като посредник, който ни позволява да изпращаме и получаваме съобщения между приложенията. Първото, което трябва да направим, е да създадем тема (topic). Достатъчно е просто да отидете в 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):
    # Когато времето за изчакване не е зададено, методът за изключение чака неограничено.
    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, която е вариация на Beam трансформация за паралелна обработка. В Python това става по специален начин — трябва да създадем клас, който наследява класа DoFn Beam. Функцията Split приема разпарсена строка от предишната функция и връща списък от речници с ключове, съответстващи на имената на колоните в нашата таблица BigQuery. Има нещо, което трябва да се отбележи за тази функция: трябваше да импортирам datetime във функцията, за да работи. Получавах съобщение за грешка при импортиране в началото на файла, което беше странно. Този списък след това се предава на функцията WriteToBigQuery, която просто добавя нашите данни в таблицата. Кодът за Batch DataFlow Job и Streaming DataFlow Job е приведен по-долу. Единственото отличие между пакетния и потоковия код е, че в пакетната обработка четем CSV от src_path, използвайки функцията ReadFromText от Beam.

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"(?> 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 (обработка на потока)

от apache_beam.options.pipeline_options импортирайте PipelineOptions
от google.cloud импортирайте pubsub_v1
от google.cloud импортирайте bigquery
импорт apache_beam като beam
импорт logging
импорт argparse
импорт sys
импорт 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()

Стартиране на конвейера

Можем да стартираме конвейера по няколко различни начина. Ако желаем, можем просто да го стартираме локално от терминала, влизаща в GCP.

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

Но ние ще го стартираме с помощта на DataFlow. Можем да направим това с помощта на следната команда, задавайки следните задължителни параметри.

• проект — 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. За целите на дебъгването може да бъде много полезно да преминем в логовете, а след това в Stackdriver за преглед на подробните логове. Това ми помогна да разреша проблеми с конвейера в редица случаи.

Рисунка 4: Beam конвейер

Достъп до нашите данни в BigQuery

И така, вече трябва да имаме активен конвейер, който получава данни в нашата таблица. За да проверим това, можем да отидем в BigQuery и да прегледаме данните. След като използвате командата по-долу, трябва да видите първите няколко реда от набора данни. Сега, когато имаме данни, съхранени в BigQuery, можем да извършим допълнителен анализ, както и да споделим данните с колеги и да започнем да отговаряме на бизнес въпроси.

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

Рисунка 5: BigQuery

Заключение

Надяваме се, че този пост ще послужи като полезен пример за създаване на потоков конвейер за данни, както и за търсене на начини за улесняване на достъпа до данните. Съхраняването на данните в такъв формат ни дава много предимства. Сега можем да започнем да отговаряме на важни въпроси, например, колко хора използват нашия продукт? Расте ли базата ни от потребители с времето? С какви аспекти на продукта хората взаимодействат най-много? И има ли грешки там, където не трябва да има? Това са въпросите, които биха били интересни за организацията. На базата на идеите, произтичащи от отговорите на тези въпроси, можем да усъвършенстваме продукта и да увеличим ангажираността на потребителите.

Beam наистина е полезен за този вид упражнения и има редица други интересни приложения. Например, можете да анализирате данни за борсовите тикери в реално време и да извършвате сделки въз основа на анализа. Може да имате данни от сензори, постъпващи от превозни средства, и искате да изчислите нивото на трафика. Също така можете, например, да бъдете игрална компания, която събира данни за потребителите и ги използва за създаване на информационни панели за проследяване на ключовите показатели. Добре, господа, това е тема за друг пост, благодаря за четенето, а за тези, които искат да видят целия код, ето линк към моя GitHub.

https://github.com/DFoly/User_log_pipeline

На това е всичко. Прочетете първата част.

Източник: habr.com