Title: Przetwarzanie równoległe i strumieniowe

PRS

Przetwarzanie równoległe i strumieniowe

Wykład 9

Przetwarzanie danych strumieniowych w Streamz. Apache Flink - architektura

Streamz vs NumPy or Pandas

Algorytmy do obsługi danych strumieniowych różnią się od tych, które można znaleźć w bibliotekach takich jak NumPy i Pandas, które zakładają, że znają wszystkie dane z góry.

Dla strumieni danych możemy używać NumPy i Pandas, ale trzeba je sprytnie połączyć z innymi rozwiązaniami i przechowywać wystarczającą ilość danych pośrednich, aby obliczyć marginalne aktualizacje, gdy pojawią się nowe dane.

Streamz

Streamz pomaga budować przetwarzanie do zarządzania ciągłymi strumieniami danych. Jest skuteczny w użyciu w prostych przypadkach, ale obsługuje również złożone przetwarzania, które obejmują rozgałęzienia, łączenie, kontrolę przepływu, sprzężenie zwrotne i tak dalej.

Streamz

Streamz został opublikowany przez Matthew Rocklina w 2017 roku i jest cały czas rozwijany :

https://github.com/python-streamz/streamz

Streamz

Ciągłe strumienie danych pojawiają się w wielu aplikacjach, takich jak:

- Przetwarzanie logów z serwerów WWW
- Dane z instrumentów naukowych, takie jak telemetria lub potoki przetwarzania obrazu
- Finansowe szeregi czasowe
- Potoki uczenia maszynowego do uczenia się w czasie rzeczywistym i on-line

Streamz

Czasami przetwarzanie danych jest stosunkowo proste,
z liniową sekwencją etapów przetwarzania:

https://streamz.readthedocs.io/en/latest/

Streamz

Czasami przetwarzanie jest bardziej złożone i obejmuje rozgałęzienia, powroty, informacje zwrotne dotyczące wcześniejszych etapów i nie tylko.

https://streamz.readthedocs.io/en/latest/

Streamz - przykład

Przetwarzanie rozpoczynamy tworząc strumień danych :

source = Stream()

i kończymy operacją sink, która aplikuje funkcję podaną w parametrze do wszystkich elementów strumienia i kończy strumień.

source.sink(print)

Streamz - przykład

Mając zdefiniowany proces przetwarzania danych możemy zacząć dodawać dane do przetwarzania. Wykorzystujemy do tego metodę emit().

>>> source.emit(1)
2
>>> source.emit(2)
3
>>> source.emit(10)
11

Streamz - map

Map jest pośrednią funkcją strumienia danych która aplikuje wybraną w parametrze funkcje do wszystkich elementów strumienia i zwraca strumień nowych, zmienionych wartości.

def increment(x):
    return x + 1

source = Stream()
source.map(increment).sink(print)

Streamz - accumulate

Map i sink przekazują dane bezpośrednio jako strumień. Wchodzi jeden element danych, wychodzi jeden lub zero.

Accumulate umożliwia śledzenie niektórych stanów w potoku. Przyjmuje funkcję akumulacji, która przyjmuje jako wejście poprzedni stan i nowy element, a następnie zwraca nowy stan i nowy element do wyemitowania. W poniższym przykładzie tworzymy akumulator, który przechowuje bieżącą sumę elementów widzianych do tej pory.

def add(x, y):
    return x + y

source = Stream()
source.accumulate(add).sink(print)

Streamz - accumulate

from streamz import Stream

def num_distinct(state, new):
    state.add(new)
    return state, len(state)

source = Stream()
source.accumulate(num_distinct, returns_state=True, start=set()).sink(print)

source.emit('cat')
source.emit('dog')
source.emit('cat')
source.emit('mouse')

Streamz - Flow Control

Funkcje kontroli przepływu danych:

buffer(upstream, n, **kwargs) - Pozwala na gromadzenie danych w punkcie strumienia
flatten([upstream, upstreams, stream_name, ...]) - Rozpakowuje strumień list na strumień wartości
partition(upstream, n[, timeout, key]) - Dzieli strumień na tuple równym rozmiarze
sliding_window(upstream, n[, return_partial]) - Produkuje okna danych o równej wartości
union(*upstreams, **kwargs) - Łączy strumienie
unique(upstream[, maxsize, key, hashable]) - Usuwa duplikaty ze strumienia

Streamz - Flow Control

Streamz - Flow Control

Umożliwia sprawdzenie zdarzenia, które zaszło w obrębie ostatnich X odczytów (np alarm).

source = Stream()
source.sliding_window(3, return_partial=False).sink(print)

>>> source.emit(1)
>>> source.emit(2)
>>> source.emit(3)
(1, 2, 3)
>>> source.emit(4)
(2, 3, 4)
>>> source.emit(5)
(3, 4, 5)

Streamz - Branching and Joining

Umożliwiają łączenie wielu strumieni w jeden strumień danych :

combine_latest(*upstreams, **kwargs)

