Word Count в YTsaurus Flow (Python)

Простейший пример stateful-пайплайна на Python: подсчёт количества вхождений каждого слова с использованием внутреннего YSON-стейта.

Исходный код

Структура

Пайплайн состоит из одного transform-компьютейшена mapper, который читает слова из входного стрима и обновляет счётчик в стейте.

__main__.py

Точка входа: создание пайплайна и регистрация единственного компьютейшена.

def main():
    log.info("Starting companion execution")

    pipeline = Pipeline()
    pipeline.add("mapper", WordCountMapper())

    log.info("Starting pipeline...")
    pipeline.run()
    log.info("Pipeline completed")

word_count_mapper.py

RowFunction, использующая ctx.state() для работы с YSON-стейтом. Для каждого ключа сообщения хранится словарь с полями word и count.

class WordCountMapper(RowFunction):
    """Processes Word messages and maintains word count state."""

    def on_message(self, message, output, ctx):
        state = ctx.state("word-state", message)
        data = state.get_or_default({"word": message.payload["word"], "count": 0})
        data["count"] += 1
        state.set(data)

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

  • Простейший stateful-пайплайн с одним компьютейшеном.
  • Внутренний YSON-стейт через ctx.state() с get_or_default / set.
  • Ключ стейта определяется group_by_schema из спеки (в данном случае -- по полю word).

См. также

Предыдущая
Тестирование
Следующая
Shuffle