Table API i SQL

Table API to najwyższy poziom abstrakcji, który pozwala pisać zapytania zbliżone do SQL. Można go używać również poprzez Flink SQL.

Charakterystyka:

• deklaratywne API, przypominające zapytania do baz danych,

• wspiera optymalizację zapytań (Calcite),

• ułatwia pracę z systemami BI i hurtowniami danych.

Przykład w Table API (Java):

Table table = tableEnv.fromDataStream(stream, $("value"));
Table filtered = table.filter($("value").like("a%")).select($("value").upperCase());

Przykład w SQL:

SELECT UPPER(value)
FROM stream
WHERE value LIKE 'a%'

Zalety:

• szybki rozwój aplikacji strumieniowych bez skomplikowanej logiki,

• integracja z BI (np. Apache Superset, Grafana),

• kompatybilne z JDBC, Hive, katalogami danych.

Wady:

• mniejsza elastyczność,

• trudniejsze do debugowania w złożonych scenariuszach,

• ograniczony dostęp do szczegółów (np. watermarków, timerów).

DataStream API

DataStream API to niskopoziomowe, typowane API do pracy ze strumieniami danych.

Charakterystyka:

• API w Javie i Scali (Python w PyFlink),

• operacje: map, filter, keyBy, window, reduce, process, itd.,

• obsługuje zarówno przetwarzanie w czasie rzeczywistym, jak i wsadowe (od Flink 1.12+),

• elastyczne i wydajne, ale wymagające większej liczby linii kodu.

Przykład:

DataStream<String> stream = env.fromElements("a", "b", "c");
stream
    .filter(x -> x.startsWith("a"))
    .map(String::toUpperCase)
    .print();

Zalety:

• pełna kontrola nad strumieniem danych,

• obsługa stanów, timerów, watermarks.

Wady:

• wymaga pisania większej ilości kodu,

• mniej deklaratywne niż SQL lub Table API.

ProcessFunction API

ProcessFunction to niskopoziomowe API w ramach DataStream, zapewniające maksymalną kontrolę nad przetwarzaniem.

Charakterystyka:

• dostęp do timestampów, watermarks, timerów,

• możliwość używania state (ValueState, ListState, MapState),

• wspiera side outputs (boczne strumienie),

• przeznaczona do implementacji niestandardowych operatorów i logiki czasowej.

Przykład:

public class MyProcessFunction extends ProcessFunction<String, String> {
    @Override
    public void processElement(String value, Context ctx, Collector<String> out) {
        if (value.equals("alert")) {
            ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 5000);
        }
        out.collect("Odebrano: " + value);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
        out.collect("Timer wyzwolony: " + timestamp);
    }
}

Zalety:

• pełna elastyczność i niskopoziomowa kontrola,

• niezbędne w przypadkach, gdy potrzebna jest niestandardowa logika przetwarzania zdarzeń.

Wady:

• złożoność i ryzyko błędów logicznych,

• większy narzut kodu i mniejsza czytelność.

Funkcja onTimer w Apache Flink jest wywoływana automatycznie, gdy wcześniej zarejestrowany timer (np. przez registerProcessingTimeTimer() lub registerEventTimeTimer()) zostanie wyzwolony.

Typowe zastosowania onTimer:

detekcja nieaktywnych użytkowników po czasie (np. brak aktywności przez 10 minut),

opóźnione alerty: np. "jeśli nie otrzymam OK w ciągu 30 sekund → alarm",

koniec czasu sesji lub okna (własna logika okienkowania),

odroczenie zapisu danych do bazy,

czyszczenie starego stanu (state.clear()).

Exercise

Uruchom przykłady z klasy KeyedProcessFunctionTimerExample.

Porównanie API

Kiedy używać którego API?

• Table API / SQL: gdy chcesz pisać deklaratywnie, np. zapytania analityczne, raporty, integracja z BI.

• DataStream API: gdy potrzebujesz programowalnej kontroli i operacji strumieniowych.

• ProcessFunction: gdy musisz zarządzać czasem, stanem, niestandardową logiką lub zdarzeniami opóźnionymi.

Podstawowa struktura

Aby użyć ProcessFunction, należy zaimplementować klasę dziedziczącą po niej i przeciążyć metodę processElement. Gdy chcemy zarządzać timerami musimy korzystać ze strumienia z kluczem KeyProcessFunction.

import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class MyProcessFunction extends KeyProcessFunction<String, String> {

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        // Główna logika przetwarzania
        out.collect("Przetworzono: " + value);

        // Możliwość użycia kontekstu:
        long timestamp = ctx.timestamp();  // znacznik czasu
        ctx.timerService().registerEventTimeTimer(timestamp + 5000);  // rejestracja timera
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        out.collect("Timer wyzwolony w czasie: " + timestamp);
    }
}

Użycie w potoku danych

