Word Count в YTsaurus Flow (Java)

Пайплайн читает поток слов и подсчитывает количество вхождений каждого слова с использованием YSON-стейта. Пример демонстрирует конфигурацию компаньона через Spring Boot.

Исходный код (Java)
Исходный код (Kotlin)

Компоненты

WordCountApplication

Единственная точка входа: класс с @SpringBootApplication запускает пайплайн, когда YT_FLOW_MODE не задана, и обслуживает его как компаньон, когда воркер выставляет YT_FLOW_MODE=Worker:

Java

Kotlin

@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)
        }
    }
}

gRPC-сервер поднимается автоматически через Spring Boot auto-config.

Регистрация стримов

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

Java

Kotlin

@Entity
@FlowMessage(streamIds = {"words"})
public class Word {
    private String word;

    public Word() {
    }

    public Word(String word) {
        this.word = word;
    }

    public String getWord() {
        return word;
    }

    public void setWord(String word) {
        this.word = word;
    }
}

@Entity
@FlowMessage(streamIds = ["words"])
class Word {
    var word: String? = null

    constructor()

    constructor(word: String?) {
        this.word = word
    }
}

reader — SourceComputation без процессной функции. Чтение и парсинг выполняются на стороне C++ worker-а.
mapper — Computation, реализованный классом WordCountMapper с аннотацией @FlowComputation(id = "mapper").
@FlowMessage(streamIds = {"words"}) на классе Word — регистрирует типизированный стрим "words", что позволяет получать сообщения как объекты Word.

WordCountMapper

Процессная функция, реализующая подсчет слов с использованием YsonStateAccessor:

Java

Kotlin

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        Word input = message.getPayload();
        StateAccessor<WordCountState> stateAccessor = ctx.getState(WORD_STATE, message);
        var state = stateAccessor.getOrDefault(new WordCountState(input.getWord(), 0));
        state.setCount(state.getCount() + 1);
    }

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val input: Word = message.getPayload()
        val stateAccessor = ctx.getState(WORD_STATE, message)
        val state = stateAccessor.getOrDefault(WordCountState(input.word, 0))
        state.count += 1
    }

Аннотация @FlowComputation(id = "mapper") регистрирует класс как компьютейшен и делает его Spring-бином (она мета-аннотирована @Component).

Ключевые паттерны

Spring Boot auto-config — не нужно вручную создавать PipelineContext и GrpcServerExecution.
@FlowComputation — процессная функция одновременно становится Spring-бином и компьютейшеном; можно использовать инъекцию зависимостей.
@FlowMessage — регистрация типизированных стримов через классы сообщений.
FlowStreams.typed — типизированный доступ к сообщениям через Java-объекты.

Запуск

Пайплайн запускается одним классом WordCountApplication:

./run.sh tech.ytsaurus.flow.examples.wordcount.WordCountApplication --config pipeline.yson --flow-bin flow_server

Он же служит компаньоном: воркер запускает его с YT_FLOW_MODE=Worker, и в этом режиме стартер поднимает gRPC-сервер вместо запуска пайплайна.

См. также

Переменные окружения

Shuffle