Static Table Join в YTsaurus Flow (Java и Kotlin)

Пайплайн загружает статический справочник во внешний стейт и обогащает события из очереди по общему ключу. Java и Kotlin используют один SDK и одинаковую топологию пайплайна.

Исходники Java

Исходники Kotlin

Структура пайплайна

  • reference_reader читает статический справочник;
  • reference_loader нормализует значения и записывает их в /reference_state через external_state_managers;
  • event_reader читает события из очереди;
  • enricher читает /reference_state через external_state_joiners и публикует успешные совпадения в enriched.

Менеджер и joiner ссылаются на одну динамическую таблицу стейта. Её ключи должны совпадать с group_by_schema компьютейшенов. См. спеку Java или спеку Kotlin.

Загрузка справочника

@Override
public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
    String name = message.get("name", String.class);
    String normalizedName = name.strip().toLowerCase();

    ExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
    stateAccessor.set(stateAccessor.getOrDefault().toBuilder()
            .set("normalized_name", normalizedName)
            .finish());
}

override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
    val normalizedName = message.get("name", String::class.java)?.trim()?.lowercase() ?: return

    val stateAccessor = ctx.getState(REFERENCE_STATE, message)
    stateAccessor.set(
        stateAccessor.orDefault.toBuilder()
            .set("normalized_name", normalizedName)
            .finish()
    )
}

Обогащение событий

Read-only дескриптор выбирает строку по ключу текущего сообщения. Если строка или normalized_name не найдены, пример не публикует результат.

@Override
public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
    ReadOnlyExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
    String normalizedName = stateAccessor.getOrDefault().get("normalized_name", String.class);
    if (normalizedName == null) {
        return;
    }

    output.addMessage(ctx.createMessageBuilder("enriched")
            .set("key", message.get("key", Long.class))
            .set("name", normalizedName)
            .finish());
}

override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
    val stateAccessor = ctx.getState(REFERENCE_STATE, message)
    val normalizedName = stateAccessor.orDefault.get("normalized_name", String::class.java) ?: return

    output.addMessage(
        ctx.createMessageBuilder("enriched")
            .set("key", message.get("key", Long::class.java))
            .set("name", normalizedName)
            .finish()
    )
}

Оба компьютейшена находятся по @FlowComputation; схемы трёх стримов, которые используют нативные источники и компаньонские функции, объявлены в спеке пайплайна.

См. также

Предыдущая
URL Downloader
Следующая
External State Join