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

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

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

Днешната публикация е базирана на задача, с която наскоро се занимавах на работа. Бях наистина радостен, че мога да я реализирам и да опиша свършената работа във формат на блог публикация, тъй като това ми даде възможност да се занимавам с дата инженеринг, а също така да направя нещо, което би било наистина полезно за моя отбор. Неотдавна открих, че в нашите системи се съхранява доста голям потребителски лог, свързан с един от нашите продукти за работа с данни. Оказа се, че никой не е използвал тези данни, затова веднага се заинтересувах какво можем да научим, ако започнем редовно да ги анализираме. Въпреки това, на пътя стояха няколко проблема. Първият проблем беше, че данните се съхраняваха в много различни текстови файлове, които не бяха достъпни за мигновен анализ. Вторият проблем беше, че те бяха запазени в затворена система, затова не можех да използвам нито един от моите любими инструменти за анализ на данни.

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

Така че наличието на данни в облачна база данни изглеждаше разумен начин за решаване на първия ми проблем, но какво да направя с проблем номер 2? За щастие, имаше начин да пренеса тези данни в среда, където можех да получа достъп до инструменти като Python и Google Cloud Platform (GCP). Въпреки това, това беше дълъг процес, така че трябваше да направя нещо, което да ми позволи да продължа разработката, докато чаках приключването на преноса на данните. Решението, до което стигнах, беше да създам фалшиви данни, използвайки библиотеката Faker в Python. Никога преди не бях използвал тази библиотека, но бързо осъзнах колко е полезна. Използването на този подход ми позволи да започна да пиша код и да тествам потока без реални данни.

С оглед на всичко казано дотук, в този пост ще разкажа как построих описания по-горе поток, използвайки някои от технологиите, налични в GCP. По-специално, ще използвам Apache Beam (Python версия), Dataflow, Pub/Sub и Big Query за събиране на потребителски журнали, трансформиране на данни и предаване на тях в база данни за по-нататъшен анализ. В моя случай необходима ми беше само пакетната функционалност на Beam, тъй като данните ми не идваха в реално време, така че Pub/Sub не беше нужен. Въпреки това, ще се спра и на потоковата версия, тъй като това е нещо, с което можете да се срещнете на практика.

Въведение в GCP и Apache Beam

Google Cloud Platform предлага набор от наистина полезни инструменти за обработка на големи данни. Ето някои от инструментите, които ще използвам:

Pub/Sub е услуга за обмен на съобщения, използваща модела Издател-Абонат, която ни позволява да получаваме данни в реално време.
DataFlow е услуга, която опростява изграждането на потоци за данни и автоматично разрешава задачи като мащабиране на инфраструктурата, което означава, че можем да се съсредоточим само върху написването на кода за нашия поток.
BigQuery е облачно хранилище за данни. Ако сте запознати с други SQL бази данни, с BigQuery не би трябвало да имате трудности.
И накрая, ще използваме Apache Beam, по-специално, ще се фокусираме върху версията на Python за създаване на нашия конвейер. Този инструмент ще ни позволи да създадем конвейер за потокова или пакетна обработка, който се интегрира с GCP. Той е особено полезен за паралелна обработка и е подходящ за задачи, свързани с извличане, трансформация и зареждане (ETL), така че, ако ни трябва да преместваме данни от едно място на друго с извършване на трансформации или изчисления, Beam е добър избор.

На GCP има голямо разнообразие от инструменти, така че може да бъде трудно да се следят всички и каква е тяхната цел, но ето един обзор за справка.
В GCP има много инструменти, така че обхващането на всичките, включително и предназначението им, може да бъде сложно, но все пак ето кратко изложение за справка.

Визуализация на нашия конвейер

Нека визуализираме компонентите на нашия конвейер на рисунка 1. На високо ниво искаме да събираме данни от потребители в реално време, да ги обработваме и да ги предаваме в BigQuery. Логовете се създават, когато потребителите взаимодействат с продукта, изпращайки заявки на сървъра, които след това се регистрират. Тези данни могат да бъдат особено полезни за разбиране как потребителите взаимодействат с нашия продукт и дали той функционира правилно. Като цяло конвейерът ще съдържа следните етапи:

Beam прави този процес много прост, независимо дали имаме потоков източник на данни или файл CSV, и искаме да извършим пакетна обработка. По-късно ще видите, че в кода има само минимални промени, необходими за превключване между тях. Това е едно от предимствата на използването на Beam.

Рисунок 1: Основният конвейер за данни: Източник:

Създаване на псевдоданни с помощта на Faker

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

192.52.197.161 - - [30/Apr/2019:21:11:42] "PUT /tag/category/tag HTTP/1.1" [401] 155 "https://harris-lopez.com/categories/about/" "Mozilla/5.0 (Macintosh; PPC Mac OS X 10_11_2) AppleWebKit/5312 (KHTML, like Gecko) Chrome/34.0.855.0 Safari/5312"

На базата на горния ред искаме да създадем нашата променлива LINE, използвайки 7 променливи в фигурни скоби по-долу. Също така ще ги използваме като имена на променливи в нашата таблица по-късно.

""" {remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}" """

Ако извършвахме пакетна обработка, кодът щеше да бъде много сходен, макар да трябваше да създадем набор от примери в определен времеви диапазон. За да използваме Faker, просто създаваме обект и извикваме необходимите ни методи. В частност, Faker е бил полезен за генериране на IP адреси, а също и уебсайтове. Използвах следните методи:

фалшиво.ipv4() фалшиво.uri_path() фалшиво.uri() фалшиво.user_agent()

from faker import Faker
import time
import random
import os
import numpy as np
from datetime import datetime, timedelta



LINE = """
{remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"
"""


def generate_log_line():
    fake = Faker()
    now = datetime.now()
    remote_addr = fake.ipv4()
    time_local = now.strftime('%d/%b/%Y:%H:%M:%S')
    request_type = random.choice(["GET", "POST", "PUT"])
    request_path = "\/" + fake.uri_path()

    status = np.random.choice([200, 401, 404], p = [0.9, 0.05, 0.05])
    body_bytes_sent = random.choice(range(5, 1000, 1))
    http_referer = fake.uri()
    http_user_agent = fake.user_agent()

    log_line = LINE.format(
        remote_addr=remote_addr,
        time_local=time_local,
        request_type=request_type,
        request_path=request_path,
        status=status,
        body_bytes_sent=body_bytes_sent,
        http_referer=http_referer,
        http_user_agent=http_user_agent
    )

    return log_line

Край на първа част.

В следващите дни ще споделим с вас продължение на статията, а в момента традиционно очакваме вашите коментари ;-).

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