Задача, ради которой всё затевается

Мошенник получил данные карты и проверяет их серией мелких платежей. Если система заметит серию через пять минут, деньги уже ушли. Если через секунду, банк успевает отклонить следующий платёж. Весь курс мы строим такую систему и называем её PayGuard.

У этой задачи три свойства, которые определяют выбор инструмента. Данные не заканчиваются, поэтому нельзя дождаться конца файла и посчитать результат. Решение зависит от истории карты за последние минуты, поэтому система обязана помнить прошлое. Ошибка стоит денег в обе стороны, поэтому потерянный или задвоенный платёж недопустим даже при падении сервера.

Apache Flink создан под такие условия. Это движок обработки потоков с состоянием, то есть программа на нём получает события по одному и держит между ними данные, которые переживают падения.

Шаттл и траволатор

Представь аэропорт. До выхода на посадку можно добраться автобусом-шаттлом, который отправляется каждые пять минут и забирает всех, кто успел подойти. А можно встать на траволатор, и он повезёт тебя в ту же секунду, как ты на него шагнул.

Spark Structured Streaming в обычном режиме работает как шаттл. Он копит записи, запускает маленькую пакетную задачу (micro-batch) и выдаёт результат всей порции разом. Flink работает как траволатор. Его операторы запущены постоянно, и запись идёт через них сразу после появления в источнике.

Сдвинь ползунок и посмотри, как интервал шаттла влияет на задержку. Модель упрощённая, в ней нет времени на саму обработку, зато видна причина разницы.

Из этой разницы не следует, что Flink всегда быстрее по записям в секунду. На больших объёмах пакетная обработка очень эффективна, и Spark силён именно там. У Flink задержку определяют логика задания и сетевые буферы, у micro-batch ещё и интервал запуска.

protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {
    DataInputStatus status = inputProcessor.processInput();
    switch (status) {
        case MORE_AVAILABLE:
            if (taskIsAvailable()) {
                return;
            }
            break;
        case NOTHING_AVAILABLE:
            break;
        case END_OF_RECOVERY:

public static final ConfigOption<Duration> BUFFER_TIMEOUT =
        ConfigOptions.key("execution.buffer-timeout.interval")
                .durationType()
                .defaultValue(Duration.ofMillis(100))
                .withDeprecatedKeys("execution.buffer-timeout")
                .withDescription(
                        Description.builder()
                                .text(
                                        "The maximum time frequency (milliseconds) for the flushing of the output buffers. By default "
                                                + "the output buffers flush frequently to provide low latency and to aid smooth developer "
                                                + "experience. Setting the parameter can result in three logical modes:")
                                .list(
                                        text(
                                                "A positive value triggers flushing periodically by that interval"),
                                        text(
                                                FLUSH_AFTER_EVERY_RECORD

Четыре вещи, которые Flink берёт на себя

Написать цикл чтения из Kafka можно и на обычной Java. Flink ценен тем, что решает трудные части за тебя, и курс построен вокруг них.

• Время события. Платёж мог прийти с опозданием из-за сети, а считать его надо в той минуте, когда он случился. Об этом уроки 5 и 6.
• Состояние. Счётчик платежей карты живёт внутри задания, может занимать терабайты и читается без обращения к внешней базе. Об этом уроки 7 и 19.
• Восстановление без потерь. Flink регулярно снимает согласованный снимок всего задания и после падения продолжает с него. Об этом уроки 11 и 12.
• Изменение на ходу. Задание можно остановить со снимком, обновить код, сменить параллелизм и продолжить с того же места. Об этом урок 13.

Когда Flink не нужен

Ночной отчёт по таблице в озере данных Spark посчитает проще и дешевле. Переход не окупится и тогда, когда задержка в минуту никому не мешает и команда уже знает Spark. Простое преобразование топика без состояния часто проще сделать на Kafka Connect или маленьком консьюмере. Flink оправдан там, где одновременно нужны низкая задержка, состояние и корректность при сбоях.

Что мы построим

PayGuard читает платежи из Kafka, чистит их и проверяет правилами, а алерты отправляет обратно в Kafka. Цепочка будет расти от урока к уроку. Сегодня работает только левая часть схемы, источник и печать в лог.

Почему твой конфиг молчит

В папке проекта лежит твой Makefile с образом flink:latest и файлом flink-conf.yaml. Если выполнить make start, контейнеры поднимутся, в логе JobManager не будет ошибок, и при этом REST-интерфейс покажет "taskmanagers":0. Процесс TaskManager исполняет задания, и кластер без него не умеет ничего. Причин две. Первая в том, что flink:latest на момент проверки это Flink 2.3.0, а вторая в самом файле.

public static final String FLINK_CONF_FILENAME = "config.yaml";

// get Flink yaml configuration file
Configuration configuration;
File yamlConfigFile = new File(confDirFile, FLINK_CONF_FILENAME);
if (!yamlConfigFile.exists()) {
    throw new IllegalConfigurationException(
            "The Flink config file '"
                    + yamlConfigFile
                    + "' ("
                    + yamlConfigFile.getAbsolutePath()
                    + ") does not exist.");
} else {

Если перенести те же настройки в правильный файл, всплывут другие проблемы. Таблица показывает, что произойдёт с каждой строкой.

Последняя строка проверена запуском. С твоими значениями памяти TaskManager падает при старте и пишет, что вычисленный размер сетевой памяти равен нулю.

org.apache.flink.configuration.IllegalConfigurationException: TaskManager memory configuration failed:
If Total Flink, Task Heap and (or) Managed Memory sizes are explicitly configured then the Network Memory size
is the rest of the Total Flink memory after subtracting ...
Caused by: Derived Network Memory size (0 bytes) is not in configured Network Memory range [64.000mb (67108864 bytes), ...]

Проверь себя

Что будет в конце

На твоей машине работают Kafka и кластер Flink из одного JobManager и одного TaskManager. Генератор пишет платежи в топик, а задание печатает их в лог. Это скелет, на котором будет расти весь проект.

Шаг 1. Создай проект

Создай папку payguard рядом со своим Makefile, а внутри два файла. Первый задаёт имя проекта. Gradle ставить не обязательно, wrapper можно скопировать из solutions/lesson-01, это папка gradle/ и файлы gradlew и gradlew.bat. Wrapper сам скачает Gradle 9.8.1 при первом запуске.

rootProject.name = "payguard"

Второй файл описывает сборку и зависимости.

plugins {
    java
    id("com.gradleup.shadow") version "9.6.1"
}

group = "dev.study.payguard"
version = "1.0.0"

java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(17)
    }
}

repositories {
    mavenCentral()
}

val flinkVersion = "2.2.1"
val kafkaConnectorVersion = "5.0.0-2.2"
val jacksonVersion = "2.21.0"

// Классы, которые уже лежат в образе Flink: нужны для компиляции и тестов, в jar не попадают.
val flinkProvided: Configuration by configurations.creating
configurations.compileOnly { extendsFrom(flinkProvided) }
configurations.testImplementation { extendsFrom(flinkProvided) }

dependencies {
    flinkProvided("org.apache.flink:flink-streaming-java:$flinkVersion")
    flinkProvided("org.apache.flink:flink-clients:$flinkVersion")
    flinkProvided("org.apache.flink:flink-connector-base:$flinkVersion")

    implementation("org.apache.flink:flink-connector-kafka:$kafkaConnectorVersion")
    implementation("com.fasterxml.jackson.core:jackson-databind:$jacksonVersion")

    runtimeOnly("org.slf4j:slf4j-simple:2.0.17")

    testImplementation(platform("org.junit:junit-bom:6.1.3"))
    testImplementation("org.junit.jupiter:junit-jupiter")
    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
    testImplementation("org.assertj:assertj-core:3.27.7")
}

tasks.test {
    useJUnitPlatform()
}

tasks.shadowJar {
    archiveClassifier = ""
    manifest { attributes["Main-Class"] = "dev.study.payguard.job.PayGuardJob" }
    mergeServiceFiles()
    filesMatching("META-INF/services/**") { duplicatesStrategy = DuplicatesStrategy.INCLUDE }
    dependencies { exclude(dependency("org.slf4j:.*:.*")) }
    exclude("META-INF/*.SF", "META-INF/*.DSA", "META-INF/*.RSA")
}

tasks.build {
    dependsOn(tasks.shadowJar)
}

tasks.register<JavaExec>("loadgen") {
    group = "application"
    description = "Запускает генератор платежей: ./gradlew loadgen --args='--rate 200'"
    classpath = sourceSets.main.get().runtimeClasspath
    mainClass = "dev.study.payguard.loadgen.LoadGenerator"
    systemProperty("org.slf4j.simpleLogger.defaultLogLevel", "warn")
}

Разберём места, где сборка Flink-задания отличается от обычного Java-приложения. Самое важное здесь разделение зависимостей.

// Классы, которые уже лежат в образе Flink: нужны для компиляции и тестов, в jar не попадают.
val flinkProvided: Configuration by configurations.creating
configurations.compileOnly { extendsFrom(flinkProvided) }
configurations.testImplementation { extendsFrom(flinkProvided) }

dependencies {
    flinkProvided("org.apache.flink:flink-streaming-java:$flinkVersion")
    flinkProvided("org.apache.flink:flink-clients:$flinkVersion")
    flinkProvided("org.apache.flink:flink-connector-base:$flinkVersion")

    implementation("org.apache.flink:flink-connector-kafka:$kafkaConnectorVersion")
    implementation("com.fasterxml.jackson.core:jackson-databind:$jacksonVersion")

    runtimeOnly("org.slf4j:slf4j-simple:2.0.17")

Блок tasks.shadowJar собирает толстый jar, в котором лежит твой код и все зависимости из implementation. Строка с Main-Class нужна, чтобы flink run нашёл точку входа без лишних аргументов. Вызовы mergeServiceFiles и filesMatching склеивают файлы META-INF/services из разных библиотек. Без них коннектор теряет регистрацию своих компонентов. Исключение подписей *.SF нужно потому, что подпись чужого jar становится недействительной после пересборки.

Шаг 2. Опиши стенд в Docker Compose

Положи в корень проекта docker-compose.yml.

name: payguard

x-flink-common: &flink-common
  image: flink:2.2.1-java17
  environment:
    FLINK_PROPERTIES: |
      jobmanager.rpc.address: jobmanager
      taskmanager.numberOfTaskSlots: 4
      parallelism.default: 2
      jobmanager.memory.process.size: 1600m
      taskmanager.memory.process.size: 2048m
  volumes:
    - ./build/libs:/jobs:ro
    - ./data:/flink-data

services:
  kafka:
    image: apache/kafka:4.3.2
    ports:
      - "9094:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:9094
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list"]
      interval: 5s
      timeout: 10s
      retries: 20

  kafka-init:
    image: apache/kafka:4.3.2
    depends_on:
      kafka:
        condition: service_healthy
    entrypoint: ["/bin/bash", "-c"]
    command:
      - |
        for topic in payments.raw payments.dlq fraud.alerts rules.config; do
          /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 \
            --create --if-not-exists --topic "$$topic" --partitions 4 --replication-factor 1
        done

  jobmanager:
    <<: *flink-common
    command: jobmanager
    ports:
      - "8081:8081"

  taskmanager:
    <<: *flink-common
    command: taskmanager
    depends_on:
      - jobmanager

Настройки Flink переданы переменной FLINK_PROPERTIES. Скрипт запуска образа дописывает её содержимое в родной config.yaml, поэтому ключи здесь проверенные и файл монтировать не нужно. Папка build/libs смонтирована в контейнер как /jobs, так собранный jar виден JobManager. У Kafka два слушателя. Контейнеры ходят на kafka:9092, а твоя машина на localhost:9094. Служба kafka-init создаёт топики с четырьмя партициями и завершается.

Шаг 3. Модель и генератор платежей

Платёж описан неизменяемым record. Сумма хранится в минимальных единицах валюты в long, чтобы не ловить ошибки округления double.

package dev.study.payguard.model;

/** Платёж по карте: неизменяемый факт, который случился в момент {@code eventTimeMs}. */
public record PaymentEvent(
        String paymentId,
        String cardId,
        String merchantId,
        long amountMinor,
        String currency,
        String country,
        long eventTimeMs) {
}

Генератор создаёт нагрузку, близкую к настоящей. Треть платежей идёт в одного продавца m-0000, часть приходит с опозданием, часть дублируется, часть битая. Эта грязь понадобится нам в следующих уроках.

package dev.study.payguard.loadgen;

import dev.study.payguard.model.PaymentEvent;
import java.util.List;
import java.util.Random;
import java.util.UUID;

/** Создаёт правдоподобные платежи. Random передаётся снаружи, чтобы тесты были повторяемыми. */
final class PaymentFactory {

    private static final int CARD_COUNT = 5_000;
    private static final int MERCHANT_COUNT = 50;
    private static final String HOT_MERCHANT = "m-0000";
    private static final double HOT_MERCHANT_SHARE = 0.3;
    private static final long MAX_LATENESS_MS = 30_000;
    private static final List<String> COUNTRIES = List.of("DE", "FR", "PL", "ES", "NL");

    private final Random random;
    private final double lateShare;

    PaymentFactory(Random random, double lateShare) {
        this.random = random;
        this.lateShare = lateShare;
    }

    PaymentEvent next(long nowMs) {
        return new PaymentEvent(
                new UUID(random.nextLong(), random.nextLong()).toString(),
                "c-%04d".formatted(random.nextInt(CARD_COUNT)),
                nextMerchant(),
                100 + random.nextInt(50_000),
                "EUR",
                COUNTRIES.get(random.nextInt(COUNTRIES.size())),
                nextEventTime(nowMs));
    }

    private String nextMerchant() {
        return random.nextDouble() < HOT_MERCHANT_SHARE
                ? HOT_MERCHANT
                : "m-%04d".formatted(1 + random.nextInt(MERCHANT_COUNT - 1));
    }

    private long nextEventTime(long nowMs) {
        return random.nextDouble() < lateShare
                ? nowMs - (long) (random.nextDouble() * MAX_LATENESS_MS)
                : nowMs;
    }
}

package dev.study.payguard.loadgen;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dev.study.payguard.model.PaymentEvent;
import java.util.Random;

/** Превращает платежи в строки JSON и подмешивает грязь: дубликаты и битые записи. */
final class RawRecordSource {

    static final String BAD_RECORD = "{\"paymentId\": broken";

    private final ObjectMapper mapper = new ObjectMapper();
    private final PaymentFactory factory;
    private final Random random;
    private final double duplicateShare;
    private final double badShare;
    private String previousRecord;

    RawRecordSource(PaymentFactory factory, Random random, double duplicateShare, double badShare) {
        this.factory = factory;
        this.random = random;
        this.duplicateShare = duplicateShare;
        this.badShare = badShare;
    }

    String next(long nowMs) {
        if (random.nextDouble() < badShare) {
            return BAD_RECORD;
        }
        if (previousRecord != null && random.nextDouble() < duplicateShare) {
            return previousRecord;
        }
        previousRecord = toJson(factory.next(nowMs));
        return previousRecord;
    }

    private String toJson(PaymentEvent event) {
        try {
            return mapper.writeValueAsString(event);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("Не удалось сериализовать платёж", e);
        }
    }
}

package dev.study.payguard.loadgen;

import java.util.HashMap;
import java.util.Map;

/** Настройки генератора. Каждая доля задаётся числом от 0 до 1. */
record GeneratorConfig(
        String bootstrapServers,
        String topic,
        int ratePerSecond,
        double lateShare,
        double duplicateShare,
        double badShare) {

    static GeneratorConfig fromArgs(String[] args) {
        Map<String, String> options = parseOptions(args);
        return new GeneratorConfig(
                options.getOrDefault("bootstrap", "localhost:9094"),
                options.getOrDefault("topic", "payments.raw"),
                Integer.parseInt(options.getOrDefault("rate", "100")),
                share(options, "late", 0.05),
                share(options, "dup", 0.0),
                share(options, "bad", 0.0));
    }

    private static Map<String, String> parseOptions(String[] args) {
        if (args.length % 2 != 0) {
            throw new IllegalArgumentException("Ожидаются пары --имя значение");
        }
        Map<String, String> options = new HashMap<>();
        for (int i = 0; i < args.length; i += 2) {
            if (!args[i].startsWith("--")) {
                throw new IllegalArgumentException("Опция должна начинаться с --: " + args[i]);
            }
            options.put(args[i].substring(2), args[i + 1]);
        }
        return options;
    }

    private static double share(Map<String, String> options, String name, double defaultValue) {
        double value = options.containsKey(name) ? Double.parseDouble(options.get(name)) : defaultValue;
        if (value < 0 || value > 1) {
            throw new IllegalArgumentException("--" + name + " должно быть от 0 до 1, получено " + value);
        }
        return value;
    }
}

package dev.study.payguard.loadgen;

import java.time.Duration;
import java.util.Properties;
import java.util.Random;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/** Пишет поток платежей в Kafka с заданной скоростью, пока процесс не остановят. */
public final class LoadGenerator {

    private static final Duration TICK = Duration.ofMillis(100);
    private static final long REPORT_EVERY_RECORDS = 1_000;

    public static void main(String[] args) throws InterruptedException {
        GeneratorConfig config = GeneratorConfig.fromArgs(args);
        Random random = new Random();
        RawRecordSource source = new RawRecordSource(
                new PaymentFactory(random, config.lateShare()), random,
                config.duplicateShare(), config.badShare());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProperties(config))) {
            produceForever(producer, source, config);
        }
    }

    private static void produceForever(
            KafkaProducer<String, String> producer, RawRecordSource source, GeneratorConfig config)
            throws InterruptedException {
        int recordsPerTick = Math.max(1, config.ratePerSecond() / 10);
        long sent = 0;
        while (!Thread.currentThread().isInterrupted()) {
            for (int i = 0; i < recordsPerTick; i++) {
                producer.send(new ProducerRecord<>(config.topic(), source.next(System.currentTimeMillis())));
                if (++sent % REPORT_EVERY_RECORDS == 0) {
                    System.out.println("Отправлено записей: " + sent);
                }
            }
            Thread.sleep(TICK.toMillis());
        }
    }

    private static Properties producerProperties(GeneratorConfig config) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServers());
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return properties;
    }

    private LoadGenerator() {
    }
}

Шаг 4. Первое задание

Создай src/main/java/dev/study/payguard/job/PayGuardJob.java.

package dev.study.payguard.job;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/** Точка входа Flink-задания PayGuard. */
public final class PayGuardJob {

    private static final String BOOTSTRAP_SERVERS = "kafka:9092";
    private static final String RAW_TOPIC = "payments.raw";

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

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(BOOTSTRAP_SERVERS)
                .setTopics(RAW_TOPIC)
                .setGroupId("payguard")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        env.fromSource(source, WatermarkStrategy.noWatermarks(), "payments-raw")
                .print()
                .name("stdout");

        env.execute("payguard");
    }

    private PayGuardJob() {
    }
}

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(BOOTSTRAP_SERVERS)
                .setTopics(RAW_TOPIC)
                .setGroupId("payguard")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        env.fromSource(source, WatermarkStrategy.noWatermarks(), "payments-raw")
                .print()
                .name("stdout");

        env.execute("payguard");

