Datastream

Strumienie danych w bibliotece Flink nazywamy DataStream - strumieniami danych. Służą one do opisu zarówno danych do przetwarzania strumieniowego jak i batchowego, które w bibliotece Flink posiadają ten sam interfejs do wykonywania operacji na danych.

Przetwarzanie strumieni danych DataStream we Flink to zwykłe programy, które realizują transformacje na strumieniach danych (np. filtrowanie, aktualizowanie stanu, definiowanie okien, agregowanie). Strumienie danych są początkowo tworzone z różnych źródeł (np. kolejki komunikatów, strumienie gniazd, pliki). Wyniki są zwracane przez ujścia (sink), które mogą na przykład zapisywać dane do plików lub na standardowe wyjście (na przykład terminal wiersza poleceń). Programy Flink działają w różnych kontekstach, samodzielnie lub osadzone w innych programach. Wykonanie może odbywać się w lokalnej maszynie JVM lub w klastrach wielu maszyn.

Struktura programu w Apache Flink

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

Środowisko uruchomieniowe

StreamExecutionEnvironment jest podstawą wszystkich programów Flink. W naszym przypadku będziemy korzystać z getExecutionEnvironment(), ponieważ wywołuje on potrzebną funkcję w zależności od kontekstu: jeśli wykonujesz swój program w IDE lub jako zwykły program Java, utworzy lokalne środowisko, które wykona twój program na twojej lokalnej maszynie. Ś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).

Przykład tworzenia środowiska i ustawiania / odczytu paramertów można znaleźć poniżej:

    // Tworzenie środowiska wykonawczego DataStream API
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Przełączenie trybu przetwarzania na wsadowy (Batch)
import org.apache.flink.api.common.RuntimeExecutionMode;

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

// Przykład prostego DataStream - batch
DataStream<Integer> data = env.fromElements(1, 2, 3, 4, 5);
data.print();
env.execute();

Exercise

Uruchom przykłady w klasach BatchEnvironment.java i StreamEnvironment.java.

Tryb wykonywania (runtime_mode) STREAMING to sposób wykonywania programów w zależności czy dane są skończone czy też nie. Streaming powinno być używane w przypadku zadań bez ograniczeń, które wymagają ciągłego przetwarzania przyrostowego i oczekuje się, że pozostaną w trybie online przez nieokreślony czas.

Ponadto istnieje tryb wykonywania w stylu wsadowym, który nazywamy trybem wykonywania BATCH. Wykonuje on zadania w sposób bardziej przypominający ramy przetwarzania wsadowego, takie jak MapReduce. Batch powinno być używane w przypadku zadań ograniczonych, dla których dane wejściowe są znane i ograniczone.

Trzecia opcja to AUTOMATIC pozwalająca wybrać tryb przetwarzania bazując na typie źródła danych.

Zunifikowane podejście Apache Flink do przetwarzania strumieniowego i wsadowego oznacza, że aplikacja DataStream wykonywana na ograniczonych danych wejściowych będzie generować takie same wyniki końcowe niezależnie od skonfigurowanego trybu wykonywania.

Ważne jest, aby zauważyć, co oznacza tutaj final: zadanie wykonywane w trybie STREAMING może generować przyrostowe aktualizacje (pomyśl o upsertach w bazie danych), podczas gdy zadanie BATCH dałoby tylko jeden końcowy wynik na końcu.

Źródła danych

Najprostszym źródłem danych jest kolekcja. Tworzy ono źródło strumienia. Najprostsze przetwarzanie wypisze ten strumień na ekran. Uwaga! Aby wywołać uruchomić przetwarzanie na środowisku musimy wykonać metodę execute().

env.fromElements(1, 2, 3, 4, 5);
env.fromData(List.of("lower", "UPPER", "MiDdLe"));

Jak widać różnica w tm wypadku to sposób wypisywania danych w przypadku strumieniowego podejścia widzimy też na ekranie identyfikator jednostki przetwarzającej zadanie.

