Работа со стейтами в YTsaurus Flow (C++)

Все примеры пользовательской логики на этой странице используют process function. Не создавайте для работы со стейтом собственный класс-наследник Computation.

Internal State

Простейший способ хранить стейт внутри Computation. Данные автоматически подгружаются в начале эпохи и записываются при коммите. Не требует самостоятельного создания таблиц — Flow управляет ими автоматически.

Использование

Для работы с Internal State необходимо:

В качестве типа стейта можно использовать любой тип, для которого определены функции сериализации и десериализации. По умолчанию поддерживаются все типы, для которых определены стандартные функции Serialize/Deserialize в YT (в том числе TYsonStruct-наследники); для произвольного типа достаточно предоставить их перегрузки.

По умолчанию стейт считается пустым, если он равен значению по умолчанию (TMyState{}), а очистка пересоздаёт его. Чтобы переопределить это поведение, тип стейта может унаследовать опциональные миксины из NYT::NFlow:

• ICustomStateOps — собственная логика Clear() и IsEmpty() (задаются парой);
• ICustomYsonView — собственное представление ToYsonView() для read-state-интроспекции.

Необходимо:

1. Завести поле TMutableStateKeyClient<TMyState> MyStateClient_ в своей process function.
2. Переопределить Init(const IRuntimeInitContextPtr& initContext) и вызвать в нём initContext->InitClient<TMyState>(MyStateClient_, "my_state"). В качестве имени стоит взять уникальную в рамках компьютейшена строку.
3. Для получения стейта по ключу использовать MyStateClient_.GetState(message->Key). Возвращаемый аксессор TStateAccessor<TMyState> ведёт себя как умный указатель на TMyState (state->..., *state) и действителен только в пределах текущей эпохи — сохранять его в полях нельзя.
4. Для очистки стейта (удаления строки из таблицы) вызвать state.Clear() на аксессоре.

Если стейт пуст, соответствующая строчка будет удалена.

Пример:

struct TMyState
    : public NYTree::TYsonStruct
{
    std::optional<ui64> SomeValue;

    REGISTER_YSON_STRUCT(TMyState);

    static void Register(TRegistrar registrar)
    {
        registrar.Parameter("some_value", &TThis::SomeValue)
            .Default();
    }
};

class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitClient(MyStateClient_, "my_state");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = MyStateClient_.GetState(message->Key);
        state->SomeValue = 42;
        // ...
    }

    void ProcessTimer(
        const TInputTimerConstPtr& timer,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = MyStateClient_.GetState(timer->Key);
        // Очистить стейт (удалить строку из таблицы):
        state.Clear();
    }

private:
    TMutableStateKeyClient<TMyState> MyStateClient_;
};

Сжатие

Internal State поддерживает сжатие данных. Оно настраивается в DynamicSpec для каждого стейта по отдельности — в секции state_manager компьютейшена, в overrides/<имя_стейта>/format:

• compress — включить сжатие (по умолчанию false);
• recode_probability — вероятность перекодировать стейт в заданный формат при очередной обработке; обеспечивает постепенную миграцию стейтов после смены формата (по умолчанию 0.1).

Yson State Reader

External State

External State — стейт, хранимый в пользовательской динамической таблице YTsaurus. В отличие от Internal State, таблицы создаются и управляются пользователем.

External state manager объявляется в спеке Computation под уникальным именем в секции external_state_managers и подключается к Computation через типизированный клиент TMutableStateKeyClient<TState>. Имена в спеке должны начинаться с / (например, "/state") и совпадают с именем, передаваемым в InitExternalStateClient. Реализация ищется в реестре по external_state_manager_class_name.

Использование

Для работы с External State необходимо:

1. Завести поле TMutableStateKeyClient<TState> StateClient_ в своей process function, где TState — тип стейта, который возвращает соответствующий external state manager (см. ниже про конкретные реализации).
2. Переопределить Init(const IRuntimeInitContextPtr& initContext) и вызвать в нём initContext->InitExternalStateClient(StateClient_, "/state"). В качестве имени стоит взять уникальную в рамках компьютейшена строку — это же имя должно фигурировать в спеке.
3. Для получения стейта по ключу использовать StateClient_.GetState(message->Key). Возвращаемый аксессор TStateAccessor<TState> ведёт себя как умный указатель на TState и действителен только в пределах текущей эпохи.

