Създаваме конвейер за потокова обработка на данни. Част 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('Публикуването на съобщение в {} предизвика изключение {}.'.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, която сканира данните и извлича съответстваща низ на базата на списък с ШАБЛОНИ, използвайки функцията 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 за временно съхранение на файловете от задачите, генерирани по време на работа на конвейера.
• стрийминг

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