Shuffle в YTsaurus Flow (Java)

Пайплайн читает поток событий, группирует их по ключу и подсчитывает количество уникальных событий с использованием внешнего стейта (ExternalStateAccessor). Пример демонстрирует конфигурацию компаньона через Spring Boot.

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

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

Компоненты компаньона

EventMapper

Процессная функция для source-компьютейшена reader. Выполняет парсинг и трансформацию входных данных. Ниже показана упрощённая версия — в реальном коде дополнительно выполняется JSON-парсинг поля data с помощью Jackson ObjectMapper:

Java

Kotlin

    @SuppressWarnings("unchecked")
    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        var messageBuilder = ctx.createMessageBuilder("event");
        String data = message.get("data", String.class);
        try {
            Map<String, Object> map = ysonMapper.readValue(data, Map.class);
            messageBuilder.set("value", map.get("value"));
            messageBuilder.set("key_a", map.get("key_a"));
            messageBuilder.set("key_b", map.get("key_b"));
            messageBuilder.set("key_c", map.get("key_c"));
            messageBuilder.set("key_d", map.get("key_d"));
            output.addMessage(messageBuilder.finish());
        } catch (JsonProcessingException e) {
            log.error("Error parsing json from data filed", e);
            throw new RuntimeException(e);
        }
    }

    @Suppress("UNCHECKED_CAST")
    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val messageBuilder = ctx.createMessageBuilder("event")
        val data = message.get("data", String::class.java)
        try {
            val map = ysonMapper.readValue(data, Map::class.java) as Map<String, Any?>
            messageBuilder.set("value", map["value"])
            messageBuilder.set("key_a", map["key_a"])
            messageBuilder.set("key_b", map["key_b"])
            messageBuilder.set("key_c", map["key_c"])
            messageBuilder.set("key_d", map["key_d"])
            output.addMessage(messageBuilder.finish())
        } catch (e: JsonProcessingException) {
            log.error("Error parsing json from data filed", e)
            throw RuntimeException(e)
        }
    }

EventReducer

Процессная функция с использованием ExternalStateAccessor для подсчета количества событий:

Java

Kotlin

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        ExternalStateAccessor externalStateAccessor = ctx.getState(SHUFFLE_STATE, message);
        Payload state = externalStateAccessor.getOrDefault();
        PayloadBuilder stateBuilder = state.toBuilder();
        if (state.get("count", Long.class) == null) {
            stateBuilder.set("count", 1L);
        } else {
            stateBuilder.set("count", state.get("count", Long.class) + 1);
        }
        externalStateAccessor.set(stateBuilder.finish());
    }

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val externalStateAccessor = ctx.getState(SHUFFLE_STATE, message)
        val state = externalStateAccessor.getOrDefault()
        val stateBuilder = state.toBuilder()
        val count = state.get("count", Long::class.java) ?: 0L
        stateBuilder.set("count", count + 1)
        externalStateAccessor.set(stateBuilder.finish())
    }

Логика работы:

Получаем ExternalStateAccessor для стейта "shuffle-state", привязанного к ключу текущего сообщения.
Извлекаем текущее значение стейта. Если стейта нет — getOrDefault() вернет пустой Payload.
Создаем PayloadBuilder из текущего стейта, увеличиваем счетчик.
Сохраняем обновленный стейт.

Регистрация компьютейшенов

Компьютейшен-источник reader регистрируется аннотацией @FlowSourceComputation, а трансформация reducer — аннотацией @FlowComputation:

Java

Kotlin

@FlowSourceComputation(id = "reader")

@FlowComputation(id = "reducer")

@FlowSourceComputation(id = "reader")

@FlowComputation(id = "reducer")

PipelineMain

Единственная точка входа (запускает пайплайн или обслуживает его как компаньон — по YT_FLOW_MODE):

Java

Kotlin

    public static void main(String[] args) throws Exception {
        new SpringApplicationBuilder(PipelineMain.class).run(args);
    }

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

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

Конфигурация через Spring Boot — компьютейшены регистрируются аннотациями @FlowSourceComputation / @FlowComputation; flow-spring-boot-starter управляет жизненным циклом gRPC-сервера.
ExternalStateAccessor — работа с внешним стейтом через Payload и PayloadBuilder.
SourceComputation с ProcessFunction — reader использует EventMapper для трансформации входных данных на стороне компаньона.

См. также

Word Count

Async Request