JobManager

JobManager otrzymuje JobGraph, który jest reprezentacją przepływu danych składającą się z operatorów (JobVertex) i wyników pośrednich (IntermediateDataSet). JobManager przekształca JobGraph w ExecutionGraph. ExecutionGraph jest równoległą wersją JobGraph: dla każdego JobVertex zawiera ExecutionVertex na równoległe podzadanie. Operator z równoległością 100 będzie miał jeden JobVertex i 100 ExecutionVertices.

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.

JobManager ma szereg obowiązków związanych z koordynacją rozproszonego wykonywania Aplikacji Flink: decyduje, kiedy zaplanować następne zadanie (lub zestaw zadań), reaguje na zakończone zadania lub niepowodzenia wykonania, koordynuje punkty kontrolne i koordynuje odzyskiwanie po awariach. Ten proces składa się z trzech różnych elementów:

Menedżer zasobów jest odpowiedzialny za de/alokację zasobów i przydzielanie ich w klastrze Flink.

Scheduler zapewnia interfejs REST do przesyłania aplikacji Flink do wykonania i uruchamia nowy JobMaster dla każdego przesłanego zadania. Uruchamia również Flink WebUI, aby dostarczać informacji o wykonywaniu zadań.

JobMaster jest odpowiedzialny za zarządzanie wykonaniem pojedynczego JobGraph. Wiele zadań może działać jednocześnie w klastrze Flink, z których każde ma swój własny JobMaster.

Zawsze jest co najmniej jeden JobManager. Zaawansowana konfiguracja może mieć wielu menedżerów zadań, z których jeden jest zawsze liderem, a pozostali sie mu podporządkowują.

Task Manager

Menedżerowie zadań (nazywani również workerami) wykonują zadania przepływu danych oraz buforują i wymieniają strumienie danych.

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.

Task Slots

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ń.

Dostosowując liczbę przedziałów zadań, użytkownicy mogą określić, w jaki sposób podzadania są od siebie odizolowane. Posiadanie jednego gniazda na menedżera zadań oznacza, że ​​każda grupa zadań działa w oddzielnej maszynie wirtualnej (którą można na przykład uruchomić w osobnym kontenerze). Posiadanie wielu gniazd oznacza, że ​​więcej podzadań współużytkuje tę samą maszynę JVM.

SQL interface

Pozwala wykorzystywać standardowy SQL do wykonywania zapytań na danych.

Python Table API

Python Table API umożliwia użytkownikom procesowanie danych poprzez wykorzystanie metod funkcyjnych jak select(), group_by(), where() które są automatycznie tłumaczone na niskopoziomowe metody.

Oba interfejsy Table API oraz SQL korzystają ze wspólnego modelu danych, którym są tabele. Wyrażenia na tablicach są "lazy evaluated" (czyli operacje w strumieniu wykonywane są dopiero w momencie, gdy dane są nam potrzebne na wyjściu).

Wejściowym obiektem na którym będziemy pracować będzie tzw. "TableEnvironment". W ramach tego środowiska będziemy rejestrować tabele i źródła danych, wykonywać zapytania oraz uruchamiać dedykowane funkcje.

EnvironmentSettings oraz TableEnvironment z modułu pyflink.table będą wykorzystywane do tworzenia TableEnvironment dla zadań batchowych.

TableEnvironment

Żeby utworzyć zadanie we Flink musimy utworzyć "środowisko" tabel. Do wyboru mamy dwie opcje przetwarzanie batchowe i strumieniowe. Na tych zajęciach skupimy się na przetwarzaniu batchowym (wsadowym).

    env_settings = EnvironmentSettings.in_batch_mode()
table = TableEnvironment.create(env_settings)

Zadanie 1

Zobacz przykłady tworzenia środowiska tabel w pliku TableEnv.py.

Model danych

Tabele służą za źródła danych jak i jako obiekty pośrednie przy transformacji danych. Tabele możemy tworzyć zapisując obiekty na sztywno w kodzie jako kolekcje w Pythonie lub odczytując je ze źródeł takich jak pliki, tabele czy też zewnętrzne systemy (jak Kafka).

Tworzenie tabel na postawie zapisanych na sztywno danych :

Pyflink wspiera użytkowników w tworzeniu tabel źródłowych z podanej listy. Poniższy przykład definiuje tabelę zawierającą trzy wiersze danych: [(Hello, 1), ((world, 2), ((Flink, 3)]), która ma dwie kolumny z nazwami kolumn a i B oraz typami varchar i bigint.

tab = t_env.from_elements([("hello", 1), ("world", 2), ("flink", 3)], ['a', 'b'])

Ta metoda jest zwykle używana w fazie testowej, gdyż może szybko utworzyć tabelę źródła danych w celu weryfikacji logiki zadania. Pierwszy parametr służy do określenia listy danych. Każdy element na liście musi być typu krotki; Drugi parametr określa schemat tabeli

Tworzenie tabel za pomocą SQL API.

DDL jest najbardziej zalecanym sposobem definiowania tabel źródeł i ujść danych, a wszystkie konektory obsługiwane w Java Table API & SQL mogą być używane w zadaniach pyflink Table API poprzez DDL. Wśród nich są: Kafka, Upsert Kafka, Firehose, Kinesis, JDBC, Elasticsearch, FileSystem, HBase, DataGen, Print, BlackHole, Hive. Podobnie jak w przypadku tworzenia tabeli źródła danych, można również utworzyć tabelę wynikową za pomocą języka DDL. Wynik przetwarzania możemy zapisywać w plikach, albo przesyłać do zewnętrznych systemów, czy kolejek.

Zadanie 2

Zobacz przykłady tworzenia tabeli z kolekcji w pliku DataSourceExample.py.

Zadanie 3

Zobacz przykłady tworzenia środowiska tabel z pliku CSV w pliku FileSourceCSV.py.

Zadanie 4

Zobacz przykłady tworzenia środowiska tabel źródła danych i ujścia danych z pliku CSV w pliku FileSink.py. Zobacz co stanie się, gdy nie wykonamy wait() po operacji insert.

Wait if necessary for at most the given time (milliseconds) for the data to be ready. For a select operation, this method will wait until the first row can be accessed locally. For an insert operation, this method will wait for the job to finish

Projekcje

select - działa jak select w SQL, pozwala nam wybierać kolumny z tabel, dodawać, usuwać kolumny, czy też nadawać im aliasy, albo tworzyć nowe na podstawie istniejących danych.

Dodatkowo możemy na wyniku wykonywać znane operacje z SQL takie jak limit(), czy distinct().

Lista funckji SQL z których możemy korzystać jest dostępna tutaj (w postaci API SQL i Table API) :

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/functions/systemfunctions/

Zadanie 5

Zobacz przykłady selectów w pliku Select.py.

Filtry

where - umożliwia filtrowanie wierzy za pomocą logicznego warunku, podobną rolę spełnia element DataStream api o nazwie filter

Zadanie 6

Zobacz przykłady selectów w pliku Filtering.py.

Join

join - służy do łączenia dwóch tabel w jedną tabelę. Uwaga! Flink nie radzi sobie z konfliktami nazw i w związku z tym nie możemy mieć tej samej nazwy kolumny w obu tabelach!

Zadanie 7

Zobacz przykłady selectów w pliku Join.py.

Aggregacje

group_by - służy do agregacji wierszy i tworzenia podsumowań takich jak średnie, zliczenie (count) czy sumowanie

Zadanie 8

Zobacz przykłady selectów w pliku Aggregations.py.