Быстрый старт в YTsaurus Flow (Java)

Поддержка вычислений на Java и Kotlin во Flow реализуется через механизм компаньонов. Код на Java или Kotlin выполняется в отдельном gRPC-процессе, который взаимодействует с C++ воркером.

Исходный код Java SDK для Flow

Примеры

Архитектура приложения

Любой пайплайн Flow состоит из трёх составных частей:

  • Runner — запускает пайплайн и устанавливает новую версию спеки.
  • Controller — управляет работой пайплайна.
  • Worker — выполняет непосредственно обработку данных.

Java и Kotlin используются в Runner и Worker.

Controller(standby)Controller(standby)Controller(leader)Runner(Console Java App)...Worker Pod 1Worker(C++)Companion(Java)Worker Pod NWorker(C++)Companion(Java)

Два подхода к конфигурации

Java SDK Flow (с поддержкой Kotlin) предоставляет два подхода для настройки компаньона:

  1. Ручной (PipelineContext + FlowApplication.run) — подходит для простых случаев, когда не нужна инъекция зависимостей.
  2. Spring Boot (auto-config с аннотациями @FlowComputation) — рекомендуемый подход для production-сервисов с развитой конфигурацией и зависимостями.

Computation и SourceComputation

Для создания компьютейшена на Java необходимо выбрать подходящий билдер, соответствующий типу Computation-a в C++:

  • Computation.builder() — для TTransformCompanionComputation и TSwiftMapCompanionComputation.
  • SourceComputation.builder() — для TSwiftOrderedSourceCompanionComputation и TTransformOrderedSourceCompanionComputation.
// SourceComputation для чтения данных из источника
var reader = SourceComputation.builder()
       .setComputationId("reader")
       .build();

// Computation для обработки данных
var mapper = Computation.builder()
       .setComputationId("mapper")
       .setProcessFunction(new WordCountMapper())
       .build();

// SourceComputation для чтения данных из источника
val reader = SourceComputation.builder()
       .setComputationId("reader")
       .build()

// Computation для обработки данных
val mapper = Computation.builder()
       .setComputationId("mapper")
       .setProcessFunction(WordCountMapper())
       .build()

У Computation.builder() два обязательных параметра:

  • Computation id — по нему происходит маппинг запросов между worker-ом и компаньоном.
  • Process function — функция с логикой обработки сообщений.

Process Function

Есть два вида ProcessFunction:

  • RowFunction — получает сообщения и таймеры по одному, предоставляет методы onMessage и onTimer.
  • BatchFunction — получает весь батч сообщений и таймеров, предоставляет методы onMessages и onTimers.

Подробнее — в разделе Computation (Java).

Точка входа

Без Spring Boot точка входа пайплайна — класс с main-методом, который вызывает FlowApplication.run(args, context). Со Spring Boot метод main запускает Spring, который настраивает Flow; см. Интеграция со Spring Boot. Роль процесса выбирается по переменной среды YT_FLOW_MODE:

  • переменная не задана — процесс работает как runner: дополняет спеку пайплайна и передаёт запуск flow_server;
  • YT_FLOW_MODE=Worker — процесс работает как компаньон: поднимает gRPC-сервер и обслуживает запросы воркера. Эту переменную воркер выставляет сам, запуская компаньон.

В main-методе нужно сконфигурировать компьютейшены и стримы и добавить их в PipelineContext:

import tech.ytsaurus.flow.computation.Computation;
import tech.ytsaurus.flow.context.PipelineContext;
import tech.ytsaurus.flow.pipeline.FlowApplication;

public class PipelineMain {
    public static void main(String[] args) throws Exception {
        var mapper = Computation.builder()
            .setComputationId("mapper")
            .setProcessFunction(new WordCountMapper())
            .build();

        var context = new PipelineContext();
        context.registerComputation(mapper);
        context.registerTypedStreams(Word.class);

        FlowApplication.run(args, context);
    }
}

import tech.ytsaurus.flow.computation.Computation
import tech.ytsaurus.flow.context.PipelineContext
import tech.ytsaurus.flow.pipeline.FlowApplication

object PipelineMain {
    @JvmStatic
    fun main(args: Array<String>) {
        val mapper = Computation.builder()
            .setComputationId("mapper")
            .setProcessFunction(WordCountMapper())
            .build()

        val context = PipelineContext()
        context.registerComputation(mapper)
        context.registerTypedStreams(Word::class.java)

        FlowApplication.run(args, context)
    }
}

Runner принимает два аргумента: обязательный --config — путь к конфигу пайплайна и --flow-bin — путь к бинарю flow_server. Без --flow-bin раннер берёт путь из переменной окружения YT_FLOW_BIN.

./run.sh com.example.pipeline.PipelineMain --config pipeline.yson --flow-bin flow_server

Класс указывается полным именем: run.sh передаёт первый аргумент напрямую в java.

Из зарегистрированного пайплайна runner дополняет схемы стримов в spec.streams — по типам
зарегистрированных сообщений, если в спеке они не заданы явно. main_class у ресурса
TJavaCompanionManager задаётся в спеке пайплайна.

Если пользовательским функциям нужны дополнительные ресурсы (словарь, кэш и т.п.), main-метод — подходящее место для их создания. Такие ресурсы должны быть потокобезопасными.

Spring Boot подход

При использовании Spring Boot компьютейшен mapper регистрируется аннотацией @FlowComputation прямо на классе process-функции (источник reader — passthrough: он объявляется в спеке пайплайна и в Java-компаньоне не регистрируется):

@FlowComputation(id = "mapper")
public class WordCountMapper implements RowFunction {
    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        // обработка сообщения
    }
}

@FlowComputation(id = "mapper")
class WordCountMapper : RowFunction {
    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        // обработка сообщения
    }
}

Типизированные стримы объявляются декларативно: POJO-класс сообщения помечается аннотацией @FlowMessage со списком идентификаторов стримов (streamIds), которые он обслуживает. Класс уже помечен JPA-аннотацией @Entity, из которой выводится схема. Spring Boot находит такие классы сканированием пакетов приложения и регистрирует стримы автоматически:

@Entity
@FlowMessage(streamIds = {"words"})
public class Word {
    // поля, конструкторы, геттеры, сеттеры...
}

@Entity
@FlowMessage(streamIds = ["words"])
class Word {
    // поля, конструкторы...
}

Точка входа Spring Boot-приложения:

@SpringBootApplication
public class WordCountApplication {
    public static void main(String[] args) {
        new SpringApplicationBuilder(WordCountApplication.class)
                .run(args);
    }
}

@SpringBootApplication
open class WordCountApplication {
    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            SpringApplicationBuilder(WordCountApplication::class.java).run(*args)
        }
    }
}

Метод getStreams() позволяет зарегистрировать типизированные стримы через FlowStreams.typed(...), чтобы SDK автоматически сериализовал и десериализовал сообщения в Java-объекты.

Для Spring Boot приложения отдельная точка входа не нужна: класс с @SpringBootApplication тоже работает в обоих режимах. В runner-режиме стартер не поднимает ни gRPC-сервер, ни сервер мониторинга — он собирает объявленные стримы, запускает пайплайн и завершает процесс.

См. также

Предыдущая
Proto Parser
Следующая
Computation