Пример:

class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitExternalStateClient(StateClient_, "/state");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = StateClient_.GetState(message->Key);
        i64 count = state->GetColumnValue<std::optional<i64>>("count").value_or(0);
        TPayloadBuilder builder(state->Schema);
        builder.Set(count + 1, "count");
        state->Payload = builder.Finish();
    }

private:
    TMutableStateKeyClient<TSimpleExternalState> StateClient_;
};

Спека Computation с подключённым TSimpleExternalStateManager:

"external_state_managers" = {
    "/state" = {
        "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
        "parameters" = {
            "path" = "//path/to/state/table";
        };
    };
};

Предзагрузка и auto_preload

По умолчанию (auto_preload = %true) фреймворк перед обработкой эпохи подгружает в каждый external state manager все ключи сообщений, таймеров и визитов — по одному lookup'у на ключ. При "auto_preload" = %false; process function подгружает только нужные ключи через клиент: PreloadKeyStates(THashSet<TKey>), PreloadKeyStates(IInputContextPtr) или PreloadKeyStates(IInputContextPtr, TExtractKeysOptions); опции выбирают, ключи каких сущностей брать, например {.Visits = false} для стейта, который обработчик визитов не читает. Предзагрузка инкрементальна: её можно вызывать несколько раз за эпоху, подгружаются только ещё не загруженные ключи, а уже загруженный (возможно, изменённый) стейт остаётся как есть. GetState по ключу, не подгруженному в эпохе, бросает исключение. auto_preload = %false запрещён для companion-вычислений.

В IBatchProcessFunction::Process сущности можно раздать через хелперы из раздела Process functions:

void Process(const IInputContextPtr& input, const IOutputCollectorPtr& output, const IRuntimeContextPtr& context) override
{
    if (!context->GetSpec()->ExternalStateManagers.at("/state")->AutoPreload) {
        WaitFor(StateClient_.PreloadKeyStates(input, {.Visits = false})).ThrowOnError();
    }
    ProcessMessages(input, output, context, BIND(&TMyFunction::ProcessMessage, MakeStrong(this)));
    ProcessVisits(input, output, context, BIND(&TMyFunction::ProcessVisit, MakeStrong(this)));
}

Удаление без чтения: EraseState

StateClient_.EraseState(key) ставит строку на удаление, не подгружая и не читая её — key visitor, вычищающий строки по колонке ключа (например, timestamp), не платит за них ни одного lookup'а. Erase терминален для эпохи: после него GetState бросает исключение, а PreloadKeyStates не возвращает ключ обратно — независимо от того, был ли он загружен. TSimpleExternalStateManager, TProfileStateManager и внутренний per-key стейт удаляют только по ключу; пользовательский менеджер без переопределения EraseKeyState отвергает вызов.

for (const auto& visit : input->GetVisits()) {
    if (GetColumnValue<ui64>(visit->Key, "ts") + Week < now) {
        StateClient_.EraseState(visit->Key);   // Без lookup'а.
    }
}

TSimpleExternalStateManager

TSimpleExternalStateManager — стандартная реализация external state manager'а. Работает с одной динамической таблицей, ключи которой совпадают с group_by_schema. GetState отдаёт аксессор поверх TSimpleExternalState с полями Payload и Schema; колонки достаются и записываются через GetColumn[Value]<T> / TPayloadBuilder по имени или индексу. Кэширование стейтов происходит автоматически через общий StateCache.

Регистрировать TSimpleExternalStateManager в register.cpp не нужно: он уже зарегистрирован в самой библиотеке Flow.

Спека:

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

Динамическая спека:

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

В структуре нет основных параметров.

Полный пример использования TSimpleExternalStateManager см. в разборе примера word_count.

External State Joiner

External State Joiner предоставляет доступ только на чтение к внешним стейтам через join по ключу. Загруженные стейты кэшируются в общем StateCache с TTL: повторные обращения к тому же ключу до истечения TTL обслуживаются из кэша без обращения к YT.

External state joiner объявляется в спеке Computation под уникальным именем в секции external_state_joiners и подключается к Computation через типизированный клиент TJoinedStateKeyClient<TState>. Имена в спеке должны начинаться с / (например, "/reference") и совпадать с именем, передаваемым в InitExternalStateClient. Реализация ищется в реестре по external_state_joiner_class_name.

