Title: Przetwarzanie równoległe i strumieniowe

PRS

Przetwarzanie równoległe i strumieniowe

Wykład 10

Przetwarzanie danych strumieniowych w Apache Flink - API, model danych

Apache Flink

March 23, 2023 - Leonard Xu (@Leonardxbj)

The Apache Flink PMC is pleased to announce Apache Flink release 1.17.0.

https://flink.apache.org

PyFlink

PyFlink to interfejs API Pythona dla Apache Flink, który umożliwia tworzenie skalowalnych obliczeń wsadowych i strumieniowych.

Python API depends on Py4J (currently version 0.10.9.7), CloudPickle (currently version 2.2.0), python-dateutil(currently version >=2.8.0,<3), Apache Beam (currently version 2.43.0).

Python version (3.7, 3.8, 3.9 or 3.10) is required for PyFlink

$ python -m pip install apache-flink==1.17.1

Apache Flink - data driven application

https://flink.apache.org/use-cases/

Apache Flink - data analytics application

https://flink.apache.org/use-cases/

Apache Flink - data pipeline application

https://flink.apache.org/use-cases/

Modele przetwarzania w Apache Flink

Przetwarzanie wsadowe to paradygmat w pracy podczas przetwarzania ograniczonego strumienia danych. W tym trybie działania można wybrać pobranie całego zestawu danych przed wygenerowaniem jakichkolwiek wyników, co oznacza, że możliwe jest na przykład posortowanie danych, obliczenie statystyk globalnych lub utworzenie raportu końcowego podsumowującego wszystkie dane wejściowe.

Przetwarzanie strumieniowe obejmuje nieograniczone strumienie danych. Przynajmniej koncepcyjnie, dane wejściowe mogą nigdy się nie kończyć, więc jesteś zmuszony do ciągłego przetwarzania danych w miarę ich nadejścia.

Apache Flink - przetwarzanie

Programy we Flink są z natury równoległe i rozproszone.

Podczas wykonywania strumień ma jedną lub więcej partycji strumienia, a każdy operator ma co najmniej jedno podzadanie operatora.

Podzadania operatora są niezależne od siebie i wykonywane w różnych wątkach i ewentualnie na różnych maszynach lub kontenerach.

Apache Flink - przetwarzanie

Apache Flink data stream

DataStream to immutowalny zbiór danych, które może zawierać duplikaty.

Dane te mogą być ograniczone lub nieograniczone, interfejs API, którego używasz do pracy z nimi, jest taki sam.

Możesz utworzyć początkowy DataStream, dodając źródło w programie Flink. Następnie możesz wyprowadzić z tego nowe strumienie i połączyć je za pomocą metod API, takich jak mapowanie, filtrowanie i tak dalej.

Apache Flink program

Programy Flink wyglądają jak zwykłe programy, które przekształcają strumienie danych. Każdy program składa się z tych samych podstawowych części:

- Uzyskaj środowisko wykonawcze,
- Załaduj/utwórz dane początkowe / źródło danych,
- Określ przekształcenia na tych danych,
- Określ, gdzie umieścić wyniki swoich obliczeń,
- Uruchom wykonanie programu

Apache Flink środowisko

Podstawą przetwarzania jest środowisko uruchomieniowe naszych programów (kontekst, w którym wykonywany jest program).

StreamExecutionEnvironment jest podstawą wszystkich programów Flink. Możesz go uzyskać za pomocą :

env = StreamExecutionEnvironment.get_execution_environment()


Środowisko zapewnia metody kontrolowania wykonywania zadań (takie jak ustawienie parametrów równoległości lub odporności na błędy/punkty kontrolne) oraz interakcji ze światem zewnętrznym (dostęp do danych).

Apache Flink środowisko

Domyślny typ przetwarzania dla aplikacji Flink to przetwarzanie strumieniowe.

Włączając wykonanie BATCH, pozwalamy Flink na zastosowanie dodatkowych optymalizacji, które możemy wykonać tylko wtedy, gdy wiemy, że nasze dane wejściowe są ograniczone.

env = StreamExecutionEnvironment.get_execution_environment()

env.set_runtime_mode(RuntimeExecutionMode.BATCH)

Apache Flink środowisko

W trybie wykonywania STREAMING wszystkie zadania muszą być cały czas online/działające. Dzięki temu Flink może natychmiast przetwarzać nowe rekordy przez cały potok, co jest nam potrzebne do ciągłego przetwarzania strumienia o niskim opóźnieniu.

W trybie wykonywania BATCH zadania zadania można podzielić na etapy, które można wykonywać jeden po drugim. Możemy to zrobić, ponieważ dane wejściowe są ograniczone, a zatem Flink może w pełni przetworzyć jeden etap potoku przed przejściem do następnego.

Apache Flink źródło danych

Aby utworzyć DataStream ze źródła, należy przekazać obiekt Source do StreamExecutionEnvironment. Na przykład :

env = StreamExecutionEnvironment.get_execution_environment()

my_source = ...

env.from_source(
    my_source,
    "my_source_name")