łączy wiele strumieni w tuple gdy ktokolwiek prześle dane

zip(*upstreams, **kwargs)

łączy strumiene w tuple gdy wszyscy wyślą dane

zip_latest(lossless, *upstreams, **kwargs)

łączy wiele strumieni w tuple, wybieramy który strumień musi wysłać daną

Streamz - Branching and Joining

Streamz - Processing Time and Back Pressure

Umożliwiają kontrolę przepływu i prędkości przetwarzania danych:

delay(upstream, interval, **kwargs)

Dodaje opóżnienie

rate_limit(upstream, interval, **kwargs)

Ogranicza przepływ danych

timed_window(upstream, interval, **kwargs)

Co zadany interwał emituje dane dodane przez ten czas

Streamz - braki

- Przetwarzanie równoległe
- Pozyskiwanie danych z popularnych źródeł przesyłania strumieniowego, takich jak Kafka.
- Dostęp do danych poza kolejnością. Tu przydatniejsze są bardziej rozbudwane bibilioteki, jak Flink.
- Wydajność: Niektóre z operacji obejmują nietrywialne kopiowanie, zwłaszcza w przypadku większych okien. Streamz polega na bibliotece Pandas, która nie została zaprojektowana z myślą o szybko zmieniających się danych.
- Nie wszystkie typowe operacje (takich jak wariancja) zostały zaimplementowane
- Wykresy wyników

Apache Flink

Apache Flink

Apache Flink to framework i silnik przetwarzania rozproszonego do obliczeń stanowych na nieograniczonych i ograniczonych strumieniach danych. Flink został zaprojektowany do pracy we wszystkich typowych środowiskach klastrowych, wykonując obliczenia w pamięci, z dużą prędkością i w dowolnej skali.

Flink umożliwia przetwarzanie danych w modelu batchowym (gdy wszystkie dane są dostępne przy uruchomieniu przetwarzania - strumień ograniczony) oraz strumieniowym, w którym dane są ciagle dodawane a obliczenia i agregacje są zwykle wywoływane w oknach czasowych (strumień nieograniczony).

Apache Flink - architektura

JobClient w Apache Flink to klient, który umożliwia interakcję z JobManagerem w celu przesłania, uruchomienia i monitorowania programów Flink. JobClient dostarcza programistom interfejs do komunikacji z JobManagerem i wykonuje zadania związane z obsługą aplikacji Flink.

Apache Flink - job client

Główne funkcje i zadania, którymi zajmuje się JobClient:

1. Przesyłanie programów: JobClient pozwala programistom przesłać skompilowany program Flink (JAR) lub program napisany w Flink API do JobManagera. Klient przekazuje program do JobManagera, który następnie planuje i uruchamia zadanie przetwarzania.
2. Uruchamianie zadań: JobClient inicjuje uruchomienie zadania przetwarzania na JobManagerze.
3. Monitorowanie zadań: JobClient umożliwia programistom monitorowanie postępu wykonania zadania. Klient odbiera informacje zwrotne od JobManagera, takie jak metryki, status zadania, postęp i ewentualne błędy.
4. Zatrzymywanie zadań: JobClient umożliwia również zatrzymanie i anulowanie zadania przetwarzania.

Apache Flink

Środowisko wykonawcze Flink składa się z dwóch typów procesów: JobManagera i jednego lub więcej TaskManagerów.

JobManager w Apache Flink to główny komponent w systemie Flink, odpowiedzialny za zarządzanie i nadzorowanie wykonania programów Flink. JobManager jest węzłem sterującym, który zarządza przetwarzaniem danych i koordynuje pracę węzłów wykonawczych.

Apache Flink - JobManager

Główne zadania, którymi zajmuje się JobManager:

1. Zarządzanie programami: JobManager przyjmuje programy napisane w Flink API lub innym języku programowania, takim jak Python, Java lub Scala. Po otrzymaniu programu JobManager planuje jego wykonanie i zarządza wszystkimi zasobami potrzebnymi do przetwarzania danych.
2. Planowanie i koordynacja: JobManager jest odpowiedzialny za planowanie zadań (tasków) przetwarzania danych i ich alokację na węzłach wykonawczych. Tworzy plan wykonania programu, uwzględniając dostępne zasoby, takie jak CPU i pamięć, oraz optymalizuje plan dla wydajnego przetwarzania.

Apache Flink - JobManager

1. Zarządzanie stanem: Flink umożliwia przetwarzanie strumieniowe i wsadowe. JobManager odpowiada za zarządzanie stanem przetwarzania, co oznacza, że śledzi postęp przetwarzania, zapamiętuje stan aplikacji i umożliwia checkpointing (tworzenie migawek stanu) w celu odzyskiwania po awarii systemu.
2. Koordynacja z węzłami wykonawczymi: JobManager komunikuje się z węzłami wykonawczymi, które wykonują rzeczywiste zadania przetwarzania danych. Przesyła im zadania, monitoruje ich postęp i otrzymuje wyniki z powrotem. Koordynuje przepływ danych między węzłami wykonawczymi, aby zapewnić efektywne przetwarzanie.
3. Zarządzanie awariami: JobManager jest także odpowiedzialny za wykrywanie i obsługę awarii w systemie. W przypadku awarii węzła wykonawczego JobManager podejmuje odpowiednie działania, takie jak przydzielenie nowego zadania do innego węzła wykonawczego lub ponowne uruchomienie zadania od checkpointu.