Использование

Аналогично external state manager'у:

class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitExternalStateClient(StateReaderClient_, "/reference");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = StateReaderClient_.GetState(message->Key);
        // state — read-only аксессор TConstStateAccessor<TSimpleExternalState> (действителен в пределах эпохи).
    }

private:
    TJoinedStateKeyClient<TSimpleExternalState> StateReaderClient_;
};

Спека Computation с подключённым TSimpleExternalStateJoiner:

"external_state_joiners" = {
    "/reference" = {
        "external_state_joiner_class_name" = "NYT::NFlow::TSimpleExternalStateJoiner";
        "parameters" = {
            "path" = "//path/to/reference/table";
        };
    };
};

TSimpleExternalStateJoiner

TSimpleExternalStateJoiner — стандартная реализация external state joiner'а. По интерфейсу аналогична TSimpleExternalStateManager: читает строки одной динамической таблицы, ключи которой совпадают с group_by_schema. GetState отдаёт read-only аксессор поверх TSimpleExternalState, который содержит TPayload и схему value-колонок таблицы; колонки достаются через GetColumn[Value]<T> по имени или индексу.

Регистрировать TSimpleExternalStateJoiner в register.cpp не нужно: он уже зарегистрирован в самой библиотеке Flow.

Спека:

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

Динамическая спека (TTL кэша задаётся через секцию cache):

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

TStaticTableKeyVisitorJoiner

TStaticTableKeyVisitorJoiner — external state joiner поверх статической сортированной таблицы. Работает только в паре с key-visitor-стримом: в отличие от TSimpleExternalStateJoiner, он не делает случайных чтений — фоновый обход визитора последовательно читает таблицу теми же диапазонами ключей, что и внутренний стейт, и строка таблицы становится доступна в ProcessVisit как read-only стейт ключа визита. Ключ, которого в таблице нет, отдаёт пустой стейт (IsEmpty() == true).

Джойнер привязывается к visit-стриму перечислением его имени в external_names секции key_visitor_streams (см. параметры статической спеки); один джойнер можно привязать не более чем к одному visit-стриму (проверяется при submit'е спеки). join_on/key_schema_override не поддерживается — ключом всегда служит собственная group_by_schema.

Требования к таблице-источнику:

• статическая — динамическая таблица отвергается при чтении схемы;
• отсортирована так, что префикс её ключевых колонок совпадает с group_by_schema Computation'а по именам и типам. Таблица может лежать на другом кластере (<cluster=...> в rich path).

Канонический паттерн — зеркалирование (reconciliation): ключи таблицы участвуют в обходе наравне с ключами собственного стейта, поэтому визит приходит и для ключа, которого в стейте ещё нет (его надо создать), и для ключа, которого уже нет в таблице (его надо удалить). Периодический обход таким образом сводит стейт Computation'а к внешней таблице:

class TMirrorFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitClient<TMirrorState>(MirroredState_, "/mirror");
        initContext->InitExternalStateClient(SourceState_, "/source");
    }

    void ProcessVisit(
        const TInputVisitConstPtr& visit,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto src = SourceState_.GetState(visit->Key);
        if (!src.IsInitialized()) {
            // Диапазон источника не прочитан (mark_unreadable): состояние ключа неизвестно.
            return;
        }
        auto mirror = MirroredState_.GetState(visit->Key);
        if (src.IsEmpty()) {
            mirror.Clear();
        } else {
            mirror->Payload = src->GetColumnValue<std::optional<std::string>>("payload").value_or(std::string{});
        }
    }

private:
    TMutableStateKeyClient<TMirrorState> MirroredState_;
    TJoinedStateKeyClient<TSimpleExternalState> SourceState_;
};

Спека Computation для этого примера:

"external_state_joiners" = {
    "/source" = {
        "external_state_joiner_class_name" = "NYT::NFlow::TStaticTableKeyVisitorJoiner";
        "parameters" = {
            "path" = "//path/to/source";
        };
    };
};
"key_visitor_streams" = {
    "visit_iter" = {
        "names" = ["/mirror"];
        "external_names" = ["/source"];
    };
};