Aby użyć tej funkcji w potoku Flinka:

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ProcessFunctionExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> input = env.fromElements("a", "b", "c");

        DataStream<String> output = input.process(new MyProcessFunction());

        output.print();

        env.execute("ProcessFunction Example");
    }
}

Główne możliwości Context

K key = ctx.getCurrentKey();

long triggerTime = ctx.timerService().currentProcessingTime() + 1000;
ctx.timerService().registerProcessingTimeTimer(triggerTime);
long procTime = ctx.timerService().currentProcessingTime();
long watermark = ctx.timerService().currentWatermark();

Long timestamp = ctx.timestamp();

ctx.output(new OutputTag<String>("late-data"){}, "Zdarzenie spóźnione");

final OutputTag<String> sideOutputTag = new OutputTag<String>("side-output"){};

public class SideOutputProcessFunction extends ProcessFunction<String, String> {
    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        if (value.startsWith("x")) {
            ctx.output(sideOutputTag, "Side: " + value);
        } else {
            out.collect("Main: " + value);
        }
    }
}

DataStream<String> mainStream = input.process(new SideOutputProcessFunction());
DataStream<String> sideStream = mainStream.getSideOutput(sideOutputTag);

mainStream.print("main");
sideStream.print("side");

Przykład użycia Context

@Override
public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
    out.collect("Klucz: " + ctx.getCurrentKey());
    out.collect("Timestamp: " + ctx.timestamp());
    out.collect("Czas systemowy: " + ctx.timerService().currentProcessingTime());

    // Przykład użycia side output
    if (value.equals("late"))

Exercise

Uruchom przykłady z klasy Context.

Podsumowanie

ProcessFunction to potężne narzędzie Flinka, które daje pełną kontrolę nad strumieniem. Umożliwia implementację niestandardowej logiki, korzystanie z timerów, przetwarzanie zależne od czasu oraz tworzenie własnych operatorów opartych na stanie.

Warto sięgać po ProcessFunction gdy potrzebujemy: - reagować na zdarzenia czasowe (timery), - zarządzać stanem ręcznie, - kierować dane do wielu wyjść (side outputs), - implementować złożoną logikę warunkową.

Stany w KeyedProcessFunction

W KeyedProcessFunction możemy korzystać z różnych typów stanu:

• ValueState<T> – przechowuje pojedynczą wartość dla danego klucza,

• ListState<T> – przechowuje listę wartości dla danego klucza,

• MapState<K, V> – przechowuje mapę klucz-wartość dla danego klucza.

Przykład: użycie ValueState do detekcji zmian

Ten przykład pokazuje, jak zarejestrować i użyć ValueState w KeyedProcessFunction, aby wykrywać zmiany wartości:

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class ChangeDetectionFunction extends KeyedProcessFunction<String, String, String> {

    private transient ValueState<String> lastSeenState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<String> descriptor = new ValueStateDescriptor<>(
            "lastSeen", String.class);
        lastSeenState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        String last = lastSeenState.value();
        if (last != null && !last.equals(value)) {
            out.collect("Zmieniono z " + last + " na " + value);
        }
        lastSeenState.update(value);
    }
}

Użycie w potoku danych:

env.fromElements("a", "a", "b", "b", "a")
    .keyBy(x -> "stały_klucz")  // W praktyce: keyBy na realnym kluczu
    .process(new ChangeDetectionFunction())
    .print();

Exercise

Uruchom przykłady z klasy SectionSensorAverageJob.

Przykład: użycie ListState do kolekcjonowania elementów

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class BufferingFunction extends KeyedProcessFunction<String, String, String> {

    private transient ListState<String> buffer;

    @Override
    public void open(Configuration parameters) {
        ListStateDescriptor<String> descriptor = new ListStateDescriptor<>(
            "buffered-elements", String.class);
        buffer = getRuntimeContext().getListState(descriptor);
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        buffer.add(value);
        int count = 0;
        for (String ignored : buffer.get()) {
            count++;
        }
        if (count >= 3) {
            StringBuilder result = new StringBuilder("Buffer zawiera: ");
            for (String s : buffer.get()) {
                result.append(s).append(", ");
            }
            out.collect(result.toString());
            buffer.clear();
        }
    }
}

Użycie z keyBy:

env.fromElements("a", "b", "c", "d", "e", "f")
    .keyBy(x -> "klucz1")
    .process(new BufferingFunction())
    .print();

Podsumowanie

KeyedProcessFunction w połączeniu ze stanem pozwala budować złożone aplikacje, które:

• reagują na zmiany danych w czasie (detekcja zmian, buforowanie),

• zachowują kontekst historyczny (np. ostatnie zdarzenia),

• implementują alarmy lub alerty czasowe,

• wykonują przetwarzanie warunkowe zależne od stanu i czasu.

To elastyczny i bardzo wydajny sposób przetwarzania danych w czasie rzeczywistym w Apache Flink.

