Stateful processing в YTsaurus Flow

Stateful обработка — это обработка событий, включающая в себя read-modify-write работу со стейтами, хранящимися в YTsaurus. Простейший пример: подсчёт статистик по входящим событиям в разрезе по ключам — нужно загрузить старое значение, обновить и записать обратно.

Модель работы со стейтом

Доступ к стейту внутри компьютейшена строится из трёх сущностей:

Стейт (state) — пользовательские данные, привязанные к ключу и хранящиеся в динамической таблице YTsaurus.
Клиент стейта (state client) — типизированный по типу стейта объект (Client<TState>), который process function заводит по одному на каждый именованный стейт и инициализирует в Init. Через клиент происходит доступ к стейту по ключу. Бывает read-write (работает и с внутренним, и с внешним стейтом) и read-only (джойнер к внешнему стейту).
Аксессор (state accessor) — то, что клиент отдаёт для конкретного ключа (по входному сообщению, таймеру или явному ключу): представление стейта того же типа TState. Аксессор ведёт себя как умный указатель на стейт: read-write-аксессор позволяет читать, изменять и очищать стейт, read-only — только читать.

Важно

Аксессор действителен только в пределах текущей эпохи. Его нельзя сохранять в полях process function или переиспользовать между эпохами — на каждой эпохе стейт нужно получать через клиент заново.

Типы стейтов

Internal State

Простейший способ работы со стейтом: таблицы создавать не нужно — Flow управляет ими автоматически. Данные подгружаются в начале эпохи и записываются при коммите. Доступ read-write — тем же клиентом, что и для внешнего стейта. Тип стейта может быть произвольным; единственное требование — сериализуемость в YSON (для сохранения между эпохами).

External State

Стейт в пользовательской динамической таблице. Таблицы создаются и управляются пользователем. Доступ read-write — тем же клиентом, что и для внутреннего стейта, но бэкендом выступает state manager (TSimpleExternalStateManager). Объявляется в спеке компьютейшена в top-level секции external_state_managers, реализация резолвится по external_state_manager_class_name. Поддерживает кэширование.

External State Joiner

Read-only доступ к внешним стейтам через join по ключу — через joiner (TSimpleExternalStateJoiner). Объявляется в спеке компьютейшена в top-level секции external_state_joiners (на одном уровне с external_state_managers), реализация резолвится по external_state_joiner_class_name. Поддерживает кэширование с TTL: загруженные стейты живут в общем StateCache и переподгружаются из YT только после истечения TTL либо вытеснения из кэша.

Важно

Таблицы, к которым даёт доступ read-write state manager, должны модифицироваться исключительно через него (или при остановленном пайплайне), так как он может использовать кэши.

Одна таблица — один писатель

Писать в таблицу внешнего стейта должен ровно один компьютейшен: записи из разных партиций и транзакций нарушают консистентность стейта. State manager владеет своей таблицей на запись: TSimpleExternalStateManager объявляет её как принадлежащую ему. Read-only доступ к стейту другого компьютейшена нужно делать через external state joiner (TSimpleExternalStateJoiner) либо сообщениями компьютейшену-писателю — джойнеры таблицу не захватывают. Это проверяется валидацией спеки по владению на запись: на каждую таблицу стейта должен приходиться ровно один писатель-владелец, а пайплайн, где одну таблицу захватывают на запись два менеджера, отклоняется с ошибкой State table <путь> is claimed for writing by both .... Read-only потребители (джойнеры) таблицу на запись не захватывают, поэтому могут разделять её с писателем-владельцем.

Примечание

Для любого стейта пустое значение соответствует отсутствию строки в таблице. Если после модификации стейт пуст, соответствующая строка удаляется. По умолчанию пустота и очистка стейта определяются автоматически (сравнение со значением по умолчанию); тип стейта может переопределить это поведение.

Хранение стейтов

Стейты хранятся в динамических таблицах YTsaurus. Простейший пример схемы:

name	type	sort_order	expression
`hash`	`uint64`	`ascending`	`farm_hash(my_key)`
`my_key`	`string`	`ascending`
`my_value_1`	`string`
`my_value_2`	`string`

Согласованность group_by_schema

Из соображений корректности и производительности строго рекомендуется для Computation в качестве group_by_schema использовать схему из первых ключевых колонок динтаблицы со стейтами (строго префикс ключевых колонок). Тогда:

с событиями по одному ключу будет работать только одна партиция (корректность);
одна партиция будет работать с ограниченным числом таблетов (производительность).

Пример group_by_schema, согласованного со схемой таблицы стейтов из примера выше:

name	type	expression
`hash`	`uint64`	`farm_hash(my_key)`
`my_key`	`string`

StateCache

Flow предоставляет общий двухуровневый (uncompressed + compressed) LRU кэш для стейтов. Конфигурирование: /dynamic_spec/job_tracker/state_cache.

Источник: yt/yt/flow/library/cpp/common/spec.h

Параметр	Описание
`compressed_cache_weight`	Тип: NYT::NYTree::TSize Значение по умолчанию: `1Gi` Ограничение на `compressed`.
`uncompressed_cache_weight`	Тип: NYT::NYTree::TSize Значение по умолчанию: `100Mi` Ограничение на `uncompressed`.

Реализация на разных языках

C++: клиент TMutableStateKeyClient<TState> (read-write) или TJoinedStateKeyClient<TState> (read-only) отдаёт аксессор TStateAccessor<TState> / TConstStateAccessor<TState>; один и тот же key-client работает и с внутренним, и с внешним стейтом. Подробнее →
Java: YsonStateAccessor, ProtoStateAccessor, ExternalStateAccessor. Подробнее →
Python: ctx.state(), ctx.external_state(), ctx.proto_state(). Подробнее →
Go: flow.OpenYSONState, flow.OpenProtoState, flow.OpenExternalState, flow.OpenJoinedExternalState открывают аксессор к стейту по ключу входа. Подробнее →

См. также

Swift

Вотермарки (Watermarks)