Apache Flink źródło danych

Source może oznaczać plik, bazę danych, zewnętrzne źródło danych (jak Kafka) itp.

source = FileSource \
    .for_record_stream_format(StreamFormat.text_line_format(), 'a.csv') \
    .build()

schema = CsvSchema.builder() \
    .add_number_column('id', number_type=DataTypes.BIGINT()) \
    .add_array_column('array', separator='#', element_type=DataTypes.INT()) \
    .set_column_separator(',') \
    .build()

source = FileSource.for_record_stream_format(
    CsvReaderFormat.for_schema(schema), CSV_FILE_PATH).build()

Apache Flink źródło danych

Alternatywnie fromCollection tworzy strumień danych z podanej niepustej kolekcji. Typ strumienia danych jest typem elementów w kolekcji.

env = StreamExecutionEnvironment.get_execution_environment()

data_stream = env.from_collection(collection=[1, 2, 3, 4, 5])

Apache Flink transformacje danych

Operatory przekształcają jeden lub więcej strumieni danych w nowy strumień danych.

Wywołania łańcuchowe mogą łączyć wiele transformacji w wyrafinowane topologie przepływu danych.

Flink oferuje różne API do definicji transformacji na strumieniu danych.

Apache Flink relacyjne API

Apache Flink oferuje dwa relacyjne interfejsy API — Table API i SQL — do ujednoliconego przetwarzania strumieniowego i wsadowego.

Table API to zintegrowany z językiem interfejs API zapytań dla języków Java, Scala i Python, który umożliwia tworzenie zapytań z operatorów relacyjnych, takich jak select, filter i join w bardzo intuicyjny sposób.

Interfejsy Table API i SQL bezproblemowo integrują się ze sobą oraz z interfejsem API DataStream w Flink. Możesz łatwo przełączać się między wszystkimi interfejsami API i bibliotekami, które na nich bazują.

PyFlink

W zależności od wymaganego poziomu abstrakcji, w PyFlink można używać dwóch różnych interfejsów API: PyFlink Table API i PyFlink DataStream API.

SQL API - table environment

from pyflink.table import EnvironmentSettings, TableEnvironment

s_env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(s_env)

# create a streaming TableEnvironment
env_settings = EnvironmentSettings.in_streaming_mode()
table_env = TableEnvironment.create(env_settings)

TableEnvironment utrzymuje mapę katalogów tabel, które są tworzone z identyfikatorem. Każdy identyfikator składa się z 3 części: nazwy katalogu, nazwy bazy danych oraz nazwy obiektu.

SQL API

Instrukcje typu SELECT są określane za pomocą metody sqlQuery() środowiska TableEnvironment.

Metoda zwraca wynik instrukcji SELECT jako tabelę. Tabela może być używana w kolejnych zapytaniach SQL i API tabeli, może zostać przekonwertowana na DataStream lub zapisana w TableSink.

Aby uzyskać dostęp do tabeli w zapytaniu SQL, musi być ona zarejestrowana w TableEnvironment.

Dla wygody Table.toString() automatycznie rejestruje tabelę pod unikalną nazwą w swoim TableEnvironment i zwraca nazwę.

SQL API

Interfejs SQL (sql_query) pozwala nam na tworzenie zapytań typu SQL dla tabel i uruchamianiu ich na strumieniu danych.

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

table.execute_sql(f"""
    CREATE TABLE countries(...)
    """)

codes2 = table.sql_query(f"""
        SELECT code, name AS nazwa, CHARACTER_LENGTH(name) as nameLength
        FROM countries
    """)

PyFlink Table API

PyFlink Table API umożliwia pisanie relacyjnych zapytań w sposób podobny do używania SQL lub pracy z danymi tabelarycznymi w Pythonie.

table = t_env.from_elements(
  [(1, 'ABC'), (2, 'ABCDE')],
  schema=DataTypes.ROW([DataTypes.FIELD('id', DataTypes.INT()),
                        DataTypes.FIELD('name', DataTypes.STRING())]))