Apache Flink - JobManager

Apache Flink - TaskManager

TaskManager w Apache Flink to komponent wykonawczy, który odpowiada za wykonywanie konkretnych zadań (tasków) przetwarzania danych.

TaskManager jest uruchamiany na węźle w klastrze przetwarzania i odbiera zadania przetwarzania od JobManagera, wykonuje je i zwraca wyniki.

Apache Flink - TaskManager

Główne funkcje i zadania, którymi zajmuje się TaskManager:

1. Wykonywanie zadań: TaskManager otrzymuje od JobManagera zadania przetwarzania w postaci grafu zależności. Każde zadanie składa się z jednego lub więcej subtasków, które można wykonywać równolegle. TaskManager uruchamia subtaski i wykonuje przetwarzanie danych zgodnie z planem wykonania.
2. Komunikacja i synchronizacja: TaskManager komunikuje się z innymi TaskManagerami w klastrze w celu przesyłania danych między subtaskami.
3. Raportowanie postępu i metryk: TaskManager zbiera informacje o postępie przetwarzania, takie jak liczba przetworzonych rekordów, opóźnienia, błędy itp. Przesyła te informacje do JobManagera w celu monitorowania postępu zadania.

Apache Flink - TaskManager

Zasoby wykonawcze w Flink są definiowane za pomocą Slotów zadań. Każdy TaskManager ma jeden lub więcej slotów zadań, z których każdy może uruchamiać jeden potok równoległych zadań.

Zawsze musi istnieć co najmniej jeden menedżer zadań. Najmniejszą jednostką planowania zasobów w Menedżerze zadań jest miejsce na zadania. Liczba przedziałów zadań w Menedżerze zadań wskazuje liczbę jednoczesnych zadań przetwarzania.

Apache Flink - TaskManager

Apache Flink - Task slot

Każdy proces roboczy (TaskManager) jest procesem JVM i może wykonywać jedno lub więcej podzadań w osobnych wątkach. Aby kontrolować, ile zadań akceptuje TaskManager, ma on tak zwane sloty zadań (przynajmniej jeden).

Każdy slot zadań reprezentuje stały podzbiór zasobów TaskManagera. Na przykład TaskManager z trzema gniazdami przeznaczy 1/3 swojej zarządzanej pamięci na każdy slot. Rozmieszczenie zasobów oznacza, że ​​podzadanie nie będzie konkurować z podzadaniami innych zadań o pamięć zarządzaną, ale zamiast tego ma pewną ilość zarezerwowanej pamięci zarządzanej. Zwróć uwagę, że nie występuje tutaj izolacja procesora; obecnie sloty oddzielają tylko zarządzaną pamięć zadań.

Apache Flink

JobManager i TaskManager mogą być uruchamiane na różne sposoby: bezpośrednio na maszynach jako samodzielny klaster, w kontenerach lub zarządzane przez struktury zasobów.

Każde zadanie ma osobnego JobManagera. Zaawansowana konfiguracja może mieć wielu menedżerów zadań, z których jeden jest zawsze liderem, a pozostali sie mu podporządkowują.

Apache Flink - zadanie

Zadanie Flink jest najpierw w stanie created, następnie przełącza się na running, a po zakończeniu całej pracy przełącza się na finished.

W przypadku niepowodzenia zadanie najpierw przełącza się na failing, w którym anuluje wszystkie uruchomione zadania. Jeśli wszystkie wierzchołki zadania osiągnęły stan końcowy, a zadania nie można zrestartować, zadanie przechodzi na failed.

Jeśli zadanie można zrestartować, przejdzie w stan ponownego uruchomienia. Gdy zadanie zostanie całkowicie zrestartowane, osiągnie na nowo stan created.

PyFlink

PyFlink to interfejs API języka Python dla Apache Flink, który umożliwia tworzenie skalowalnych obliczeń wsadowych i strumieniowych, takich jak potoki przetwarzania danych w czasie rzeczywistym.

PyFlink jest dostępny w PyPi i można go zainstalować w następujący sposób:

python -m pip install apache-flink==1.17.1

PyFlink

PyFlink umożliwia użytkownikom tworzenie zadań Flink w Pythonie. Flink udostępnia trzy typy interfejsów API:

- API SQL (umożliwia pisanie zapytań relacyjnych w sposób podobny do używania SQL)
- API table (jak SQL ale logika jest wykonywana po serii optymalizacji po stronie serwera)
- API DataStream (zapewnia kontrolę na niższym poziomie nad podstawowymi blokami konstrukcyjnymi Flink, stanem i czasem)