Generycznym typem źródła danych jest obiekt Source: Przykładem Source mogą być pliki, inne źródła danych jak Kafka itp. Poniżej plik cities.csv : Program czytający plik jako wejście dla przetwarzania. Definiujemy w obiekcie schemat danych które są odczytywane jako wiersze do strumienia wejściowego. Uwaga program czyta plik więc może być uruchomiony tylko lokalnie!

FileSource
.forRecordStreamFormat(csvFormat, new Path("cities.csv"))
.build();

Funkcje TableEnvironment

• Tworzenie tabel z danych (fromValues, fromDataStream)

• Wykonywanie zapytań SQL (sqlQuery)

• Wykonywanie poleceń DDL i DML (executeSql)

• Wstawianie danych do tabel (executeInsert)

• Rejestrowanie widoków tymczasowych (createTemporaryView)

• Konfiguracja środowiska (getConfig)

Tworzenie TableEnvironment

Istnieją dwa podstawowe tryby tworzenia TableEnvironment:

1. Tryb niezależny (tylko Table API lub SQL API):

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()  // lub inBatchMode()
    .build();

TableEnvironment tableEnv = TableEnvironment.create(settings);

2. Zintegrowany z DataStream API (`StreamTableEnvironment`):

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

Uwaga: StreamTableEnvironment używamy, gdy chcemy łączyć Table API z DataStream API.

Przykład użycia table API:

Table result = tableEnv.sqlQuery("SELECT id, name FROM people WHERE id > 10");
result.execute().print();

Podsumowanie

TableEnvironment pełni kluczową rolę w pracy z danymi tabelarycznymi w Apache Flink. Jest to podstawowe narzędzie do pracy z zapytaniami SQL, definicjami tabel oraz integracją z różnymi źródłami danych (np. pliki, bazy danych, Kafka).

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.

Tabele mogą być wirtualne (WIDOK) lub zwykłe (TABELA). WIDOKI można tworzyć z istniejącego obiektu tabeli, zazwyczaj w wyniku działania interfejsu API tabeli lub zapytania SQL. TABELE opisują dane zewnętrzne, takie jak plik, tabela bazy danych lub kolejka komunikatów.

Możemy swobodnie przechodzić między tabelami a strumieniami i korzystać na zmianę z obi API.

Exercise

Uruchom przykłady z pakietu tableAPI.

Map

dataStream.map(value -> value * 2)

Przekształca każdy element wejściowy na dokładnie jeden element wyjściowy. Jest to najprostsza forma transformacji.

FlatMap

dataStream.flatMap((value, out) -> {
    for (String word : value.split(" ")) {
        out.collect(word);
    }
})

Przekształca każdy element wejściowy na zero, jeden lub wiele elementów wyjściowych.

Filter

dataStream.filter(value -> value > 0)

Filtruje elementy strumienia na podstawie warunku logicznego.

KeyBy

dataStream.keyBy(value -> value.getKey())

Grupuje elementy na podstawie wartości klucza. Operacje takie jak reduce, window, aggregate mogą być wykonywane tylko na pogrupowanych danych.

Reduce

dataStream
    .keyBy(value -> value.f1)
    .reduce((a, b) -> Tuple2.of(a.f0 + b.f0, a.f1))

Agreguje dane ze wspólnym kluczem w sposób inkrementalny.

Union

stream1.union(stream2)

Łączy wiele strumieni tego samego typu w jeden strumień.

Partitioning

dataStream.partitionCustom((key, numPartitions) -> key % numPartitions, value -> value.getKey())

Pozwala na ręczne określenie przypisania elementów do partycji.

Sink

dataStream.print();
dataStream.addSink(new MySinkFunction());

Operator końcowy, który zapisuje przetworzone dane do zewnętrznego systemu (konsola, plik, Kafka, itp.).

Podsumowanie

Operatory DataStream w Flinku pozwalają tworzyć elastyczne i złożone potoki przetwarzania danych czasu rzeczywistego. Umożliwiają zarówno przekształcenia stateless (np. map, filter), jak i stateful (np. keyBy, reduce, window).

Więcej: https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/datastream/operators/

Customowe Operatory

Operatory możemy także definiować sami implementując odpowiednie clasy implementujące interfejsy, jako funkcje lub lambdy.