table.select(table.id, table.name.alias('nazwa')

table.where(table.id == 1)

table.group_by(table.name).select(table.name, table.id.count.alias('sum')

PyFlink Datastream API

PyFlink DataStream API zapewnia kontrolę niższego poziomu nad podstawowymi blokami konstrukcyjnymi Flink, stanem i czasem, w celu tworzenia bardziej złożonych przypadków użycia przetwarzania strumieniowego.

PyFlink Datastream API

Transformacje DataStream we Flink to zwykłe programy, które realizują transformacje na strumieniach danych (np. filtrowanie, aktualizowanie stanu, definiowanie okien, agregowanie).

# Map
data_stream = env.from_collection(collection=[1, 2, 3, 4, 5])
data_stream.map(lambda x: 2 * x, output_type=Types.INT()).print()

# FlatMap
data_stream = env.from_collection(collection=['hello flink', 'ss compute'])
data_stream.flat_map(lambda x: x.split(' '), output_type=Types.STRING()).print()

# Filter
data_stream = env.from_collection(collection=[0, 1, 2, 3, 4, 5])
data_stream.filter(lambda x: x != 0).print()

PyFlink Datastream API

Z Table & SQL można również użyć do utworzenia DataStream. Możesz najpierw utworzyć tabelę za pomocą złączy Table & SQL, a następnie przekonwertować ją na DataStream (i na odwrót).

t_env.execute_sql("""
        CREATE TABLE my_source (
          a INT,
          b VARCHAR
        ) WITH (
          'connector' = 'datagen',
          'number-of-rows' = '10'
        )
    """)

ds = t_env.to_append_stream(
    t_env.from_path('my_source'),
    Types.ROW([Types.INT(), Types.STRING()]))

Apache Flink - Stany

Podczas gdy wiele operacji w przepływie danych po prostu patrzy na jedno pojedyncze zdarzenie na raz (na przykład wykrywanie stanów alarmowych), niektóre operacje zapamiętują informacje z wielu zdarzeń (na przykład operatory okien, średnie, podsumowania itp.).

Te operacje są nazywane stanowymi.

Apache Flink - Stany

Kilka przykładów operacji stanowych:

- Gdy aplikacja wyszukuje określone wzorce zdarzeń, stan przechowuje sekwencję zdarzeń, które wystąpiły do ​​tej pory.
- Podczas agregowania zdarzeń na minutę / godzinę / dzień stan przechowuje oczekujące agregaty.
- Podczas uczenia modelu uczenia maszynowego w strumieniu punktów danych stan przechowuje bieżącą wersję parametrów modelu.

Flink zachowuje stany, aby móc być odpornym na błędy przy użyciu punktów kontrolnych (checkpoints) i punktów zapisu (savepoints).

Apache Flink - Stany

Interfejsy stanu z kluczem (keyState) zapewniają dostęp do różnych typów stanu, które są objęte zakresem klucza bieżącego elementu wejściowego.

Typ stanu może być używany tylko w KeyedStream, który można utworzyć za pomocą stream.keyBy(…) w Java/Scala API lub stream.key_by(…) w Python API.

words = # type: DataStream[Row]
keyed = words.key_by(lambda row: row[0])

Apache Flink - Stany

Apache Flink - Stany

Aby uzyskać uchwyt stanu, musisz utworzyć StateDescriptor. Zawiera on nazwę stanu, typ wartości, które stan przechowuje i czasem funkcje użytkownika - np. określona funkcja, taka jak ReduceFunction.

W zależności od typu stanu, który chcesz pobrać, tworzysz ValueStateDescriptor, ListStateDescriptor, AggregatingStateDescriptor, ReducingStateDescriptor lub MapStateDescriptor.

Apache Flink ujście danych

W zależności od stosowanego API wyjściem programu może być tabela lub strumień.

Tabela jest emitowana do obiektu implementującego TableSink. TableSink to ogólny interfejs obsługujący szeroką gamę formatów plików (np. CSV, Apache Parquet, Apache Avro), systemów pamięci masowej (np. JDBC, Apache HBase, Apache Cassandra, Elasticsearch) lub systemów przesyłania wiadomości (np. Apache Kafka, RabbitMQ).

// compute a result Table using Table API operators and/or SQL queries
Table result = ...;

// Prepare the insert into pipeline
TablePipeline pipeline = result.insertInto("CsvSinkTable");

Apache Flink ujście danych

Ujścia danych zużywają strumienie danych i przesyłają je do plików, gniazd, systemów zewnętrznych lub drukują je. Flink ma wiele wbudowanych formatów wyjściowych (nie wszystkie dostepne w python):

- writeAsText() / TextOutputFormat — Zapisuje elementy liniowo jako łańcuchy. Łańcuchy są uzyskiwane przez wywołanie metody toString() każdego elementu.
- writeAsCsv(...) / CsvOutputFormat — zapisuje krotki jako pliki wartości oddzielonych przecinkami. Ograniczniki wierszy i pól można konfigurować. Wartość dla każdego pola pochodzi z metody toString() obiektów.

Apache Flink ujście danych

- print() / printToErr() — Drukuje wartość toString() każdego elementu w standardowym wyjściu / standardowym strumieniu błędów. Opcjonalnie można podać prefiks (msg), który jest dołączany do danych wyjściowych. Może to pomóc w rozróżnieniu różnych wezwań do drukowania. Jeśli równoległość jest większa niż 1, dane wyjściowe zostaną również poprzedzone identyfikatorem zadania, które wygenerowało dane wyjściowe.
- writeUsingOutputFormat() / FileOutputFormat — metoda i klasa bazowa dla niestandardowych plików wyjściowych. Obsługuje niestandardową konwersję obiektów na bajty.

Apache Flink ujście danych

- writeToSocket — zapisuje elementy do gniazda zgodnie ze schematem SerializationSchema
- addSink — Wywołuje niestandardową funkcję ujścia. Flink jest dostarczany w pakiecie ze złączami do innych systemów (takich jak Apache Kafka), które są zaimplementowane jako funkcje ujścia.