--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/concepts/distributed_throttler.md - https://ytsaurus.tech/docs/ru/flow/concepts/distributed_throttler.md - href: https://ytsaurus.tech/docs/ru/flow/concepts/distributed_throttler.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Distributed Throttler в YTsaurus Flow Распределённый троттлер — глобальный для всего [пайплайна](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline) механизм ограничения скорости. На контроллере для каждого именованного троттлера держится [token bucket](https://en.wikipedia.org/wiki/Token_bucket); [джобы](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#job) запрашивают у контроллера квоту перед обработкой сообщений или перед произвольным пользовательским действием. Квотироваться может либо число обработанных сообщений, либо суммарный размер их payload в байтах — отдельно или вместе. Типичные задачи: - ограничить суммарную нагрузку на внешний API со стороны всех джобов; - равномерно распределить пропускную способность между [партициями](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#partition) одного [`Computation`](https://ytsaurus.tech/docs/ru/flow/concepts/computation.md); - притормозить чтение из источника. ## Приоритет по отставанию {#priority} Каждая джоба прикладывает к запросу за квотой свою временную метку — ту же, по которой во входном буфере определяется отставание. Берётся минимум по всем `stabilized_event_timestamp + stream_delay` из [input streams](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream) и `read_alignment_timestamp` из [source streams](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#source). Сервер ранжирует запросы по возрастанию этой метки: сначала отдаёт квоту тем, кто отстаёт сильнее. При перекосе нагрузки между [партициями](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#partition) одного [`Computation`](https://ytsaurus.tech/docs/ru/flow/concepts/computation.md) отстающие автоматически получают больше квоты, а лидирующие — замедляются. ## Взвешенные классы квоты {#quota-classes} У именованного троттлера можно объявить классы с весами. Между классами, в которых есть ожидающие запросы, квота распределяется пропорционально весам и с полным использованием доступной пропускной способности. Если класс простаивает, его доля немедленно перераспределяется между активными классами. Вес задаёт долгосрочную долю полосы, а не строгий приоритет. Приоритет по отставанию продолжает действовать внутри каждого класса. Простаивающий класс не накапливает кредит, но после появления запроса получает возможность выдачи до следующего большого прохода по непрерывному бэклогу другого класса. Если `classes` отсутствует, сохраняется прежнее поведение с одной очередью и прежняя семантика token bucket. Зарезервированный класс `default` всегда имеет вес `1.0`. Отсутствующий, пустой или неизвестный `quota_class_id` на сервере попадает в `default`; так обслуживаются троттлеры, которым класс не задан. Класс относится ко всем запросам компьютейшена к этому троттлеру: если тот же id получен вручную через `GetThrottlerOrThrow`, эти запросы идут в том же классе, а не в `default`. Класс задаётся отдельно для каждого автоматического троттлера — `input_rows_throttler_class_id` и `input_bytes_throttler_class_id`, — и каждый проверяется против классов своего троттлера: неизвестный класс отклоняется валидацией до запуска пайплайна. ```yson "dynamic_spec" = { "throttlers" = { "output_quota" = { "limit" = 1000.0; "period" = "1s"; "request_period" = "500ms"; "max_grant_amount" = 50; "classes" = { "vip" = {"weight" = 5.0;}; "regular" = {"weight" = 3.0;}; "bulk" = {"weight" = 1.0;}; }; }; }; "computations" = { "ReaderVip" = { "input_rows_throttler_id" = "output_quota"; "input_rows_throttler_class_id" = "vip"; }; }; } ``` ### Веса как абсолютные скорости Параметр `use_class_weights_as_limit = %true` делает скоростью бакета сумму объявленных весов классов, поэтому вес читается как единицы квоты в секунду, а не как доля от отдельно заданного `limit`. С весами `5.0`, `3.0` и `1.0` троттлер выдаёт 9 единиц в секунду, и пока все классы забэкложены, каждый обслуживается со скоростью своего веса. Work conservation при этом сохраняется: пока остальные простаивают, класс по-прежнему выходит за свой вес, то есть вес — это гарантия при конкуренции, а не потолок. Параметр требует хотя бы одного класса и несовместим с `limit` — если заданы оба, конфигурация отклоняется валидацией. Зарезервированный класс `default` в сумму не входит. Запросы без класса на таком троттлере конкурируют с весом `1.0` поверх объявленной скорости и опускают каждый объявленный класс ниже его веса; это видно по сенсорам троттлера `/classless_requests` и `/unknown_class_requests`. ### Задержка переключения между классами Планировщик пересматривает активные классы после завершения текущего серверного чанка. `max_grant_amount` задаёт максимальный размер чанка в абсолютных единицах квоты; если его не задать, один запрос удерживает token bucket всё время своего prefetch-окна и ровно на столько же задерживает остальные классы. В наблюдаемую задержку также входят `drain_period`, RPC, локальный prefetch, `empty_batch_backoff` источника и commit Flow. ### Веса и пропускная способность ридеров Наблюдаемые доли совпадают с настроенными весами, только пока каждый измеряемый класс забэкложен **и** каждый ридер способен выбрать положенную ему долю. Итерация ридера несёт фиксированные накладные расходы, не зависящие от выданной скорости, поэтому при небольшом `max_rows_per_batch` высоковесный класс упирается в собственный потолок пропускной способности раньше, чем добирает свою долю; work conservation отдаёт остаток низковесным классам, и наблюдаемое распределение смещается в сторону равного. Масштабируйте `max_rows_per_batch` вместе с `limit`, чтобы накладные расходы итерации оставались малой долей времени ожидания квоты. ## Реконфигурация на лету {#reconfigure} Все параметры троттлера (`limit`, `period`, `request_period`, `retrying_channel`, `rpc_timeout`, `classes`, `max_grant_amount`, `use_class_weights_as_limit`) подхватываются без перезапуска пайплайна — достаточно обновить `dynamic_spec`. Кэшированный в пользовательском коде `IThroughputThrottlerPtr` остаётся валидным после смены конфига. Вес и размер серверного чанка меняются без перестроения prefetch-клиента. Исключение — `use_class_weights_as_limit`: тогда вес входит в скорость, поэтому изменение веса, меняющее размер prefetch-окна, приводит к перестроению клиента, а слишком малое для этого изменение оставляет клиента и уже полученную им квоту на месте. Новая скорость в любом случае применяется на сервере. Смена `input_rows_throttler_class_id` или `input_bytes_throttler_class_id` влияет на последующие RPC; уже полученная локально или находящаяся в полёте квота остаётся учтённой в прежнем классе. ## Классы квоты и watermark alignment {#watermark-alignment} Доступность квоты не отменяет watermark alignment: чтение разрешено, только если одновременно выполнены выходные лимиты, alignment и ограничение окна чтения. Поэтому класс с большим весом не ускорит чтение, если источник придерживает alignment, — вес распределяет только квоту. Настройка самого alignment описана в [watermark strategy](https://ytsaurus.tech/docs/ru/flow/concepts/watermarks.md). ## Что будет, если контроллер недоступен {#controller-unavailable} Клиент троттлера переоткрывает соединение через `retrying_channel`. Пока контроллер не отвечает, локальный prefetch-кэш доедается, и `Throttle()` блокируется в ожидании квоты — джоба прогресса не делает. Когда контроллер возвращается, ожидание разрешается само. Если же контроллер не отвечает дольше, чем суммарный бюджет ретраев `retrying_channel`, работа с троттлером выкинет исключение и джоба упадёт. ## Что будет, если квоты не хватает {#quota-insufficient} Если суммарный спрос джоб на много порядков превышает возможную выдачу, отдельные `RequestQuota` могут долго стоять в серверной очереди и в итоге не уложиться в свой `rpc_timeout` × `retry_attempts`. В таком случае `Throttle()` тоже выкинет исключение, что приведет к падению джобы. Fallback на локальный лимит ни в одном из этих случаев сейчас не поддержан. ## Конфигурация {#configuration} Все настройки — в `DynamicSpec`. Троттлеры объявляются в корне. В примере ниже `external_api_quota` — пользовательский id троттлера, `RequestEnricher` — имя computation: ```yson "dynamic_spec" = { "throttlers" = { "external_api_quota" = { "limit" = 100.0; }; }; "computations" = { "RequestEnricher" = { "input_rows_throttler_id" = "external_api_quota"; }; }; } ``` Поля `TDynamicThrottlerSpec`: Источник: [yt/yt/flow/library/cpp/common/spec.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/spec.h) Конфигурация именованного распределённого троттлера. #| || **Параметр** | **Описание** || || `limit` | **Тип**: `std::optional` Средняя скорость выдачи квоты в единицах в секунду. `none` означает `unlimited`. || || `period` | **Тип**: [TDuration](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#TDuration) **Значение по умолчанию**: `10s` Период усреднения средней скорости в алгоритме [token bucket](https://en.wikipedia.org/wiki/Token_bucket): за `period` бакет пополняется до объёма `limit * period` единиц. Этот же объём — максимально возможный burst. || || `request_period` | **Тип**: [TDuration](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#TDuration) **Значение по умолчанию**: `5s` Целевой интервал между походами клиента на контроллер (размер prefetch подстраивается под него). || || `retrying_channel` | **Тип**: `NYT::TIntrusivePtr<`[NYT::NRpc::TRetryingChannelConfig](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NRpc_TRetryingChannelConfig)`>` **Значение по умолчанию**: ```yson { "enable_exponential_retry_backoffs" = %true; "retry_attempts" = 100; "retry_timeout" = 600000; } ``` Параметры ретраев запросов к сервису троттлера. Дефолт рассчитан на то, чтобы пережить смену контроллера-лидера даже если она затянется. || || `rpc_timeout` | **Тип**: [TDuration](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#TDuration) **Значение по умолчанию**: `30s` Таймаут одного запроса `RequestQuota`. || || `classes` | **Тип**: `THashMap, NYT::TIntrusivePtr<`[NYT::NFlow::TDynamicThrottlerClassSpec](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NFlow_TDynamicThrottlerClassSpec)`>>` **Значение по умолчанию**: `{}` Именованные взвешенные классы квоты. Ключи должны соответствовать `[0-9A-Za-z_-]+`. Активные классы делят полосу пропорционально весам, а свободная доля перераспределяется. || || `max_grant_amount` | **Тип**: `std::optional` Максимальный размер одного серверного чанка в абсолютных единицах квоты. Ограничивает задержку пересмотра активных классов. Если не задан, запрос выдаётся целиком и удерживает token bucket всё своё prefetch-окно, задерживая остальные классы ровно на это время. || || `use_class_weights_as_limit` | **Тип**: `bool` **Значение по умолчанию**: `false` Трактует веса классов как абсолютные скорости: скорость выдачи становится суммой объявленных весов, поэтому забэкложенный класс обслуживается со скоростью своего веса в единицах в секунду. Требует хотя бы одного класса и несовместим с `limit`. Зарезервированный класс `default` в сумму не входит. || |# ## Применение {#usage} Есть два варианта — автоматический и ручной. Они не конфликтуют: можно пользоваться одним или сразу обоими. ### Автоматический: ограничить скорость входного батча {#auto} В `TDynamicComputationSpec` есть два поля — `input_rows_throttler_id` и `input_bytes_throttler_id`. Если они заданы, перед каждой итерацией `Process` компьютейшен ждёт квоту: - по числу сообщений во входном батче (`input_rows_throttler_id`); - по сумме `byte_size` сообщений батча — это системный размер сериализованного представления, без учёта компрессии/шифрования на сетевом уровне (`input_bytes_throttler_id`). Ожидание оформляется отдельным span'ом `Input.Throttle` в трейсинге компьютейшена (видно в Jaeger и на графиках "Epoch parts time" в UI). Id должен быть объявлен в `dynamic_spec/throttlers`. Можно задать оба поля сразу — тогда итерация ждёт обе квоты, — но они должны ссылаться на **разные** троттлеры: один троттлер меряет либо число сообщений, либо байты, и общий token bucket складывал бы штуки с байтами. Такая спека отклоняется валидацией. ### Ручной: `GetThrottlerOrThrow(id)` из пользовательского кода {#manual} Если автоматического троттлинга по входному батчу недостаточно (например, нужен rate-limit на каждый внешний запрос), process function получает троттлер из `IRuntimeContext`. Метод `context->GetThrottlerOrThrow(id)` возвращает `IThroughputThrottlerPtr` — стандартный YT-троттлер, у которого вызывают `Throttle(amount)` и ждут результат. Если id нет в `throttlers`, `GetThrottlerOrThrow` бросает исключение; `context->TryGetThrottler(id)` в этом случае возвращает `nullptr` — удобно, когда сам факт объявления троттлера в `dynamic_spec` включает троттлинг для соответствующей сущности (например, троттлер на каждую модель или внешний сервис). Возвращённый указатель стабилен в рамках всей жизни джоба: фабрика подменяет внутренний клиент при `Reconfigure`, поэтому process function может сохранить его в `Init` и использовать дальше без переполучения. Это не относится к `nullptr` от `TryGetThrottler`: он означает лишь «сейчас не объявлен», и после добавления троттлера в `dynamic_spec` его нужно запросить заново. Сейчас этот механизм поддержан только в C++. ## Мониторинг {#monitoring} У дашборда Flow есть отдельный таб **Distributed throttler**. Основные сенсоры: - серверные (контроллер): `value.rate` — фактическая выдача квоты, `queue_size` — сколько запросов ждут, `wait_time.max` — хвост ожидания. - клиентские (воркер, на каждом `Computation`): `consumed.rate` / `released.rate`, `wait_time.max` — время, которое джоба простояла в `Throttle()`, `request_count.rate` — частота prefetch-RPC. ## См. также {#see-also} - [Spec, DynamicSpec и Config](https://ytsaurus.tech/docs/ru/flow/concepts/spec.md) - [Computation](https://ytsaurus.tech/docs/ru/flow/concepts/computation.md)