Шаг 5. Собери и запусти

На macOS сначала укажи JDK 17, если по умолчанию у тебя другая версия. Порядок команд здесь не случайный. Если поднять Compose раньше сборки, Docker смонтирует пустую папку build/libs, и jar в контейнере не появится.

export JAVA_HOME=$(/usr/libexec/java_home -v 17)
./gradlew shadowJar
docker compose up -d --wait

Флаг --wait заставляет команду дождаться готовности контейнеров. Теперь спроси у кластера, как он себя чувствует. REST-интерфейс отвечает на порту 8081, там же открывается Web UI.

curl -s localhost:8081/overview | jq .

{
  "taskmanagers": 1,
  "slots-total": 4,
  "slots-available": 4,
  "jobs-running": 0,
  "jobs-finished": 0,
  "jobs-cancelled": 0,
  "jobs-failed": 0,
  "flink-version": "2.2.1",
  "flink-commit": "450c63e"
}

Поля taskmanagers и slots-total сразу показывают то, чего не было при твоём запуске через Makefile. Отправь задание и запусти генератор на пять платежей в секунду.

docker compose exec jobmanager flink run --detached /jobs/payguard-1.0.0.jar
./gradlew loadgen --args="--rate 5 --dup 0.02 --bad 0.01"

Флаг --detached отдаёт терминал сразу после отправки и оставляет задание работать. Вывод print живёт в логе TaskManager, смотри его из второго терминала.

docker compose logs -f taskmanager | grep -E '[0-9]+> '

1> {"paymentId":"3d7afcf2-701e-3e1b-be95-6415e9444aa4","cardId":"c-2982","merchantId":"m-0048","amountMinor":676,"currency":"EUR","country":"NL","eventTimeMs":1794084709909}
1> {"paymentId":"8b470b4d-c563-a975-d7fa-cd92300f2f0f","cardId":"c-0422","merchantId":"m-0000","amountMinor":17305,"currency":"EUR","country":"FR","eventTimeMs":1794084710014}

Число перед > это номер подзадачи плюс один. Открой http://localhost:8081, зайди в задание payguard и посмотри на его граф. В следующем уроке мы подробно разберём, что там нарисовано. Сервисы можно оставить запущенными или остановить командой docker compose down.

Задание. Почини свой старый конфиг

В корне проекта лежит flink-conf.yaml, с которым твой кластер получал ноль TaskManager. Перепиши его так, чтобы Flink 2.2.1 реально применил настройки. Сам исходный файл и Makefile оставь как есть, новый файл положи в conf/config.yaml внутри проекта.

1. Создай файл с правильным именем и ключами Flink 2.x. Сохрани смысл настроек, это 4 слота, параллелизм 2, состояние в куче и каталоги чекпоинтов в /tmp.
2. Подними JobManager и TaskManager обычным docker run в отдельной сети, смонтировав файл в /opt/flink/conf/config.yaml. Образ бери flink:2.2.1-java17.
3. Докажи, что настройки применились, запросами к REST.

Готово, если /overview показывает 1 TaskManager и 4 слота, а /jobmanager/config содержит твои значения.

Подсказка. Родной config.yaml в образе задаёт не только адрес JobManager. Посмотри на него командой docker run --rm flink:2.2.1-java17 cat /opt/flink/conf/config.yaml и подумай, на каком адресе процессы слушают сеть, если ключей bind-host нет.

Ответ

# Старый flink-conf.yaml, переписанный на ключи Flink 2.x.
jobmanager:
  rpc:
    address: jobmanager
    port: 6123
  bind-host: 0.0.0.0
  memory:
    process:
      size: 1600m

taskmanager:
  bind-host: 0.0.0.0
  numberOfTaskSlots: 4
  memory:
    process:
      size: 1728m

rest:
  bind-address: 0.0.0.0
  address: 0.0.0.0

parallelism:
  default: 2

state:
  backend:
    type: hashmap

execution:
  checkpointing:
    dir: file:///tmp/flink-checkpoints
    savepoint-dir: file:///tmp/flink-savepoints

web:
  submit:
    enable: true
  cancel:
    enable: true

docker network create l1net
docker run -d --name l1jm --network l1net --network-alias jobmanager -p 18081:8081 \
  -v "$PWD/conf/config.yaml:/opt/flink/conf/config.yaml:ro" flink:2.2.1-java17 jobmanager
docker run -d --name l1tm --network l1net \
  -v "$PWD/conf/config.yaml:/opt/flink/conf/config.yaml:ro" flink:2.2.1-java17 taskmanager

curl -s localhost:18081/overview | jq -c '{tm: .taskmanagers, slots: .["slots-total"], version: .["flink-version"]}'
curl -s localhost:18081/jobmanager/config \
  | jq -c '[.[] | select(.key | test("state.backend.type|numberOfTaskSlots|checkpointing.dir|parallelism.default"))]'

{"tm":1,"slots":4,"version":"2.2.1"}
[{"key":"parallelism.default","value":"2"},{"key":"taskmanager.numberOfTaskSlots","value":"4"},{"key":"execution.checkpointing.dir","value":"file:///tmp/flink-checkpoints"},{"key":"state.backend.type","value":"hashmap"}]

Глоссарий урока

Apache Flink

Движок для обработки потоков данных с состоянием. Программа на нём работает непрерывно и переживает падения без потери результатов.

Micro-batch

Режим Spark Structured Streaming, при котором записи копятся и обрабатываются небольшими пакетными задачами.

Задание (job)

Программа Flink, отправленная на кластер. В потоковом режиме работает, пока её не остановят.