Поведение при недоступности источника задаёт unavailable_source_policy. Неуспешным считается чтение, исчерпавшее read_attempts попыток:

• retry (по умолчанию) — ошибка роняет итерацию обхода, чтение повторяется; обход не продвигается за непрочитанный диапазон.
• mark_unreadable — ошибка проглатывается, обход идёт дальше, а ключи непрочитанного диапазона резолвятся неинициализированным аксессором: IsInitialized() == false, разыменование бросает исключение. Computation сам решает, что делать с ключом, состояние которого в источнике неизвестно (обычно — пропустить визит, как в примере выше).

После неуспешного чтения источник помечается недоступным на unavailable_source_backoff: пока пометка действует, обращений к источнику нет — чтения немедленно резолвятся согласно полиси; первое чтение после истечения окна снова пробует источник. Состояние джойнера видно в сенсорах воркера static_table_key_visitor_joiner/{source_unavailable,failed_reads,reader_opens,listed_size} с тегом external_state_joiner=<имя>.

Регистрировать TStaticTableKeyVisitorJoiner в register.cpp не нужно: он уже зарегистрирован в самой библиотеке Flow.

Спека:

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

Динамическая спека:

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

State Joiner

State Joiner предоставляет доступ только на чтение к внутреннему стейту другого Computation'а: один Computation может обогащаться стейтом, который накапливает другой Computation, в том числе когда ключ join'а не совпадает с собственной group_by_schema.

Joiner объявляется в спеке Computation под уникальным именем в секции state_joiners и подключается через типизированный клиент TJoinedStateKeyClient<TState>. Имена в спеке должны начинаться с / и совпадать с именем, передаваемым в InitClient. Отдельной реализации или регистрации не требуется — это встроенная возможность Flow.

В спеке указывается:

• computation_id — Computation, чей внутренний стейт читается;
• state_name — имя стейт-клиента целевого Computation'а (тот префикс, который он передал в InitClient, начинается с /);
• join_on/key_schema_override — какими колонками текущей строки задаётся ключ. Если не задана, используется собственная group_by_schema (join по тому же ключу);
• join_on/key_provider_streams — стримы, из сообщений и таймеров которых берутся ключи (nullopt — все входные стримы);
• auto_preload — если true (по умолчанию), фреймворк сам подгружает ключи перед каждым DoProcess; иначе Computation вызывает PreloadKeyStates сам.

Использование

class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitClient(UpstreamClient_, "/upstream");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = UpstreamClient_.GetState(message);
        // state — read-only аксессор TConstStateAccessor<TUpstreamState> (действителен в пределах эпохи).
    }

private:
    TJoinedStateKeyClient<TUpstreamState> UpstreamClient_;
};

Спека Computation с подключённым joiner'ом по ключу, отличному от собственной group_by_schema:

"state_joiners" = {
    "/upstream" = {
        "computation_id" = "accumulator";
        "state_name" = "/total";
        "join_on" = {
            "key_schema_override" = [
                {name = "Hash"; expression = "farm_hash(UserId)"; type = "uint64"; required = %true;};
                {name = "UserId"; type = "string"; required = %true;};
            ];
        };
    };
};

По умолчанию кэш выключен (ttl = 0): joiner заново читает стейт целевого Computation'а на SyncLastCommittedTimestamp каждую эпоху, поэтому всегда видит свежие коммиты. Кэш с TTL включается в dynamic spec через секцию cache (загруженные стейты живут до истечения ttl, повторные чтения того же ключа обслуживаются без обращения к YT). Для Swift-вычислений с требованием строгой детерминированности кэш и чтение на SyncLastCommittedTimestamp нужно учитывать.

Спека:

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

Динамическая спека (TTL кэша задаётся через секцию cache):

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

StateCache

Flow предоставляет общий двухуровневый (uncompressed + compressed) LRU кэш для стейтов. Конфигурирование осуществляется через /dynamic_spec/job_tracker/state_cache. Он используется в первую очередь для стейтов менеджеров; у joiner'ов кэш по умолчанию выключен, а данные грузятся на SyncLastCommittedTimestamp.

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

Подробнее о согласованности group_by_schema и общих принципах работы со стейтами см. Stateful processing.

См. также

• Stateful processing (концепция)
• Computation (C++)
• Быстрый старт (C++)