Podstawowa sygnatura

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window>
    extends AbstractRichFunction
    implements Function {

    public abstract void process(
        KEY key,
        Context context,
        Iterable<IN> elements,
        Collector<OUT> out
    ) throws Exception;

    // Można też nadpisać:
    public void clear(Context context) throws Exception;
}

Parametry

• IN – typ danych wejściowych w oknie,

• OUT – typ danych wyjściowych,

• KEY – typ klucza (np. userId, kraj),

• W – typ okna (np. TimeWindow, GlobalWindow).

Cechy

• Działa na zakończenie okna – jest wywoływana raz dla każdego klucza po zamknięciu okna.

• Otrzymuje wszystkie elementy w danym oknie jako Iterable<IN>.

• Umożliwia dostęp do kontekstu (np. czas końca okna, stan operatora, timerów).

• Może być używana razem z TumblingWindows, SlidingWindows, SessionWindows, CountWindows.

Przykład użycia

dataStream
    .keyBy(value -> value.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .process(new ProcessWindowFunction<MyEvent, String, String, TimeWindow>() {
        @Override
        public void process(String key, Context context, Iterable<MyEvent> events, Collector<String> out) {
            long windowEnd = context.window().getEnd();
            int count = 0;
            for (MyEvent event : events) {
                count++;
            }
            out.collect("Użytkownik: " + key + ", liczba zdarzeń: " + count + ", okno kończy się: " + windowEnd);
        }
    });

Dostępne metody w Context

context.window()       // zwraca obiekt okna (np. TimeWindow)
context.currentWatermark()
context.globalState()  // stan operatora
context.windowState()  // stan okna (dzielony między okna)

Porównanie z innymi API

Zastosowania

• Liczenie zdarzeń per użytkownik w oknie czasu,

• Grupowanie i sortowanie elementów przed emisją,

• Emitowanie danych z kontekstem (np. znacznik czasu okna),

• Zapisywanie danych do zewnętrznych źródeł tylko raz per okno.

Podsumowanie

ProcessWindowFunction to potężne narzędzie w Apache Flink, pozwalające na dokładne sterowanie tym, co, kiedy i jak emitujemy z okien. Używaj go, gdy standardowe reduce() lub aggregate() są zbyt ograniczone.

Opis

Biblioteka flink-test-utils zawiera zestaw narzędzi wspomagających testowanie aplikacji Flink. Zawiera klasy, metody i środowiska testowe, które umożliwiają:

• pisanie testów jednostkowych i integracyjnych dla operatorów strumieniowych,

• symulowanie środowiska wykonawczego Flinka (np. StreamExecutionEnvironment),

• testowanie operatorów stanowych, funkcji z timerami, i side outputów,

• używanie testowych implementacji kontekstu, takich jak TestHarness, OneInputStreamOperatorTestHarness, KeyedProcessFunctionTestHarnessAdapter, itd.

Zakres <scope>test</scope> oznacza, że ta biblioteka jest używana tylko podczas testów (np. w JUnit, TestNG) i nie będzie włączana do finalnej aplikacji produkcyjnej.

Główne komponenty

1. TestHarness-y

   Umożliwiają testowanie zachowania operatorów Flinka w sposób deterministyczny i powtarzalny.

   Przykłady:

   OneInputStreamOperatorTestHarness

   KeyedOneInputStreamOperatorTestHarness

2. Testowe środowisko wykonawcze - Możliwość inicjalizacji mini-środowiska Flinka w teście bez uruchamiania klas głównych. - Przykład: StreamExecutionEnvironment.createLocalEnvironment() w połączeniu z testami.

3. Testowanie funkcji z timerami - Biblioteka pozwala symulować zegary (ProcessingTimeService) i rejestrować timery w ProcessFunction.

4. Testowanie side outputów - Dzięki OutputTag i testowym adapterom możliwe jest testowanie wielu wyjść operatora.

Przykładowe użycie

@Test
public void testWithHarness() throws Exception {
    KeyedProcessFunction<String, String, String> function = new MyFunction();

    KeyedProcessFunctionTestHarnessAdapter<String, String, String> harness =
        KeyedProcessFunctionTestHarnesses.forKeyedProcessFunction(
            function,
            value -> value.split(":")[0],
            Types.STRING
        );

    harness.processElement("s1:100", 1000);
    harness.advanceProcessingTime(2000);

    List<String> output = harness.extractOutputValues();
    // assertions
}

Wersja

Zalecana jest zgodność wersji z rdzeniem Flinka, np.:

• Dla Flinka 1.20.1: <version>1.20.1</version>

Podsumowanie

flink-test-utils to narzędzie niezbędne do profesjonalnego testowania aplikacji Flink. Umożliwia testy jednostkowe bez uruchamiania rzeczywistego klastra Flinka, z dokładną kontrolą nad czasem, stanem i wyjściem operatorów.

Exercise

Uruchom przykłady z klasy TestExample.