Оператор (operator)

Один шаг обработки в задании, например источник, фильтр или окно.

Подзадача (subtask)

Один параллельный экземпляр оператора. При параллелизме 4 у оператора четыре подзадачи.

JobManager

Процесс, который управляет заданиями, раздаёт работу и координирует снимки. Подробно в уроке 2.

TaskManager

Рабочий процесс, в котором выполняются подзадачи. Похож на экзекутор Spark.

Слот (task slot)

Единица ресурсов TaskManager, в которой выполняются подзадачи. Их число задаёт ключ taskmanager.numberOfTaskSlots.

Состояние (state)

Данные, которые оператор хранит между записями, например счётчик платежей карты.

Сетевой буфер (network buffer)

Участок памяти, куда складываются записи перед отправкой другому TaskManager. Уходит заполненным или по таймауту.

Apache Kafka

Распределённый журнал сообщений. Записи лежат в топиках и хранятся заданное время, поэтому читать их можно повторно.

Топик (topic)

Именованный журнал в Kafka, разбитый на партиции.

Партиция (partition)

Часть топика Kafka с собственным порядком записей. Единица параллельного чтения.

Коннектор (connector)

Библиотека, связывающая Flink с внешней системой. Выходит отдельно от Flink и собирается под его конкретную версию.

Толстый jar (fat jar)

Один файл с твоим кодом и всеми его зависимостями. Его собирает плагин Shadow для Gradle.

REST-интерфейс

HTTP-интерфейс JobManager на порту 8081. Через него работает Web UI, и им же удобно пользоваться из скриптов.

Web UI

Панель Flink в браузере. Показывает граф задания, метрики, чекпоинты и ошибки.

Docker Compose

Инструмент, который поднимает несколько контейнеров по описанию из одного YAML-файла.

Record

Вид неизменяемого класса в Java. Компилятор сам создаёт конструктор, геттеры, equals и hashCode.

Авария, которую объясняет устройство кластера

Ночью PayGuard перезапустили с параллелизмом 8 на кластере с четырьмя слотами. В мониторинге статус задания RUNNING, алертов о падении нет, а платежи никто целиком не проверяет. Через несколько минут задание падает с NoResourceAvailableException. Другой частый случай выглядит так. Задание годами запускалось из CI, его перевели в режим Application, и оно упало на чтении локального файла в main.

Оба случая разбираются за минуту, если знать, какой процесс за что отвечает и в каком порядке они разговаривают. Этим мы и займёмся, причём пройдём весь путь по коду Flink 2.2.1.

Стройка на пять ролей

Представь строительную контору. Заказчик приносит проект в приёмную и может уехать домой. Приёмная регистрирует заказ и назначает на него отдельного прораба. Прораб считает, сколько рабочих мест ему нужно, и обращается к кадровику, который знает все бригады и их свободные места. Кадровик говорит бригаде, к кому идти, бригада сама приходит к прорабу, и тот выдаёт наряды.

Во Flink заказчика зовут клиентом, приёмную зовут Dispatcher, прораба JobMaster, кадровика ResourceManager, бригаду TaskManager, а рабочее место слотом. Dispatcher, ResourceManager и все JobMaster живут внутри одного процесса JobManager, это написано прямо в разделе про архитектуру.

Со Spark роли похожи только частично, и таблица показывает, где сходство обманывает.

Путь задания шаг за шагом

Жми кнопку и смотри, кто с кем говорит, когда ты запускаешь flink run для нашего задания с параллелизмом 2.

StreamGraph streamGraph = PipelineExecutorUtils.getStreamGraph(pipeline, configuration);

try (final ClusterDescriptor<ClusterID> clusterDescriptor =
        clusterClientFactory.createClusterDescriptor(configuration)) {
    final ClusterID clusterID = clusterClientFactory.getClusterId(configuration);
    checkState(clusterID != null);

    final ClusterClientProvider<ClusterID> clusterClientProvider =
            clusterDescriptor.retrieve(clusterID);
    ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();

    streamGraph.serializeUserDefinedInstances();
    return clusterClient
            .submitJob(streamGraph)

resultingConfiguration.set(DeploymentOptions.TARGET, RemoteExecutor.NAME);

DynamicPropertiesUtil.encodeDynamicProperties(commandLine, resultingConfiguration);

return resultingConfiguration;

private void persistAndRunJob(ExecutionPlan executionPlan) throws Exception {
    executionPlanWriter.putExecutionPlan(executionPlan);
    initJobClientExpiredTime(executionPlan);
    runJob(createJobMasterRunner(executionPlan), ExecutionType.SUBMISSION);
}

private JobManagerRunner createJobMasterRunner(ExecutionPlan executionPlan) throws Exception {
    Preconditions.checkState(!jobManagerRunnerRegistry.isRegistered(executionPlan.getJobID()));
    return jobManagerRunnerFactory.createJobManagerRunner(
            executionPlan,
            configuration,
            getRpcService(),
            highAvailabilityServices,
            heartbeatServices,
            jobManagerSharedServices,
            new DefaultJobManagerJobMetricGroupFactory(jobManagerMetricGroup),
            fatalErrorHandler,
            failureEnrichers,
            System.currentTimeMillis());
}

Три графа одного задания

Один и тот же код проходит через три представления. На стройке это эскиз архитектора, рабочие чертежи и наряды на конкретных людей.

• StreamGraph описывает логику. Каждый оператор твоей программы в нём отдельный узел.
• JobGraph описывает, что исполняется как единое целое. Его вершины (JobVertex) это цепочки операторов, которые будут жить в одном потоке.
• ExecutionGraph описывает параллельное исполнение. На каждую вершину приходится столько подзадач, сколько задано параллелизмом, и у каждой есть своё состояние и своя история попыток.

JobGraph jobGraph;

if (executionPlan instanceof JobGraph) {
    jobGraph = (JobGraph) executionPlan;
} else if (executionPlan instanceof StreamGraph) {
    jobGraph = ((StreamGraph) executionPlan).getJobGraph(userCodeLoader);
} else {
    throw new FlinkException(
            "Unsupported execution plan " + executionPlan.getClass().getCanonicalName());
}

public JobGraph getJobGraph(ClassLoader userClassLoader) {
    return getJobGraph(userClassLoader, jobId);
}

/** Gets the assembled {@link JobGraph} with a specified {@link JobID}. */
public JobGraph getJobGraph(ClassLoader userClassLoader, @Nullable JobID jobID) {
    return StreamingJobGraphGenerator.createJobGraph(userClassLoader, this, jobID);
}

// topologically sort the job vertices and attach the graph to the existing one
List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
if (log.isDebugEnabled()) {
    log.debug(
            "Adding {} vertices from job graph {} ({}).",
            sortedTopology.size(),
            jobName,
            jobId);
}
executionGraph.attachJobGraph(sortedTopology, jobManagerJobMetricGroup);

this.taskVertices = new ExecutionVertex[parallelismInfo.getParallelism()];

// create all task vertices
for (int i = 0; i < this.parallelismInfo.getParallelism(); i++) {
    ExecutionVertex vertex =
            createExecutionVertex(
                    this,
                    i,
                    producedDataSets,
                    timeout,
                    createTimestamp,
                    executionHistorySizeLimit,

Цепочки операторов

Цепочка (operator chain) соединяет несколько операторов в один поток, и запись идёт между ними обычным вызовом метода, без сети и сериализации. Каменщик сразу штукатурит свою стену, и материалы никто не перевозит. Документация объясняет выгоду меньшими расходами на передачу между потоками и буферизацию. Подробно о том, что происходит на разрыве цепочки, поговорим в уроке 4.

private static boolean isChainableInput(
        StreamEdge edge, StreamGraph streamGraph, boolean allowChainWithDefaultParallelism) {
    StreamNode upStreamVertex = streamGraph.getSourceVertex(edge);
    StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);

    if (!(streamGraph.isChainingEnabled()
            && upStreamVertex.isSameSlotSharingGroup(downStreamVertex)
            && areOperatorsChainable(
                    upStreamVertex,
                    downStreamVertex,
                    streamGraph,
                    allowChainWithDefaultParallelism)
            && arePartitionerAndExchangeModeChainable(
                    edge.getPartitioner(), edge.getExchangeMode(), streamGraph.isDynamic()))) {

        return false;
    }

@VisibleForTesting
static boolean arePartitionerAndExchangeModeChainable(
        StreamPartitioner<?> partitioner,
        StreamExchangeMode exchangeMode,
        boolean isDynamicGraph) {
    if (partitioner instanceof ForwardForConsecutiveHashPartitioner) {
        checkState(isDynamicGraph);
        return true;
    } else if ((partitioner instanceof ForwardPartitioner)
            && exchangeMode != StreamExchangeMode.BATCH) {
        return true;
    } else {
        return false;
    }
}

// When allowChainWithDefaultParallelism is true, any vertex with default parallelism can
// be chained, otherwise only vertices with the same parallelism can be chained.
if (allowChainWithDefaultParallelism) {
    isChainable &=
            (upStreamVertex.getParallelism() == downStreamVertex.getParallelism()
                    || upStreamVertex.getParallelism()
                            == ExecutionConfig.PARALLELISM_DEFAULT
                    || downStreamVertex.getParallelism()
                            == ExecutionConfig.PARALLELISM_DEFAULT);
} else {
    isChainable &= upStreamVertex.getParallelism() == downStreamVertex.getParallelism();
}

Слоты и slot sharing

Слот это единица планирования в TaskManager. Каждый слот получает равную долю управляемой памяти, а ядра между ними не делятся. Документация говорит об этом прямо, поэтому четыре слота на одном CPU дадут четыре потока, которые спорят за процессор.

По умолчанию подзадачи разных вершин одного задания могут жить в одном слоте. Этот механизм называется slot sharing. Правило простое. В один разделяемый слот попадает не больше одной подзадачи от каждой вершины. Отсюда простая арифметика, которая встречается в работе постоянно. Заданию с одной группой нужно столько слотов, сколько составляет наибольший параллелизм среди его вершин.

* A JobVertex only belongs to one {@link SlotSharingGroup}. A SlotSharingGroup is
* corresponding to a set of {@link ExecutionSlotSharingGroup}s. We can maintain available
* ExecutionSlotSharingGroups for each JobVertex.
*
* <p>Once an ExecutionSlotSharingGroup is created, it becomes available for all JobVertices
* in the corresponding SlotSharingGroup in the beginning.
*
* <p>Once a SchedulingExecutionVertex is added to the ExecutionSlotSharingGroup, the group
* is no longer available for other SchedulingExecutionVertices with the same JobVertexID.

Покрути планировщик. Поставь параллелизм 8 при четырёх слотах и посмотри на результат. Потом включи отдельные группы, и потребность в слотах вырастет втрое. Заодно цепочка разорвётся, потому что операторы из разных групп код склеивать запрещает. Порядок, в котором модель заполняет TaskManager, условный. Настоящий зависит от настройки taskmanager.load-balance.mode.

RUNNING ещё ничего не гарантирует

Вернёмся к ночной аварии. Статус задания и состояние подзадач разные вещи, и разница видна в одном методе.

protected void startSchedulingInternal() {
    log.info(
            "Starting scheduling with scheduling strategy [{}]",
            schedulingStrategy.getClass().getName());
    transitionToRunning();
    schedulingStrategy.startScheduling();
}

taskDeploymentDescriptorFuture
        .thenComposeAsync(
                deploymentDescriptor ->
                        taskManagerGateway.submitTask(deploymentDescriptor, rpcTimeout),
                executor)
        .whenCompleteAsync(
                (ack, failure) ->
                        handleDeploymentCompletionAndCleanup(
                                maybeOffloadedTaskRestoreCleanupRef, ack, failure),
                jobMasterMainThreadExecutor);

try {
    taskAdded = taskSlotTable.addTask(task);
} catch (SlotNotFoundException | SlotNotActiveException e) {
    throw new TaskSubmissionException("Could not submit task.", e);
}

if (taskAdded) {
    task.startTaskThread();

    setupResultPartitionBookkeeping(
            tdd.getJobId(), tdd.getProducedPartitions(), task.getTerminationFuture());
    return CompletableFuture.completedFuture(Acknowledge.get());

if (!transitionState(ExecutionState.INITIALIZING, ExecutionState.RUNNING)) {
    throw new CancelTaskException();
}

// notify everyone that we switched to running
taskManagerActions.updateTaskExecutionState(
        new TaskExecutionState(executionId, ExecutionState.RUNNING));

runWithSystemExitMonitoring(finalInvokable::invoke);

Когда слотов не хватило

Кластер standalone не умеет сам поднимать новые TaskManager, ResourceManager распределяет только то, что уже подключилось. На Kubernetes или YARN он может запросить новые процессы. Поэтому в нашем стенде нехватка слотов кончается ошибкой, и поведение зависит от возраста кластера.

private void startStartupPeriod() {
    setFailUnfulfillableRequest(false);

    final long startupPeriodMillis = startupPeriodTime.toMillis();

    if (startupPeriodMillis > 0) {
        scheduleRunAsync(
                () -> setFailUnfulfillableRequest(true),
                startupPeriodMillis,
                TimeUnit.MILLISECONDS);
    }
}

public static final ConfigOption<Duration> SLOT_REQUEST_TIMEOUT =
        key("slot.request.timeout")
                .durationType()
                .defaultValue(Duration.ofMillis(5L * 60L * 1000L))
                .withDescription("The timeout for requesting a slot from Slot Pool.");

if (sendNotEnoughResourceNotifications) {
    for (JobID jobId : unfulfillableJobs) {
        try (MdcUtils.MdcCloseable ignored =
                MdcUtils.withContext(MdcUtils.asContextData(jobId))) {
            LOG.warn("Could not fulfill resource requirements of job {}.", jobId);
            resourceEventListener.notEnoughResourceAvailable(
                    jobId, resourceTracker.getAcquiredResources(jobId));
        }
    }
}

private void failPendingRequests(Collection<ResourceRequirement> acquiredResources) {
    // only fails streaming requests because batch jobs do not require all resources
    // requirements to be fullfilled at the same time
    Predicate<PendingRequest> predicate = request -> !request.isBatchRequest();
    if (pendingRequests.values().stream().anyMatch(predicate)) {
        log.warn(
                "Could not acquire the minimum required resources, failing slot requests. Acquired: {}. Current slot pool status: {}",
                acquiredResources,
                getSlotServiceStatus());
        cancelPendingRequests(
                predicate,
                NoResourceAvailableException.withoutStackTrace(
                        "Could not acquire the minimum required resources."));
    }
}

case TIMEOUT:
    Throwable cancellationCause =
            new NoResourceAvailableException(
                    "Slot request bulk is not fulfillable! Could not allocate the required slot within slot request timeout",
                    new TimeoutException(
                            "Timeout has occurred: " + timeout));
    bulk.cancel(cancellationCause);
    break;

Групповой чат запуска

Тот же запуск глазами участников, вместе с вопросом, что случится при нехватке слотов.

Режимы Session и Application

Сегодня Flink разворачивается двумя способами. Они различаются временем жизни кластера и тем, где выполняется main. Третий режим per-job удалён во Flink 2.0, и в старых статьях он всё ещё встречается.

В нашем стенде режим Session, потому что кластер поднимает Docker Compose заранее. Развёртывание в Kubernetes разберём в уроке 20. Учти одну тонкость стенда. Команда docker compose exec jobmanager flink run выполняется в контейнере JobManager, но клиент при этом отдельная JVM. Твой main работает в ней, и JobManager его кода не исполняет.

final StreamGraph streamGraph =
        PipelineExecutorUtils.getStreamGraph(pipeline, configuration);
final JobID actualJobId = streamGraph.getJobID();

this.submittedJobIds.add(actualJobId);
LOG.info("Job {} is submitted.", actualJobId);

if (LOG.isDebugEnabled()) {
    LOG.debug("Effective Configuration: {}", configuration);
}

final CompletableFuture<JobID> jobSubmissionFuture =
        submitJob(configuration, dispatcherGateway, streamGraph, timeout);

Проверь себя

Что будет в конце

В задании появится оператор посередине, у всех операторов будут постоянные идентификаторы, скрипт отправки избавит от длинной команды, и ты увидишь цепочку в выводе CLI и REST. Новых зависимостей нет. Файл scripts/cluster-status.sh из снапшота относится к заданию этого урока, пока его создавать не нужно.

Шаг 1. Дай операторам имена и uid

Открой src/main/java/dev/study/payguard/job/PayGuardJob.java и внеси такие правки.

@@ -24,9 +24,15 @@                 .build();
 
         env.fromSource(source, WatermarkStrategy.noWatermarks(), "payments-raw")
+                .uid("payments-raw")+                .filter(record -> !record.isBlank())+                .name("drop-blank")+                .uid("drop-blank")                 .print()
-                .name("stdout");+                .name("stdout")+                .uid("stdout"); 
+        System.out.println(env.getExecutionPlan());         env.execute("payguard");
     }
 

        env.fromSource(source, WatermarkStrategy.noWatermarks(), "payments-raw")
                .uid("payments-raw")
                .filter(record -> !record.isBlank())
                .name("drop-blank")
                .uid("drop-blank")
                .print()
                .name("stdout")
                .uid("stdout");

        System.out.println(env.getExecutionPlan());
        env.execute("payguard");

// Check for user-specified ID
String userSpecifiedHash = node.getTransformationUID();

if (userSpecifiedHash == null) {
    // Check that all input nodes have their hashes computed
    for (StreamEdge inEdge : node.getInEdges()) {
        // If the input node has not been visited yet, the current
        // node will be visited again at a later point when all input
        // nodes have been visited and their hashes set.
        if (!hashes.containsKey(inEdge.getSourceId())) {
            return false;
        }
    }

    Hasher hasher = hashFunction.newHasher();
    byte[] hash =
            generateDeterministicHash(node, hasher, hashes, isChainingEnabled, streamGraph);

    if (hashes.put(node.getId(), hash) != null) {
        // Sanity check
        throw new IllegalStateException(
                "Unexpected state. Tried to add node hash "
                        + "twice. This is probably a bug in the JobGraph generator.");
    }

    return true;
} else {
    Hasher hasher = hashFunction.newHasher();
    byte[] hash = generateUserSpecifiedHash(node.getTransformationUID(), hasher);

/**
 * Creates the plan with which the system will execute the program, and returns it as a String
 * using a JSON representation of the execution data flow graph. Note that this needs to be
 * called, before the plan is executed.
 *
 * @return The execution plan of the program, as a JSON String.
 */
public String getExecutionPlan() {
    return getStreamGraph(false).getStreamingPlanAsJSON();
}

Шаг 2. Скрипт отправки

Создай scripts/submit.sh.

#!/usr/bin/env bash
# Отправляет собранный jar в кластер в фоновом режиме. Параметры передаются заданию как есть.
set -euo pipefail
docker compose exec -T jobmanager flink run --detached /jobs/payguard-1.0.0.jar "$@"

Строка set -euo pipefail останавливает скрипт при первой ошибке и при обращении к неопределённой переменной. Флаг -T у docker compose exec отключает псевдотерминал, без него скрипт падал бы в CI, где терминала нет. Флаг --detached завершает клиента сразу после отправки. Конструкция "$@" передаёт аргументы скрипта после пути к jar, поэтому они попадают в main твоего задания. Флаги самого flink run, например -p, таким способом не передать, их пиши в отдельной команде.

chmod +x scripts/submit.sh

Команда chmod +x разрешает запускать файл как программу.

Шаг 3. Собери и отправь

Если стенд остался работать с урока 1, сначала выполни docker compose down, чтобы старое задание не занимало слоты. Порядок прежний, сначала сборка, потом Compose.

./gradlew shadowJar
docker compose up -d --wait
curl -s localhost:8081/overview | jq -c .
./scripts/submit.sh

В прогоне jar занял 24327017 байт. Ответ /overview подтверждает, что свободны все четыре слота. Поля те же, что в уроке 1.

{"taskmanagers":1,"slots-total":4,"slots-available":4,"jobs-running":0,"jobs-finished":0,"jobs-cancelled":0,"jobs-failed":0,"flink-version":"2.2.1","flink-commit":"450c63e"}

Теперь вывод самой отправки. Обрати внимание на его источник. Этот JSON напечатал твой System.out.println, то есть код из main выполнился в клиенте, и результат пришёл в твой терминал.

{
  "nodes" : [ {
    "id" : 1,
    "type" : "Source: payments-raw",
    "pact" : "Data Source",
    "contents" : "Source: payments-raw",
    "parallelism" : 2
  }, {
    "id" : 2,
    "type" : "drop-blank",
    "pact" : "Operator",
    "contents" : "drop-blank",
    "parallelism" : 2,
    "predecessors" : [ {
      "id" : 1,
      "ship_strategy" : "FORWARD",
      "side" : "second"
    } ]
  }, {
    "id" : 3,
    "type" : "Sink: stdout",
    "pact" : "Data Sink",
    "contents" : "Sink: stdout",
    "parallelism" : 2,
    "predecessors" : [ {
      "id" : 2,
      "ship_strategy" : "FORWARD",
      "side" : "second"
    } ]
  } ]
}
Job has been submitted with JobID 8fe3d0eda0de9f053bb73d644c2fd485

Это StreamGraph, три узла с нашими именами. Поле ship_strategy показывает способ передачи между операторами. Значение FORWARD здесь значит, что подзадача передаёт записи подзадаче с тем же номером, и именно оно разрешает склейку. Значение parallelism равно 2, потому что в Compose задано parallelism.default: 2. Последняя строка содержит идентификатор задания, он пригодится в следующих командах.

Шаг 4. Посмотри на то, что реально работает

Спроси у REST, какие вершины запущены. Подставь свой идентификатор. Фильтр jq оставляет только нужные поля, а флаг -c печатает результат в одну строку.

curl -s localhost:8081/jobs/8fe3d0eda0de9f053bb73d644c2fd485 \
  | jq -c '{name, state, vertices: [.vertices[] | {name, parallelism, status}]}'

{"name":"payguard","state":"RUNNING","vertices":[{"name":"Source: payments-raw -> drop-blank -> Sink: stdout","parallelism":2,"status":"RUNNING"}]}

Три узла превратились в одну вершину с именем через стрелки. Это JobGraph, и склейку ты видишь своими глазами. Две подзадачи заняли два слота, значит в /overview теперь должно быть 2 в поле slots-available. При параллелизме 4 в нашем стенде заняты все четыре слота, это проверено на запуске из урока 13.

Шаг 5. CLI

Список заданий кластера выдаёт команда flink list.

docker compose exec -T jobmanager flink list

Waiting for response...
------------------ Running/Restarting Jobs -------------------
07.11.2026 20:56:54 : 8fe3d0eda0de9f053bb73d644c2fd485 : payguard (RUNNING)
--------------------------------------------------------------
No scheduled jobs.

В строке стоят время запуска, идентификатор, имя из env.execute и статус. Остановить задание без сохранения состояния можно так.

docker compose exec -T jobmanager flink cancel 8fe3d0eda0de9f053bb73d644c2fd485

Команда cancel переводит задание в отменённое состояние, и все его вычисления прекращаются. После этого curl -s localhost:8081/jobs | jq -c . покажет задание со статусом CANCELED, а все четыре слота снова будут свободны. Для остановки с сохранением состояния есть flink stop, его мы разберём в уроке 13. Запусти задание снова командой ./scripts/submit.sh, идентификатор будет новым.

Шаг 6. Web UI

Открой http://localhost:8081. На странице запущенных заданий нажми payguard и найди на графе блок с надписью Source: payments-raw -> drop-blank -> Sink: stdout. Это та же одна вершина, что в REST. Нажми на неё и открой список подзадач, там их две, и у каждой указан TaskManager. На странице Task Managers видно число всех и свободных слотов. На странице задания есть вкладки Exceptions и Checkpoints. В нашем прогоне список исключений был пуст и чекпоинтов ноль, мы их ещё не включали.

Чтобы записи шли, запусти генератор из урока 1. Он отправит в поток битые записи, и они пока пройдут через фильтр дальше в print, потому что пустыми не являются.

./gradlew loadgen --args="--rate 5 --dup 0.02 --bad 0.01"

Шаг 7. Проверь на себе нехватку слотов

Этот опыт покажет, как выглядит авария из начала урока. Флаг -p задаёт параллелизм заданию при отправке и стоит перед путём к jar.

docker compose exec -T jobmanager flink cancel <id текущего задания>
docker compose exec -T jobmanager flink run --detached -p 8 /jobs/payguard-1.0.0.jar

В этом случае подзадач восемь, а слотов четыре. Вывод такого запуска мы не снимали, поэтому смотри на него сам и сверяй с теорией. Статус задания должен стать RUNNING, затем в логе JobManager появится NoResourceAvailableException. Если кластер только что поднят, ждать придётся до пяти минут. Логи можно читать командой docker compose logs jobmanager. Для продолжения курса верни обычный запуск через ./scripts/submit.sh.

Задание. Скрипт состояния кластера

Когда заданий и TaskManager становится больше, смотреть в браузер неудобно. Напиши scripts/cluster-status.sh, который одной командой показывает то, что мы сегодня проверяли руками. Пользуйся только curl и jq, CLI Flink не нужен.

1. Первая строка показывает число TaskManager, сколько слотов занято из общего числа и версию Flink.
2. Для каждого задания со статусом RUNNING скрипт показывает его имя, статус и идентификатор.
3. Под каждым заданием идут его вершины, у каждой параллелизм, статус и имя.
4. Адрес кластера по умолчанию http://localhost:8081, переменная окружения FLINK_REST его заменяет.
5. При ошибке в команде скрипт завершается, как и submit.sh.

Готово, если при работающем задании вывод показывает одну вершину с параллелизмом 2 и два занятых слота, а после flink cancel остаётся только первая строка с нулём занятых слотов.

Подсказки. Адреса нужны такие, /overview, /jobs и /jobs/<id>. Занятые слоты придётся вычислить из двух полей. В имени поля с дефисом jq требует запись в скобках. Для вставки значений в строку в jq есть интерполяция \(...), а флаг -r убирает кавычки вокруг результата. Посмотри ответ /jobs глазами, прежде чем писать фильтр, названия полей статуса в разных адресах отличаются.

Ответ

#!/usr/bin/env bash
# Печатает состояние кластера и всех вершин запущенных заданий через REST API JobManager.
set -euo pipefail
REST="${FLINK_REST:-http://localhost:8081}"

curl -s "$REST/overview" | jq -r '"TaskManager: \(.taskmanagers)  слоты: \(.["slots-total"] - .["slots-available"]) заняты из \(.["slots-total"])  версия: \(.["flink-version"])"'

for job_id in $(curl -s "$REST/jobs" | jq -r '.jobs[] | select(.status == "RUNNING") | .id'); do
  curl -s "$REST/jobs/$job_id" | jq -r '"\nЗадание \(.name) [\(.state)] \(.jid)", (.vertices[] | "  \(.parallelism)x  \(.status)  \(.name)")'
done

set -euo pipefail
REST="${FLINK_REST:-http://localhost:8081}"

curl -s "$REST/overview" | jq -r '"TaskManager: \(.taskmanagers)  слоты: \(.["slots-total"] - .["slots-available"]) заняты из \(.["slots-total"])  версия: \(.["flink-version"])"'

for job_id in $(curl -s "$REST/jobs" | jq -r '.jobs[] | select(.status == "RUNNING") | .id'); do
  curl -s "$REST/jobs/$job_id" | jq -r '"\nЗадание \(.name) [\(.state)] \(.jid)", (.vertices[] | "  \(.parallelism)x  \(.status)  \(.name)")'
done

chmod +x scripts/cluster-status.sh
./scripts/cluster-status.sh
FLINK_REST=http://localhost:9999 ./scripts/cluster-status.sh; echo "код выхода $?"

Глоссарий урока

Клиент (client)

Программа, которая готовит задание и отправляет его на кластер. Это JVM команды flink run или твоего Java-приложения. В исполнение задания не входит.

JobManager

Процесс, который управляет кластером. Содержит Dispatcher, ResourceManager и по одному JobMaster на задание.

Dispatcher

Компонент JobManager. Даёт REST-интерфейс для отправки заданий, запускает для каждого новый JobMaster и обслуживает Web UI.

ResourceManager

Компонент JobManager, который ведёт учёт слотов кластера и раздаёт их заданиям. На Kubernetes и YARN также запрашивает новые TaskManager.

JobMaster

Компонент JobManager, отвечающий за одно задание. Держит его ExecutionGraph, просит слоты, развёртывает подзадачи, реагирует на сбои.

TaskManager

Рабочий процесс JVM. Выполняет подзадачи в отдельных потоках и обменивается данными с другими TaskManager.

Слот (task slot)

Единица планирования ресурсов в TaskManager. Делит управляемую память поровну и не изолирует CPU. Число задаёт ключ taskmanager.numberOfTaskSlots.

Slot sharing

Размещение подзадач разных вершин одного задания в одном слоте. Включено по умолчанию. Задаётся через группу, по умолчанию она называется default.

Подзадача (subtask)

Один параллельный экземпляр вершины задания. Выполняется в своём потоке на TaskManager.

Параллелизм (parallelism)

Число подзадач вершины. Задаётся в коде, флагом -p или ключом parallelism.default.

StreamGraph

Логический граф программы, где каждый оператор отдельный узел. Создаётся при вызове env.execute. Клиент во Flink 2.2 отправляет на кластер именно его.

JobGraph и JobVertex

Граф, в котором операторы склеены в цепочки. Каждая цепочка это одна вершина (JobVertex). Строится из StreamGraph классом StreamingJobGraphGenerator.

ExecutionGraph

Параллельная версия JobGraph, которую держит JobMaster. На каждую вершину в нём столько элементов, сколько задано параллелизмом.

Цепочка операторов (operator chain)

Операторы, которые работают в одном потоке и передают записи вызовом метода, без сети. Разрывается при keyBy, разном параллелизме или разных группах slot sharing.

FORWARD

Способ передачи, при котором подзадача отправляет записи подзадаче следующего оператора с тем же номером. Без него операторы не склеиваются.

Режим Session

Кластер работает независимо от заданий и обслуживает несколько сразу. main выполняется в клиенте.

Режим Application

Кластер создаётся на одно приложение, а main выполняется в JobManager. Jar уже лежит на classpath кластера.

Режим per-job

Прежний режим развёртывания, удалённый во Flink 2.0. Вместо него используется Application.

Standalone

Способ запуска кластера без системы управления ресурсами. Процессы поднимаешь сам, как у нас в Docker Compose, и новых TaskManager кластер запросить не может.

uid

Постоянный идентификатор оператора, заданный вручную. По нему Flink находит состояние оператора при восстановлении из savepoint.

Savepoint

Снимок состояния задания, который ты создаёшь вручную, чтобы остановить задание и продолжить его с того же места. Разбирается в уроке 13.

Стратегия перезапуска (restart strategy)

Правило, по которому Flink решает, перезапускать ли задание после сбоя и с какой задержкой.

Чекпоинт (checkpoint)

Согласованный снимок состояния, который Flink снимает автоматически. Подробно в уроке 11.

NoResourceAvailableException

Исключение, которым кончается запрос слотов, если кластер не может их дать.

REST-интерфейс

HTTP-интерфейс Dispatcher на порту 8081. Через него работают Web UI, клиент flink и твои скрипты.

Web UI

Панель Flink в браузере. Показывает вершины, подзадачи, слоты и исключения.

jq

Консольная утилита для разбора и преобразования JSON.

Driver (Spark)

Процесс Spark, который выполняет main и планирует работу приложения.

Executor (Spark)

Рабочий процесс Spark, который выполняет задачи. Похож на TaskManager.

Kubernetes

Система управления контейнерами. Flink на ней может сам запрашивать новые TaskManager. Разбирается в уроке 20.

YARN

Менеджер ресурсов из экосистемы Hadoop, на котором тоже можно запускать Flink.

Apache Kafka

Распределённый журнал сообщений. Из него наше задание читает платежи.

Одна битая запись против целого задания

Ночью партнёр выкатывает новую версию сервиса, и в топик платежей попадает строка с обрезанным JSON. Парсер в задании бросает исключение. Дальше всё зависит от настроек, и ни один из вариантов по умолчанию тебя не обрадует.

Без чекпоинтов (периодических снимков задания) задание переходит в состояние FAILED и перестаёт читать платежи, пока кто-то не запустит его заново. С чекпоинтами Flink перезапускает его и возвращается к смещениям из последнего снимка. Та же запись приезжает снова, и всё повторяется. Такую запись называют ядовитой (poison pill).

 * @param value The input value.
 * @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
 *     {@link TimerService} for registering timers and querying the time. The context is only
 *     valid during the invocation of this method, do not store it.
 * @param out The collector for returning result values.
 * @throws Exception This method may throw exceptions. Throwing an exception will cause the
 *     operation to fail and may trigger recovery.
 */
public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;

@Override
public void processElement(StreamRecord<IN> element) throws Exception {
    collector.setTimestamp(element);
    context.element = element;
    userFunction.processElement(element.getValue(), context, collector);
    context.element = null;
}

.text(
        "If checkpointing is disabled, the default value is %s. "
                + "If checkpointing is enabled, the default value is %s, and the default values of %s related config options will be used.",
        code(NO_RESTART_STRATEGY.getMainValue()),
        code(EXPONENTIAL_DELAY.getMainValue()),
        code(EXPONENTIAL_DELAY.getMainValue()))

@Documentation.OverrideDefault("infinite")
public static final ConfigOption<Integer> RESTART_STRATEGY_EXPONENTIAL_DELAY_ATTEMPTS =
        ConfigOptions.key("restart-strategy.exponential-delay.attempts-before-reset-backoff")
                .intType()
                .defaultValue(Integer.MAX_VALUE)
                .withDescription(

Линия розлива

Представь завод лимонада. Бутылки едут по ленте, автомат клеит этикетки. Бутылка со сколом может заклинить автомат, и тогда встанет вся линия. После ремонта на ленту приедет та же бутылка, потому что с неё и начали.

Поэтому перед автоматом ставят отбраковщик. Он сбрасывает плохую бутылку на боковой лоток и вешает бирку с причиной. Лента едет дальше, а контролёр смены разбирает лоток, когда дойдут руки.

В Flink лента называется основным потоком, лоток называется побочным выходом (side output), а ящик с бирками это топик для отбракованных записей, dead letter queue или DLQ. Выбери вариант и посмотри на судьбу задания. Модель упрощённая, в ней 16 записей и две из них битые.

Такой же цикл есть в реальных замерах курса. В уроке 15 задание с CDC после первого UPDATE уходит в цикл RESTARTING из-за одной и той же ошибки, и само оттуда не выходит.

DataStream это план

Прежде чем чинить, разберёмся, из чего строится задание. DataStream<T> внешне похож на Dataset<T> в Spark. Это неизменяемое описание потока с известным типом, и каждый вызов вроде map или process возвращает новый объект. Ни одна запись при этом никуда не едет.

protected <R> SingleOutputStreamOperator<R> doTransform(
        String operatorName,
        TypeInformation<R> outTypeInfo,
        StreamOperatorFactory<R> operatorFactory) {

    // read the output type of the input Transform to coax out errors about MissingTypeInfo
    transformation.getOutputType();

    OneInputTransformation<T, R> resultTransform =
            new OneInputTransformation<>(
                    this.transformation,
                    operatorName,
                    operatorFactory,
                    outTypeInfo,
                    environment.getParallelism(),
                    false);

    @SuppressWarnings({"unchecked", "rawtypes"})
    SingleOutputStreamOperator<R> returnStream =
            new SingleOutputStreamOperator(environment, resultTransform);

    getExecutionEnvironment().addOperator(resultTransform);

    return returnStream;
}

Основные преобразования, которые встретятся в курсе.

Ленивость тебе знакома по Spark, там план тоже начинает работать после start(). Разница в том, что происходит дальше. Spark в обычном режиме исполняет план заново на каждой порции. Flink один раз разворачивает его в операторы, и они живут, пока живёт задание.

Побочный выход

Разделить поток на хорошие и плохие записи можно двумя фильтрами, но тогда каждую запись придётся разбирать дважды, причём оба фильтра должны согласованно решать, что считать ошибкой. Побочный выход решает это за один проход. Функция один раз разбирает строку и отправляет результат либо в обычный Collector, либо в выход с меткой OutputTag. Типы двух выходов могут различаться. У нас основной поток содержит PaymentEvent, а побочный RejectedRecord.

public OutputTag(String id) {
    Preconditions.checkNotNull(id, "OutputTag id cannot be null.");
    Preconditions.checkArgument(!id.isEmpty(), "OutputTag id must not be empty.");
    this.id = id;

    try {
        this.typeInfo = TypeExtractor.createTypeInfo(this, OutputTag.class, getClass(), 0);
    } catch (InvalidTypesException e) {
        throw new InvalidTypesException(
                "Could not determine TypeInformation for the OutputTag type. "
                        + "The most common reason is forgetting to make the OutputTag an anonymous inner class. "
                        + "It is also not possible to use generic type variables with OutputTags, such as 'Tuple2<A, B>'.",
                e);
    }
}

@Override
public <X> void output(OutputTag<X> outputTag, X value) {
    if (outputTag == null) {
        throw new IllegalArgumentException("OutputTag must not be null.");
    }
    output.collect(outputTag, new StreamRecord<>(value, element.getTimestamp()));
}

@Override
public <X> void collect(OutputTag<X> outputTag, StreamRecord<X> record) {
    if (this.outputTag == null || !this.outputTag.equals(outputTag)) {
        // we are not responsible for emitting to the side-output specified by this
        // OutputTag.
        return;
    }

    pushToOperator(record);
}

public <X> SideOutputDataStream<X> getSideOutput(OutputTag<X> sideOutputTag) {
    sideOutputTag = clean(requireNonNull(sideOutputTag));

    // make a defensive copy
    sideOutputTag = new OutputTag<X>(sideOutputTag.getId(), sideOutputTag.getTypeInfo());

    TypeInformation<?> type = requestedSideOutputs.get(sideOutputTag);
    if (type != null && !type.equals(sideOutputTag.getTypeInfo())) {
        throw new UnsupportedOperationException(
                "A side output with a matching id was "
                        + "already requested with a different type. This is not allowed, side output "
                        + "ids need to be unique.");
    }

    requestedSideOutputs.put(sideOutputTag, sideOutputTag.getTypeInfo());

    SideOutputTransformation<X> sideOutputTransformation =
            new SideOutputTransformation<>(this.getTransformation(), sideOutputTag);
    return new SideOutputDataStream<>(this.getExecutionEnvironment(), sideOutputTransformation);
}

final List<Integer> virtualResultIds = new ArrayList<>();
final Transformation<?> parentTransformation = parentTransformations.get(0);
for (int inputId : context.getStreamNodeIds(parentTransformation)) {
    final int virtualId = Transformation.getNewNodeId();
    streamGraph.addVirtualSideOutputNode(inputId, virtualId, transformation.getOutputTag());
    virtualResultIds.add(virtualId);
}
return virtualResultIds;

Писать в побочный выход можно из ProcessFunction и родственных функций, полный список есть в документации по ссылке выше. У MapFunction и FlatMapFunction контекста с методом output нет, поэтому парсер у нас написан на ProcessFunction.

В Spark встроенного побочного выхода нет. Обычно строку разбирают через from_json, который на битых данных даёт null, и затем фильтруют результат на два направления внутри foreachBatch. Во Flink разделение происходит внутри одного оператора и на каждой записи.

Что класть в DLQ и что с ним делать

DLQ стоит проектировать как данные, которые придётся разбирать. Мы кладём три поля. Исходный текст нужен, чтобы после исправления парсера отправить запись обратно в payments.raw. Код причины позволяет считать отказы по видам. Время отказа показывает, когда началась проблема.

Лог TaskManager эту задачу не решает, потому что из него неудобно переигрывать записи. Топик Kafka хранит их заданное время и читается любым потребителем. Цена в том, что теперь задание зависит от записи в этот топик. Если продюсер получит ошибку, она дойдёт до задачи и уронит её.

@Override
public void flush(boolean endOfInput) throws IOException, InterruptedException {
    if (deliveryGuarantee != DeliveryGuarantee.NONE || endOfInput) {
        LOG.debug("final flush={}", endOfInput);
        currentProducer.flush();
    }

    checkAsyncException();
}

private void checkAsyncException() throws IOException {
    // reset this exception since we could close the writer later on
    Exception e = asyncProducerException;
    if (e != null) {

        asyncProducerException = null;
        numRecordsOutErrorsCounter.inc();
        throw new IOException(
                "One or more Kafka Producer send requests have encountered exception", e);
    }
}

@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
    super.prepareSnapshotPreBarrier(checkpointId);
    if (!endOfInput) {
        sinkWriter.flush(false);
        emitCommittables(checkpointId);
    }