Spec, DynamicSpec и Config в YTsaurus Flow

Спека — основной способ описания желаемого пайплайна. Любой высокоуровневый API к Flow генерирует спеку пайплайна.

У пайплайна есть две спеки — Spec и DynamicSpec:

  • Spec — статический объект, содержащий топологию всего пайплайна. Этот объект можно менять только при условии остановки пайплайна. Гарантируется, что в момент выполнения вся система имеет одинаковое представление о Spec.
  • DynamicSpec — динамическая часть спеки. Её можно менять в любой момент, она применится ко всем частям системы асинхронно. В DynamicSpec обычно хранятся размеры буферов, таймауты, число партиций и т. п.

Также важной частью является Config ноды. В отличие от спеки, конфиг задаётся ровно один раз в рамках релиза и не подлежит динамическому изменению. Возможно, в будущем часть параметров конфига будут перемещены в спеки.

Важно

Все параметры во Flow имеют оптимальные значения по умолчанию. Не меняйте настройки, если вы не уверены, для чего они нужны и на что могут повлиять.

Значения по умолчанию можно найти поиском по имени параметра в кодовой базе Flow. В основном они сгруппированы в файле spec.cpp.

Spec

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

Параметр

Описание

validate_binary_version

Тип: bool
Значение по умолчанию: false

binary_version

Тип: std::string
Значение по умолчанию: TString("")

validate_binary_checksum

Тип: bool
Значение по умолчанию: false

binary_checksum

Тип: std::string
Значение по умолчанию: TString("")

computations

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TComputationIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TComputationSpec>>
Значение по умолчанию: {}
Именованное перечисление всех узлов пайплайна. Ключи должны соответствовать [0-9A-Za-z_-]+.

resources

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TResourceIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TResourceSpec>>
Значение по умолчанию: {}
Именованное перечисление всех ресурсов. Ключи должны соответствовать [0-9A-Za-z_-]+.

streams

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TStreamSpec>>
Значение по умолчанию: {}
Именованное перечисление всех потоков данных. Ключи должны соответствовать [0-9A-Za-z_-]+.

Computation

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

Параметр

Описание

computation_class_name

Тип: std::string
Обязательный параметр
Имя встроенного класса Computation или адаптера process function.

Для новой пользовательской логики на C++ выберите один из встроенных адаптеров TProcessFunction*Computation. Саму функцию зарегистрируйте через YT_FLOW_DEFINE_PROCESS_FUNCTION и укажите в processing_function. Пользовательские классы Computation создавать не следует.

process_function_graph

Тип: NYT::TIntrusivePtr<NYT::NFlow::TProcessFunctionGraphSpec>
Несколько process function, связанных идентификаторами стримов и выполняемых в одном TProcessFunctionComputation или TProcessFunctionSwiftMapComputation. Можно объединять batch и per-message функции. ID внутренних сообщений выводятся из исходных родителей на границе графа, выходного стрима и suffix; timestamps и внешние ID назначаются по правилам host computation. Swift требует детерминированного replay; batch функции задают родителей каждого результата через SetParents. Объединение разных входных сообщений требует parameters/allow_batching_with_relaxed_guarantees = %true и даёт гарантию at-least-once. Раннер заменяет граф полями processing_function, processing_function_parameters, timer_streams, streams_dependency и allow_timer_self_dependency, поэтому не задавайте эти поля вместе с графом.

processing_function

Тип: std::optional<std::string>
Полное имя process function, зарегистрированной через YT_FLOW_DEFINE_PROCESS_FUNCTION. Обязательно для адаптеров TProcessFunction*Computation, если не задан process_function_graph.

processing_function_parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Статические параметры process function. Тип параметров задаётся при регистрации YT_FLOW_DEFINE_PROCESS_FUNCTION.

group_by_schema

Тип: NYT::TIntrusivePtr<NYT::NTableClient::TTableSchema>
Значение по умолчанию: {'value': [], 'attributes': {'strict': true, 'unique_keys': false}}
Схема для группировки всех входных потоков, является схемой динтаблицы. Свойства и требования к схеме:

  • Не содержит никаких правил на порядок.
  • Может содержать вычислимые поля.
  • Должна быть пустой только у SourceComputation.
  • Для остальных Computation первая колонка должна иметь тип uint64. Рекомендуется, чтобы эта колонка была вычислимым хешем от остальных колонок (например, farm_hash). Нужно иметь ввиду, что использование операции взятия остатка при целочисленном делении % и bigb_hash в expression запрещено.

experimental_enable_non_uint_key

Тип: std::optional<bool>

input_stream_ids

Тип: THashSet<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>>
Значение по умолчанию: []
Входные потоки.

Схемы всех потоков должны быть приводимы к group_by_schema.

output_stream_ids

Тип: THashSet<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>>
Значение по умолчанию: []
Выходные потоки Computation'а. В частности, могут использоваться как входы у других Computation'ов.

streams_dependency

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, THashSet<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>>>
Значение по умолчанию: {}
Описание зависимостей между потоками, включая внутренние source и timer потоки.

Пример: {stream1, {stream2, stream3}} — stream1 зависит от потоков stream2 и stream3.

Указывается только для timer- и output-потоков. В случае отсутствия описания для потока список зависимостей будет сгенерирован автоматически:

  • у timer-потоков в списке будут все input- и source-потоки;
  • у output-потоков — все timer, input и source.

watermark_strategy

Тип: NYT::TIntrusivePtr<NYT::NFlow::TWatermarkStrategySpec>
Значение по умолчанию: {}
Настройки, связанные с EventWatermark.

Необязательный параметр. Должен быть указан в рамках SourceComputation.

required_resource_ids

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TResourceIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TResourceDescription>>
Значение по умолчанию: {}
Список необходимых ресурсов для работы Computation.

Имеет параметр alias для локального переименования ресурса (например, Computation ожидает ресурс с именем YTClient, а подаваемый ресурс имеет другое имя). Через параметры controller и worker можно управлять созданием ресурса на Controller и на Worker.

parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Значение по умолчанию: {}
Произвольные параметры для Computation и ComputationController

timer_streams

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TTimerSpec>>
Значение по умолчанию: {}
Настройки всех timer потоков. Ключи должны соответствовать [0-9A-Za-z_-]+.

key_visitor_streams

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TKeyVisitorStreamSpec>>
Значение по умолчанию: {}
Настройки всех key_visitor потоков. Ключи должны соответствовать [0-9A-Za-z_-]+.

source_streams

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TSourceSpec>>
Значение по умолчанию: {}
Настройки всех source потоков. Ключи должны соответствовать [0-9A-Za-z_-]+.

sinks

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TSinkIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TSinkSpec>>
Значение по умолчанию: {}
Настройки всех sink. Ключи должны соответствовать [0-9A-Za-z_-]+.

external_state_managers

Тип: THashMap<std::string, NYT::TIntrusivePtr<NYT::NFlow::TExternalStateManagerSpec>>
Значение по умолчанию: {}
Декларативное объявление external state менеджеров для этого Computation. Ключ — имя клиента, которое process function передаёт в InitExternalStateClient (должен начинаться с /, например /state); значение содержит имя класса менеджера и его параметры.

external_state_joiners

Тип: THashMap<std::string, NYT::TIntrusivePtr<NYT::NFlow::TExternalStateJoinerSpec>>
Значение по умолчанию: {}
Декларативное объявление external state joiner'ов (read-only-доступ к внешним стейтам через join по ключу) для этого Computation. Ключ — имя клиента, которое process function передаёт в InitExternalStateClient (должен начинаться с /, например /state); значение содержит имя класса joiner'а и его параметры.

state_joiners

Тип: THashMap<std::string, NYT::TIntrusivePtr<NYT::NFlow::TStateJoinerSpec>>
Значение по умолчанию: {}
Декларативное объявление state joiner'ов (read-only-доступ к внутреннему стейту другого Computation через join по ключу) для этого Computation. Ключ — имя клиента, которое process function передаёт в InitClient (должен начинаться с /); значение указывает целевой computation_id, его state_name и join_on.

heavy_hitters

Тип: NYT::NFlow::THeavyHittersSpec
Значение по умолчанию: {}
Настройки обнаружения высокочастотных ключей.

pivot_finder

Тип: NYT::NFlow::TPivotFinderSpec
Значение по умолчанию: {}

input_ordering

Тип: NYT::TIntrusivePtr<NYT::NFlow::TInputOrderingSpec>
Значение по умолчанию: {}
Настройки порядка обработки входных потоков.

distribution_ordering

Тип: NYT::NFlow::EDistributionOrdering
Значение по умолчанию: strict
Настройки порядка распределения выходных сообщений. Управляет тем, может ли быть гонка между останавливающимися и работающими партициями с точки зрения порядка распределения выходных сообщений порожденных из входящих сообщений с одним и тем же ключом.

worker_group

Тип: NYT::TStrongTypedef<std::string, NYT::NFlow::TWorkerGroupIdTag, NYT::TStrongTypedefOptions{true}>
Значение по умолчанию: TString("")

allow_timer_self_dependency

Тип: bool
Значение по умолчанию: false

use_compact_input_messages

Тип: std::optional<bool>
Использовать компактную таблицу compact_input_messages для дедупликации входных сообщений вместо input_messages. Компактный ключ дедупликации хранит только хеш key[0] и CityHash128(message_id), что снижает размер строки (~200 байт → ~71 байт).

Необязательный параметр. Если значение не задано, таблица выбирается автоматически: компактная используется всегда, кроме случая, когда включён experimental_enable_non_uint_key (тогда первая колонка ключа может быть не uint64, и для дедупликации нужен полный ключ). Явно заданное значение имеет приоритет над автоматическим выбором. Включать use_compact_input_messages вместе с experimental_enable_non_uint_key запрещено.

WatermarkStrategy

Рекомендуется заполнять у SourceComputation. Остальные Computation могут не поддерживать или поддерживать ограниченно.

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

Параметр

Описание

event_timestamp_assigner

Тип: NYT::TIntrusivePtr<NYT::NFlow::TEventTimestampAssignerSpec>
Значение по умолчанию: {}
Автоматическое заполнение EventTimestamp всех output сообщений на базе значения определённой колонки.

watermark_generator

Тип: NYT::TIntrusivePtr<NYT::NFlow::TWatermarkGeneratorSpec>
Настройки алгоритма генерации EventWatermark. Для Computation хотя бы с одним source отсутствие параметра эквивалентно пустому блоку с настройками по умолчанию. Computation с несколькими source обрабатывается как несколько независимых single-source Computation: для каждого source по отдельности и в этом порядке применяются эвристики недоступных групп, idle-партиций и late-data-партиций, а подготовленные партиции объединяются только после этого. idle_partitions.max_ratio применяется отдельно к каждому source, а решения о подавлении availability-групп одного source не влияют на другие. Поэтому полностью idle source не считается partial-idle stall и сам по себе не поднимает /idle_partitions_watermark_stall, даже если он удерживает итоговый watermark Computation, пока другой source остаётся активным. Идентификаторы availability-групп между источниками могут различаться.

watermark_alignment

Тип: NYT::TIntrusivePtr<NYT::NFlow::TWatermarkAlignmentSpec>
Настройки выравнивания source между разными партициями, в том числе разных Computation

watermark_percentile

Тип: NYT::TIntrusivePtr<NYT::NFlow::TWatermarkPercentileSpec>
Значение по умолчанию: {}
Настройка "персентильного" EventWatermark. Используется для игнорирования самых старых событий в пределах сконфигурированного процента.
EventTimestampAssigner

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

Параметр

Описание

column

Тип: std::optional<std::string>
Колонка со временем

format

Тип: NYT::NFlow::ETimestampFormat
Значение по умолчанию: seconds
Используемый формат времени (seconds и milliseconds)

limit_by_system_timestamp

Тип: bool
Значение по умолчанию: false
Ограничить EventTimestamp значением SystemTimestamp для защиты от битых значений из будущего
WatermarkGenerator

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

Параметр

Описание

use_source_watermark

Тип: bool
Значение по умолчанию: false
Использовать SourceWatermark для генерации EventWatermark. Эта опция полностью отключает все эвристики для определения EventWatermark и её нужно использовать строго в тех случаях, когда во входном потоке есть специальные маркеры с EventWatermark.

out_of_orderness_bound

Тип: TDuration
Значение по умолчанию: 0
Ограничение сверху на возможное переупорядочивание событий, используемое для оценки EventWatermark. По умолчанию переупорядочивание не учитывается.

idle_partitions

Тип: NYT::TIntrusivePtr<NYT::NFlow::TIdlePartitionsSpec>
Настройки эвристики для игнорирования партиций с нулевым потоком записи.

late_data_partitions

Тип: NYT::TIntrusivePtr<NYT::NFlow::TLateDataPartitionsSpec>

Настройки IdlePartitions:

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

Параметр

Описание

duration

Тип: TDuration
Значение по умолчанию: 1m
Время, как долго партиция должна быть пустой, чтобы поверить, что в неё не идёт записи.

max_ratio

Тип: double
Значение по умолчанию: 0.4
Доля партиций, которые могут быть исключены из расчёта вотермарка этой логикой. Идейный смысл опции — защита от продвижения вотермарка, когда поставщик данных прекратил писать сообщения в очередь по причине инцидента на его стороне. Выставление значения опции в 1.0 по сути отключает эту защиту.

Настройки UnavailablePartitionGroups:

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

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

WatermarkAlignment

Модуль используется для выравнивания входных потоков. Может приводить к полной остановке чтения в случае проблем с доступностью и/или оценкой Watermark у отдельных партиций или SourceComputation.

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

Параметр

Описание

group_name

Тип: NYT::TStrongTypedef<std::string, NYT::NFlow::TWatermarkAlignmentGroupTag, NYT::TStrongTypedefOptions{true}>
Значение по умолчанию: default-alignment-group
Имя группы

drift_bound

Тип: TDuration
Значение по умолчанию: 20m
Максимальное возможное опережение вотермарка отдельной партиции от вотермарка всей группы

read_delays

Тип: std::optional<THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, TDuration>>
Опция для продвинутого использования. Если EventTimestamp выходного сообщения больше, чем EventWatermark - Delay хотя бы для одного указанного потока, то чтение приостанавливается. При неправильной настройке опция может привести к полной остановке чтения.
WatermarkPercentile

Модуль используется для игнорирования старых данных в потоках при расчёте EventWatermark. Каждая партиция по каждому исходящему потоку поддерживает множество ещё необработанных сообщений. По умолчанию для расчета EventWatermark берется минимальное значение EventTimestamp среди всех сообщений (то есть учитывается 100% событий).

Если событие попадает в value персентиль - оно безусловно учитывается для расчёта EventWatermark. Если событие не попадает в указанный персентиль, однако отличается от EventTimestamp этих событий не более, чем на delay - то оно тоже будет учтено для расчёта EventWatermark. Таким образом, алгоритм расчёта сводится к EventWatermark = max(MinEventTimestamp, PercentileEventTimestamp - delay), где MinEventTimestamp - минимум по всем событиям, а PercentileEventTimestamp - значение EventTimestamp в соответствующем value персентиле.

Нужно учитывать, что inflight одного потока в рамках партиции может содержать в моменте десятки сообщений, поэтому значения вида 99, 99.9 могут работать неотличимо от 100 - то есть не приводя к желаемому эффекту. Кроме того, даже если старых событий от всего потока меньше 1%, эти старые события могут быть распределены неравномерно между партициями. Поэтому разумнее выставлять значения вида 80, 90 и т.п. Чтобы избежать негативных эффектов в виде неучёта свежих событий, существует параметр delay: благодаря ему будет теряться информация лишь о действительно старых событиях. Если таковых нет, то система будет работать так же, как и при value, равном 100, без потери какой-либо информации.

Данный модуль влияет только на расчёт EventWatermark и никак не влияет на SystemWatermark.

Важно

Настройки никак не влияют на агрегацию информации с разных партиций - будет взято в любом случае минимальное значение по всем партициям.

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

Параметр

Описание

value

Тип: NYT::TStrongTypedef<double, NYT::NFlow::TWatermarkPercentileTag, NYT::TStrongTypedefOptions{true}>
Значение по умолчанию: 100.0
Процент безусловно учитываемых для расчёта EventWatermark событий

delay

Тип: TDuration
Значение по умолчанию: 1m
Порог свежести для событий ниже выбранного персентиля
LateDataPartitions

Модуль используется для игнорирования партиций, содержащих исключительно старые данные. Работает аналогично WatermarkPercentile, но на уровне всех партиций, а не сообщений внутри партиции.

WatermarkPercentile позволяет игнорировать старые сообщения внутри партиции, но если вся партиция содержит только старые данные, она всё равно будет тормозить общий EventWatermark. Модуль LateDataPartitions решает эту проблему, позволяя игнорировать такие партиции целиком.

Алгоритм расчёта: FinalWatermark = max(MinPartitionWatermark, PercentilePartitionWatermark - Delay), где MinPartitionWatermark — минимальный watermark среди всех партиций, а PercentilePartitionWatermark — watermark партиции в указанном value перcентиле (партиции сортируются по watermark).

Если партиция попадает в value персентиль — она безусловно учитывается. Если партиция не попадает в указанный персентиль, однако её watermark отличается от персентильного не более, чем на delay — она тоже будет учтена. Партиции с watermark ниже порога (PercentileWatermark - Delay) считаются содержащими только late data, и их watermark «скрывается» (приводится к FinalWatermark).

Например, при value=90 и 100 партициях: 90% партиций (с наибольшими watermark) учитываются безусловно, а оставшиеся 10% (с наименьшими watermark) — только если их отставание от 90-го персентиля не превышает delay.

Важно

Этот модуль применяется после IdlePartitions. Idle-партиции уже исключены из расчёта до применения логики LateDataPartitions.

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

Параметр

Описание

value

Тип: NYT::TStrongTypedef<double, NYT::NFlow::TWatermarkPercentileTag, NYT::TStrongTypedefOptions{true}>
Значение по умолчанию: 100.0

delay

Тип: TDuration
Значение по умолчанию: 1m

Timer

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

Параметр

Описание

time_type

Тип: NYT::NFlow::ETimeType
Значение по умолчанию: event_time
Используемый таймером тип времени: event_time, system_time, real_time

streams

Тип: std::optional<THashSet<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>>>
Перечисление потоков, за которыми таймер следит.

streams_with_delays

Тип: std::optional<THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, TDuration>>
Перечисление потоков, за которыми таймер следит с указанием индивидуальной задержки

deduplicate_equal_timestamps

Тип: bool
Значение по умолчанию: true
Включение логики дедупликации таймеров: при попытке создания таймера с тем же ключом и trigger_timestamp, останется только один, с меньшим значением EventTimestamp

Если не указаны streams и streams_with_delays, то таймер следит за потоками согласно streams_dependency с нулевой дополнительной задержкой.

Source

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

Параметр

Описание

source_class_name

Тип: std::string
Значение по умолчанию: TString("")
Имя соответствующего класса.

Должен быть зарегистрирован с помощью YT_FLOW_DEFINE_SOURCE.

parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Значение по умолчанию: {}
Произвольные параметры для соответствующего класса.

Встроенные реализации сорсов описаны в документации по коннекторам.

Sink

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

Параметр

Описание

sink_class_name

Тип: std::string
Значение по умолчанию: TString("")
Имя соответствующего класса.

Должен быть зарегистрирован с помощью YT_FLOW_DEFINE_SINK.

input_stream_ids

Тип: THashSet<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>>
Значение по умолчанию: []
Список входных потоков.

Здесь могут быть только потоки, указанные в output_stream_ids всего Computation.

parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Значение по умолчанию: {}
Произвольные параметры для соответствующего класса.

Встроенные реализации синков описаны в документации по коннекторам.

HeavyHitters

Модуль обнаружения высокочастотных ключей.

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

Параметр

Описание

window

Тип: TDuration
Значение по умолчанию: 5m
Временное окно.

threshold

Тип: double
Значение по умолчанию: 0.01
Ключи, которые встречаются в потоке чаще, чем threshold, будут считаться высокочастотными. Алгоритм использует O(1 / threshold) памяти.

limit

Тип: long
Значение по умолчанию: 5
Максимальное количество показываемых высокочастотных ключей.

InputOrdering

Настройки порядка обработки событий из входных потоков. Первыми будут обработаны события с наименьшим временем.

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

Параметр

Описание

time_type

Тип: NYT::NFlow::ETimeType
Значение по умолчанию: event_time
Время для сортировки: event_time, system_time, real_time

stream_delays

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, TDuration>
Значение по умолчанию: {}
Задержки для входных потоков. Если поток не указан — то задержки не будет

Streams

В streams должны быть зарегистрированы только output потоки. Source потоки существуют только внутри соответствующих партиций и никак не персистятся в YT. А timer потоки имеют фиксированную схему.

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

Параметр

Описание

class_name

Тип: std::optional<std::string>

schema

Тип: NYT::TIntrusivePtr<NYT::NTableClient::TTableSchema>
Обязательный параметр
Cхема потока. Является схемой динтаблицы. Не может содержать сортировки и вычислимые поля.

Изменение схем рекомендуется осуществлять через полную остановку пайплайна: в этом случае внутри пайплайна не останется output сообщений.

migration_function

Тип: std::string
Значение по умолчанию: NYT::NFlow::DefaultMigrationFunction

Resource

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

Параметр

Описание

resource_class_name

Тип: std::string
Обязательный параметр
Имя соответствующего класса.

Должен быть зарегистрирован с помощью макроса YT_FLOW_DEFINE_RESOURCE.

parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Значение по умолчанию: {}
Произвольные параметры для соответствующего класса.

file_providers

Тип: THashMap<NYT::TStrongTypedef<std::string, NYT::NFlow::TFileProviderIdTag, NYT::TStrongTypedefOptions{true}>, NYT::TIntrusivePtr<NYT::NFlow::TFileProviderSpec>>
Значение по умолчанию: {}
Именованные файловые провайдеры ресурса. Контроллер добавляет файловый снимок в целевую ревизию только после обнаружения всех имён; собственная часть целевой спецификации контроллера может публиковаться независимо. Воркеры материализуют точные ревизии из полученной цели. Такие ресурсы поддерживаются только на воркерах и должны иметь controller = %false во всех путях required_resource_ids, которые могут до них дойти. Ключи должны соответствовать [0-9A-Za-z_-]+.

dependencies

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TResourceIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TResourceDescription>>
Значение по умолчанию: {}
Ресурсы, от которых зависит данный ресурс. Повторяет структуру required_resource_ids из Computation.

required_capabilities

Тип: THashMap<std::string, long>
Значение по умолчанию: {}

preload_required

Тип: bool
Значение по умолчанию: false

always_on

Тип: bool
Значение по умолчанию: false
Если true, ресурс загружается заранее при старте юнита и держится в памяти всё время жизни юнита (не выгружается), а не лениво при первом обращении. Загружается только на тех юнитах, где ресурс требует хотя бы одна computation (см. worker / controller в required_resource_ids): ресурс, нужный только на worker'е, не грузится на controller'е, а ресурс, который не требует ни одна computation, не грузится вовсе. Взаимоисключающе с preload_required.

DynamicSpec

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

Параметр

Описание

target_state

Тип: NYT::NFlow::EPipelineState
Значение по умолчанию: unknown
Целевое состояние системы. Для изменения используйте команды start-pipeline, stop-pipeline, pause-pipeline.

computations

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TComputationIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TDynamicComputationSpec>>
Значение по умолчанию: {}
Именованное перечисление всех узлов пайплайна.

resources

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TResourceIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TDynamicResourceSpec>>
Значение по умолчанию: {}

throttlers

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TThrottlerIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TDynamicThrottlerSpec>>
Значение по умолчанию: {}
Набор именованных распределённых троттлеров, общих для всех Computation пайплайна. Ключи являются id троттлеров и должны соответствовать [0-9A-Za-z_-]+. Подробнее в разделе Distributed Throttler.

job_manager

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicJobManagerSpec>
Значение по умолчанию: {}
Настройки JobManager.

message_distributor

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicMessageDistributorSpec>
Значение по умолчанию: {}
Настройки MessageDistributor.

job_tracker

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicJobTrackerSpec>
Значение по умолчанию: {}
Настройки JobTracker.

controller_connector

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicControllerConnectorSpec>
Значение по умолчанию: {}
Настройки ControllerConnector.

singletons

Тип: NYT::TIntrusivePtr<NYT::TSingletonsDynamicConfig>
Значение по умолчанию: {}

enable_mermaid_graph_describe

Тип: bool
Значение по умолчанию: false

flow_view_cache_codec

Тип: NYT::NCompression::ECodec
Значение по умолчанию: zstd_2

Stream и Resource в данный момент не имеют или не поддерживают динамические параметры.

Computation

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

Параметр

Описание

batch_duration

Тип: TDuration
Значение по умолчанию: 1s
Время максимального набора входного батча. Для очередей YTsaurus неполный батч добирается, пока с предыдущего батча не пройдёт это время; после неполного ответа очереди следующий запрос тоже выполняется не раньше этого момента.

max_rows_per_batch

Тип: NYT::NYTree::TSize
Значение по умолчанию: 1K
Максимальный размер входного батча в строках. Считается отдельно для input_streams, timer_streams и для каждого из source_streams. Для очередей YTsaurus считаются распакованные сообщения Flow; батч может превысить лимит на одну строку очереди.

max_bytes_per_batch

Тип: NYT::NYTree::TSize
Значение по умолчанию: 10Mi
Максимальный размер входного батча в байтах. Считается аналогично max_rows_per_batch. Для очередей YTsaurus считается размер распакованных сообщений; превышение — как для max_rows_per_batch.

max_keys_per_batch

Тип: std::optional<NYT::NYTree::TSize>

draining

Тип: bool
Значение по умолчанию: false
Режим Draining нужен для сброса пайплайна — в нём Computation перестанет запускать таймеры и доставать новые события из Source.

Команда stop-pipeline останавливает весь пайплайн через полный сброс. В отдельных Computation режим можно выставлять для дебага.

empty_batch_backoff

Тип: TDuration
Значение по умолчанию: 250ms
В случае пустой эпохи Computation дополнительно «спит» указанное время. Не рекомендуется выставлять параметр меньше 100 мс.

lease_check_period

Тип: TDuration
Значение по умолчанию: 1m

retryable_request

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicRetryableRequestSpec>
Значение по умолчанию: {}

tracer

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicPartitionTracerSpec>
Значение по умолчанию: {}

parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Значение по умолчанию: {}
Произвольные динамические параметры класса Computation.

processing_function_parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Динамические параметры process function. Тип параметров задаётся при регистрации YT_FLOW_DEFINE_PROCESS_FUNCTION. Для графа process function содержит динамические параметры узлов в nodes.

source_streams

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TDynamicSourceSpec>>
Значение по умолчанию: {}
Динамические параметры всех Sources.

key_visitor_streams

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TStreamIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TDynamicKeyVisitorStreamSpec>>
Значение по умолчанию: {}

sinks

Тип: THashMap<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TSinkIdTag>, NYT::TIntrusivePtr<NYT::NFlow::TDynamicSinkSpec>>
Значение по умолчанию: {}
Динамические параметры всех Sinks.

external_state_managers

Тип: THashMap<std::string, NYT::TIntrusivePtr<NYT::NFlow::TDynamicExternalStateManagerSpec>>
Значение по умолчанию: {}

external_state_joiners

Тип: THashMap<std::string, NYT::TIntrusivePtr<NYT::NFlow::TDynamicExternalStateJoinerSpec>>
Значение по умолчанию: {}

state_joiners

Тип: THashMap<std::string, NYT::TIntrusivePtr<NYT::NFlow::TDynamicStateJoinerSpec>>
Значение по умолчанию: {}

state_manager

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicStateManagerSpec>
Значение по умолчанию: {}

input_store

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicInputStoreSpec>
Значение по умолчанию: {}

timer_store

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicTimerStoreSpec>
Значение по умолчанию: {}

output_store

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicOutputStoreSpec>
Значение по умолчанию: {}

timer_store_count_limit

Тип: NYT::NYTree::TSize
Значение по умолчанию: 250K

timer_store_byte_size_limit

Тип: NYT::NYTree::TSize
Значение по умолчанию: 100Mi

blocked_time_window

Тип: TDuration
Значение по умолчанию: 10m
Окно усреднения для доли времени, которую джоба провела заблокированной на каждом из лимитов (blocked_share в диагностиках). Доля нормируется на время жизни джобы, поэтому джоба, заблокированная с самого старта, показывает ~1 независимо от своего возраста.

input_rows_throttler_id

Тип: std::optional<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TThrottlerIdTag>>
Id троттлера для ограничения скорости обработки сообщений. Если задан, перед каждой итерацией Computation ждёт квоту, равную числу сообщений во входном батче. Id должен присутствовать в dynamic_spec/throttlers. Подробнее в разделе Distributed Throttler.

input_bytes_throttler_id

Тип: std::optional<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TThrottlerIdTag>>
Аналогично input_rows_throttler_id, но квотируется суммарный byte_size сообщений батча — системный размер их сериализованного представления.

input_rows_throttler_class_id

Тип: std::optional<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TQuotaClassIdTag>>
Класс квоты для троттлера из input_rows_throttler_id. Класс должен быть объявлен в этом троттлере; default выбирает зарезервированный класс с весом 1.0. Требует заданного input_rows_throttler_id.

input_bytes_throttler_class_id

Тип: std::optional<NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TQuotaClassIdTag>>
Класс квоты для троттлера из input_bytes_throttler_id. Класс должен быть объявлен в этом троттлере; default выбирает зарезервированный класс с весом 1.0. Требует заданного input_bytes_throttler_id.

skip_if_expression

Тип: std::optional<std::string>
YTQL-предикат, по которому входные сообщения отфильтровываются (пропускаются) ещё до обработки Computation. Сообщение пропускается (дропается), если предикат вычисляется в булево true.

Предикат вычисляется над колонками payload сообщения плюс мета-колонками $message_id, $stream_id, $system_timestamp, $event_timestamp, $alignment_timestamp. Результат обязан быть булевым: NULL или небулевой результат приводит к исключению.

Применяется как к input-, так и к source-потокам. Число пропущенных сообщений экспортируется метриками input_streams/skipped_by_expression_count и source_streams/skipped_by_expression_count.

Source

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

Параметр

Описание

parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Значение по умолчанию: {}
Произвольные динамические параметры соответствующего класса Source.

draining

Тип: bool
Значение по умолчанию: false

Sink

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

Параметр

Описание

parameters

Тип: NYT::TIntrusivePtr<NYT::NYTree::IMapNode>
Значение по умолчанию: {}
Произвольные динамические параметры соответствующего класса Sink.

JobManager

Конфиг модуля управления джобами на контроллере.

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

Параметр

Описание

balancer_type

Тип: NYT::NFlow::EJobBalancerType
Значение по умолчанию: cpu_aware

use_cpu_aware_balancer

Тип: std::optional<bool>
Включает экспериментальную балансировку CPU. Балансировка по умолчанию может приводить к сильному перекосу ресурсов CPU между воркерами.

rebalance_delay_after_pipeline_sync

Тип: TDuration
Значение по умолчанию: 30s

rebalance_target_deviation

Тип: double
Значение по умолчанию: 0.1
Максимальное допустимое отклонение.

rebalance_hot_mode_coeff

Тип: double
Значение по умолчанию: 2.0

rebalance_action_min_time

Тип: TDuration
Значение по умолчанию: 4s

rebalance_action_max_time

Тип: TDuration
Значение по умолчанию: 12s

rebalance_sync_period

Тип: TDuration
Значение по умолчанию: 10s
Период cpu_aware балансировки. Не рекомендуется выставлять значение меньше 15min, так как балансировка смотрит на метрики 10min.

rebalance_count_exceeded_allowed

Тип: double
Значение по умолчанию: 1.2

rebalance_even_load_thresholds

Тип: THashMap<NYT::NFlow::EBalanceResource, NYT::TIntrusivePtr<NYT::NFlow::TEvenLoadThresholds>>
Значение по умолчанию: {}
Пороги гейта «нагрузка распределена ровно» по ресурсам, например {cpu = {spread = 2.0; ratio = 1.5}}. Учитываются только ресурсы с ненулевым весом. Незаданные поля берут значения по умолчанию: разброс 1.0 ядро для CPU и 1 GB для памяти, отношение 1.2.

rebalance_min_cpu_spread

Тип: std::optional<double>
Устаревший параметр: то же, что spread в записи cpu параметра rebalance_even_load_thresholds. Явно заданное значение в rebalance_even_load_thresholds имеет приоритет.

rebalance_min_cpu_ratio

Тип: std::optional<double>
Устаревший параметр: то же, что ratio в записи cpu параметра rebalance_even_load_thresholds. Явно заданное значение в rebalance_even_load_thresholds имеет приоритет.

balance_weights

Тип: THashMap<NYT::NFlow::EBalanceResource, double>
Значение по умолчанию: {'cpu': 1.0, 'memory': 0.0}
Относительная важность ресурсов для cpu_aware балансировки, например {cpu = 80; memory = 20}. Указанные ключи накладываются на значение по умолчанию {cpu = 1; memory = 0}; веса нормируются, значимы только пропорции. Ресурс с нулевым весом не участвует в балансировке, поэтому по умолчанию балансируется только CPU. Порог rebalance_target_deviation применяется к нормированной взвешенной сумме: добавление второго ресурса пропорционально уменьшает вклад первого. При ненулевом весе memory память партиции включает измеренную память её джобы и байты входных буферов её потоков. Каждый входной поток учитывается по большему из его demand, то есть запроса до урезания пулом буферов, и фактически занятых байтов. Поток без demand, например с явно переопределённым лимитом или на воркере без buffer manager v2, учитывается по фактически занятым байтам.

balancer_metrics_source

Тип: NYT::NFlow::EBalancerMetricsSource
Значение по умолчанию: job
Откуда cpu_aware балансировщик берёт загрузку CPU партиции. partition — 10-минутная метрика текущей джобы, а пока её нет — сохранённая история партиции (замер предыдущей джобы, переживает переезд и рестарт); 30-секундные и мгновенные метрики не используются. job — прежнее поведение: только метрики текущей джобы с откатом на 30-секундную и мгновенную. По умолчанию выключено, пока изменение раскатывается по пайплайнам; временный переключатель.

balance_warmup_protection

Тип: bool
Значение по умолчанию: false
Не переезжать работающие джобы, пока их метрики не созрели: есть 10-минутная метрика и прошло ещё одно 10-минутное окно после её появления. Такая джоба ещё расплачивается за предыдущий переезд, а её вес неизвестен. Защита действует для всех ресурсов балансировки: конфигурация только по памяти тоже ждёт созревания CPU-метрики. Перегруженный по количеству воркер, у которого подходящая для кика партиция не созрела, в этом раунде пропускается целиком. Джобы моложе двух минут остаются подвижными: они ещё ничего не вложили, а на рестарте пайплайна воркеры регистрируются не одновременно. Партиции без джобы размещаются как обычно. По умолчанию выключено, пока изменение раскатывается по пайплайнам; временный переключатель.

balance_warmup_idle_worker_share

Тип: double
Значение по умолчанию: 0.2
Доля воркеров группы без джоб этой группы, начиная с которой warm-up protection снимается: заполнить простаивающие воркеры важнее, чем метрики переезжающих джоб, например когда после рестарта пайплайна воркеры регистрируются с разницей в минуты. Временный переключатель для отката.

worker_coef_mode

Тип: NYT::NFlow::EWorkerCoefMode
Значение по умолчанию: legacy
Как cpu_aware балансировщик оценивает относительную скорость воркеров. probing — по партициям, переехавшим между воркерами: сравнивается CPU на сообщение до и после переезда, наблюдения по парам воркеров накапливаются в balancer_state и решаются совместно; без переездов все коэффициенты равны 1. legacy — прежняя оценка по текущей загрузке партиций каждого воркера относительно средних по компьютейшнам. По умолчанию выключено, пока изменение раскатывается по пайплайнам; временный переключатель.

worker_coef_half_life

Тип: TDuration
Значение по умолчанию: 1d
Период полураспада накопленного наблюдения пары воркеров: при слиянии нового наблюдения старое теряет вес пропорционально прошедшему времени. Без новых наблюдений оценка не меняется.

worker_coef_retention

Тип: TDuration
Значение по умолчанию: 7d
Сколько помнить наблюдения воркера, отсутствующего в группе; вернувшийся раньше воркер получает свой коэффициент обратно.

worker_coef_prior_weight

Тип: double
Значение по умолчанию: 0.05
Вес априорного «коэффициент воркера равен 1» в единицах веса наблюдений (переезд одной партиции весит 1 / N, где N — число партиций на воркере-приёмнике). Фиксирует среднее коэффициентов группы на 1 и гасит одиночные наблюдения.

worker_coef_max_ratio

Тип: double
Значение по умолчанию: 4.0
Предохранитель: коэффициенты воркеров ограничены отрезком [1 / max_ratio, max_ratio].

disable_even_load_gate

Тип: std::optional<bool>

async_balancing

Тип: bool
Значение по умолчанию: true

graceful_move

Тип: bool
Значение по умолчанию: true

zero_queue_latency

Тип: TDuration
Значение по умолчанию: 1s

planning_horizon

Тип: TDuration
Значение по умолчанию: 10m

preloading_timeout

Тип: TDuration
Значение по умолчанию: 30m
Сколько балансер resource_queue держит предзагрузку модели, которая ещё грузится, считая от запроса. До этого предзагрузка не снимается, а воркер считается воркером вычислений, которым нужна модель; после — обычные правила.

minimum_worker_count

Тип: unsigned long
Значение по умолчанию: 1
Минимальное число воркеров, на которые стоит распределять нагрузку.

Должно быть обязательно заполнено, чтобы избежать ошибок Out of memory и перегрузки отдельных воркеров при старте системы. Рекомендуемое значение — около 60% от целевого количества воркеров (чтобы пережить потерю части воркеров, например при отключении одного дата-центра).

lost_job_timeout

Тип: TDuration
Значение по умолчанию: 10s
Через какое время признавать Job потерянным, если в контроллер не приходит обновление статуса джоба. То же время используется, чтобы признавать потерянным воркера.

faulty_address_window

Тип: TDuration
Значение по умолчанию: 5m
Окно на котором считается экспоненциально затухающий счетчик числа потерь соединения контроллера с воркером. Если этот счетчик превышает faulty_address_attempts, то воркер считается faulty и он игнорируется при балансировке. Решает проблему воркеров, у которых постоянно теряется коннект с контроллером. В том числе проблему двух воркеров с разными incarnation id и одним address (когда в деплое оказывается два пода считающих, что у них один и тот же FQDN).

faulty_address_attempts

Тип: unsigned long
Значение по умолчанию: 5
Смотрите описание параметра faulty_address_window.

partitioning

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicPartitioningSpec>
Значение по умолчанию: {}
Настройки партиционирования пайплайна.

worker_group_override

Тип: THashMap<NYT::TStrongTypedef<std::string, NYT::NFlow::TWorkerGroupIdTag, NYT::TStrongTypedefOptions{true}>, NYT::TIntrusivePtr<NYT::NFlow::TDynamicJobManagerGroupSpec>>
Значение по умолчанию: {}

partition_history_limit

Тип: long
Значение по умолчанию: 4096
Сколько историй партиций балансер хранит между перезапусками джоб, см. balancer_metrics_source. Истории лежат в одном персистентном документе с ограничением на размер, поэтому при достижении предела самая лёгкая история уступает место более тяжёлой, а более лёгкие не сохраняются. При остановке пайплайна истории не сохраняются вовсе.

JobTracker

Конфиг модуля управления джобами на воркере.

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

Параметр

Описание

job_control_threads

Тип: int
Значение по умолчанию: 3
Количество управляющих тредов.

В большинстве случаев достаточно значения по умолчанию.

job_threads

Тип: std::optional<int>
Количество тредов для выполнения джобов.

Если не задано, размер пула вычисляется автоматически на основе CPU-лимита ноды.
Если при этом вычислить лимит ноды не получилось, то будет создано 30 тредов.

buffer_state_manager

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicBufferStateManagerSpec>
Значение по умолчанию: {}
Конфиг модуля управления буферами для входящих и исходящих сообщений.

load_throughput_throttler

Тип: NYT::TIntrusivePtr<NYT::NFlow::TLoadThroughputThrottlerSpec>
Значение по умолчанию: {'limit': 134217728.0}

state_cache

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicStateCacheSpec>
Значение по умолчанию: {}

mark_performance_metrics_steady_after_first_iteration

Тип: bool
Значение по умолчанию: false
Сообщать счётчики скоростей джобы (CPU, входные сообщения, память) как установившиеся только после первой завершённой итерации с входными данными, чтобы балансировщик читал оконные метрики cpu_usage_10m и подобные после того, как инициализация джобы (выкачка стейта, построение индексов) вымоется из окон; сами счётчики не сбрасываются. Джоба, не получавшая входных данных минуту, тоже считается установившейся: простаивающая джоба сообщает свои околонулевые скорости, а не остаётся неизмеренной. По умолчанию выключено, пока изменение раскатывается по пайплайнам; временный переключатель.

BufferStateManager

Конфиг модуля управления размерами буферов.

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

Параметр

Описание

manage_period

Тип: TDuration
Значение по умолчанию: 1s
Частота пересчёта размера буфферов.

demand_window

Тип: TDuration
Значение по умолчанию: 1m
Временное окно, на котором оценивается утилизация одного буффера.

input_buffer

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicBufferStateManagerSpec::TOneSideBufferSpec>
Значение по умолчанию: {}
Настройки входного буффера.

output_buffer

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicBufferStateManagerSpec::TOneSideBufferSpec>
Значение по умолчанию: {}
Настройки выходного буффера.

enable_v2

Тип: bool
Значение по умолчанию: true
Включает v2-стратегию сайзинга буфферов: лимит = пик используемого + headroom, с полом v2_gain_epochs × demand × epoch (bandwidth-delay product). Потолок demand × max_duration поднимается до анонсированного input-backlog'а или, для производящего output, до равной доли половины fair_share_pool, поскольку эпохи продюсера не показывают период подтверждений downstream. Лимиты, подтверждённые спросом, распределяются раньше спекулятивных output-probe. Суммарно Σ(лимитов) ≤ fair_share_pool с учётом чужого in-flight; job_limit остаётся границей одного стрима. Стримы с job_overrides живут целиком вне пула (как в v1): память воркера под буфферы стоит закладывать как fair_share_pool + сумма фактического in-flight оверрайднутых стримов. По умолчанию включено; установите false, чтобы использовать прежнюю v1-формулу.
Дополнительные параметры

epoch_cycle_window_samples

Тип: int
Значение по умолчанию: 16
Число сэмплов в окне медианы длительности эпохи джобы.

max_rate_estimator_buckets

Тип: int
Значение по умолчанию: 8
Число бакетов в windowed-max оценщике скорости дрейна; изменение сбрасывает накопленную оценку.

warmup_refresh_period

Тип: TDuration
Значение по умолчанию: 30s
Как часто джоба опрашивает сошедшийся warmup для персиста.

v2_gain_epochs

Тип: double
Значение по умолчанию: 2.0
Целевой пол буфера в единицах эпох: сколько эпох спроса держать как минимум (запас на всплески скорости и latency). Действует только при enable_v2.

v2_use_offered_rate

Тип: bool
Значение по умолчанию: true
Учитывать ли в оценке спроса скорость по анонсированному backlog'у (offered rate). Можно выключить, если продюсер завышает анонсированный backlog. Действует только при enable_v2.

v2_floor

Тип: NYT::NYTree::TSize
Значение по умолчанию: 2Mi
Минимальный грант стриму с непустым backlog'ом; должен вмещать самое большое сообщение. Действует только при enable_v2.

v2_headroom_growth_factor

Тип: double
Значение по умолчанию: 2.0
Во сколько раз растёт headroom за один тик менеджера, пока утилизация высокая. Действует только при enable_v2.

v2_high_utilization_threshold

Тип: double
Значение по умолчанию: 0.65
Порог утилизации, выше которого headroom растёт, а ниже половины — затухает. Действует только при enable_v2.

v2_publish_threshold

Тип: double
Значение по умолчанию: 0.25
Не публиковать новый лимит, если он отличается от текущего меньше чем на эту долю (антиосцилляция); сжатия публикуются всегда. Действует только при enable_v2.

Все размерные (измеряемые в количестве элементов или в байтах) параметры в этом конфиге могут парситься из строк вида "50K".

ControllerConnector

Конфиг управления соединением воркера и контроллера.

Не рекомендуется менять без необходимости.

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

Параметр

Описание

controller_wait_timeout

Тип: TDuration
Значение по умолчанию: 5m
Время, через которое воркер принудительно отключит все джобы в случае потери связи с контроллером.

controller_discover_period

Тип: TDuration
Значение по умолчанию: 10s
Период повторного поиска контроллера.

controller_heartbeat_period

Тип: TDuration
Значение по умолчанию: 1s
Период хартбитов воркера в контроллер.

worker_statistics_report_period

Тип: TDuration
Значение по умолчанию: 30s

controller_heartbeat_rpc_timeout

Тип: TDuration
Значение по умолчанию: 10s
Таймаут хартбита.

controller_heartbeat_failure_backoff

Тип: TDuration
Значение по умолчанию: 1s
Время выдержки на случай ошибки рукопожатия.

controller_handshake_rpc_timeout

Тип: TDuration
Значение по умолчанию: 10s
Таймаут рукопожатия.

controller_handshake_failure_backoff

Тип: TDuration
Значение по умолчанию: 1s

orchid_update_period

Тип: TDuration
Значение по умолчанию: 1s
Период обновления orchid на контроллере.

Parameters

Поле parameters есть в спеках Computation, Source, Sink и Resource. Оно содержит параметры выбранной встроенной реализации, например адаптера process function. Проверяйте доступные поля в документации конкретного класса и не используйте этот блок для параметров пользовательской логики.

Параметры C++ process function задаются отдельно в processing_function_parameters. Объявите для них обычный yson struct, зарегистрируйте тип вместе с функцией и читайте параметры через runtime-контекст:

struct TMyParameters
    : public NYTree::TYsonStruct
{
    i64 Threshold;

    REGISTER_YSON_STRUCT(TMyParameters);

    static void Register(TRegistrar registrar);
};

YT_FLOW_DEFINE_PROCESS_FUNCTION(TMyFunction, TMyParameters);

"processing_function_parameters" = {
    "threshold" = 10;
};

Статические параметры доступны в Init через initContext->GetParameters<TMyParameters>(), динамические — через context->GetDynamicParameters<TMyParameters>(). Подробнее см. в разделе Process function.

Config

Конфиг ноды кластера Flow.

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

Параметр

Описание

logging

Тип: NYT::TIntrusivePtr<NYT::NLogging::TLogManagerConfig>
Значение по умолчанию:

{
    "high_backlog_watermark" = 100000;
    "low_backlog_watermark" = 100000;
    "min_disk_space" = 0;
    "rules" = [
        {
            "exclude_categories" = [];
            "max_level" = "maximum";
            "min_level" = "info";
            "writers" = [
                "Stderr";
            ];
        };
    ];
    "writers" = {
        "Stderr" = {
            "common_fields" = {};
            "enable_host_field" = %false;
            "enable_native_tags" = %false;
            "enable_source_location" = %false;
            "enable_span_id_field" = %false;
            "enable_system_fields" = %true;
            "format" = "plain_text";
            "type" = "stderr";
            "yson_format" = "text";
        };
    };
}

Настройки логирования.

jaeger

Тип: NYT::TIntrusivePtr<NYT::NTracing::TJaegerTracerConfig>
Значение по умолчанию: {}

pipe_io_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NPipeIO::TPipeIODispatcherConfig>
Значение по умолчанию: {}

solomon_registry

Тип: NYT::TIntrusivePtr<NYT::NProfiling::TSolomonRegistryConfig>
Значение по умолчанию: {}

error_backtrace_enricher

Тип: NYT::TIntrusivePtr<NYT::NFlow::TBacktraceEnricherSpec>
Значение по умолчанию: {}

query_engine_config

Тип: NYT::TIntrusivePtr<NYT::NQueryClient::TQueryEngineConfig>
Значение по умолчанию: {}

cluster_url

Тип: std::string
Обязательный параметр
Кластер YTsaurus для работы.

path

Тип: TString
Обязательный параметр
Рабочая папка пайплайна на кластере cluster_url.

proxy_role

Тип: std::optional<std::string>
Роль rpc proxy.

rpc_port

Тип: int
Значение по умолчанию: 0
Основной порт для общения воркеров и контроллеров, может быть отдельным у каждого инстанса.

monitoring_port

Тип: int
Значение по умолчанию: 0
API для получения метрик.

companion

Тип: NYT::TIntrusivePtr<NYT::NFlow::NCompanion::TCompanionConfig>
Параметры процесса-компаньона. Нужны любому воркеру, который запускает компаньон (Python, Java, Go, C++-компаньон). В vanilla-запуске заполняются автоматически.

controller

Тип: NYT::TIntrusivePtr<NYT::NFlow::NController::TControllerConfig>
Значение по умолчанию: {}
Параметры Controller.

worker

Тип: NYT::TIntrusivePtr<NYT::NFlow::NWorker::TWorkerConfig>
Значение по умолчанию: {}
Параметры Worker.

authenticator

Тип: NYT::TIntrusivePtr<NYT::NFlow::TAuthenticatorConfig>
Значение по умолчанию: {}

bus_server

Тип: NYT::TIntrusivePtr<NYT::NBus::NTcp::TBusServerConfig>
Значение по умолчанию: {}

rpc_server

Тип: NYT::TIntrusivePtr<NYT::NRpc::TServerConfig>
Значение по умолчанию: {}

core_dumper

Тип: NYT::TIntrusivePtr<NYT::NCoreDump::TCoreDumperConfig>

solomon_exporter

Тип: NYT::TIntrusivePtr<NYT::NProfiling::TSolomonExporterConfig>
Значение по умолчанию: {'enable_solomon_aggregates': true}

solomon_proxy

Тип: NYT::TIntrusivePtr<NYT::NProfiling::TSolomonProxyConfig>
Значение по умолчанию: {}

http_client_config

Тип: NYT::TIntrusivePtr<NYT::NHttp::TClientConfig>
Значение по умолчанию: {}

https_client_config

Тип: NYT::TIntrusivePtr<NYT::NHttps::TClientConfig>
Значение по умолчанию: {}

http_poller_threads

Тип: int
Значение по умолчанию: 1

abort_on_unrecognized_options

Тип: bool
Значение по умолчанию: true
При выставлении значения true Flow Node не будет запускаться при наличии неизвестных опций в конфиге.

ignore_singletons_dynamic_config

Тип: bool
Значение по умолчанию: false

enable_porto_resource_tracker

Тип: bool
Значение по умолчанию: true
Дополнительные параметры

address_resolver

Тип: NYT::TIntrusivePtr<NYT::NNet::TAddressResolverConfig>
Значение по умолчанию: {}

grpc_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NRpc::NGrpc::TDispatcherConfig>
Значение по умолчанию: {}

tcmalloc

Тип: NYT::TIntrusivePtr<NYT::NTCMalloc::TTCMallocConfig>
Значение по умолчанию: {'guarded_sampling_rate': 16777216}

fiber_manager

Тип: NYT::TIntrusivePtr<NYT::NConcurrency::TFiberManagerConfig>
Значение по умолчанию: {}

resource_tracker

Тип: NYT::TIntrusivePtr<NYT::NProfiling::TResourceTrackerConfig>
Значение по умолчанию: {}

tcp_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NBus::NTcp::TDispatcherConfig>
Значение по умолчанию: {}

rpc_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NRpc::TDispatcherConfig>
Значение по умолчанию: {}

protobuf_interop

Тип: NYT::TIntrusivePtr<NYT::NYson::TProtobufInteropConfig>
Значение по умолчанию: {}

clients_cache

Тип: NYT::TIntrusivePtr<NYT::NClient::NCache::TClientsCacheConfig>
Значение по умолчанию: {}
Настройки соединений кэша YT-клиентов ноды. Через него ходят коннектор пайплайна, writer логов в очередь и всё, что берёт клиента из TComputationContext::ClientsCache — компьютейшены, коннекторы и менеджеры состояния. Аутентификатор пайплайна создаёт своего клиента отдельно и этими настройками не управляется.

clients_cache_factory

Тип: NYT::TIntrusivePtr<NYT::NYTree::INode>
Параметры фабрики корневого кэша YT-клиентов, установленной через SetRootClientsCacheFactory(). Схему параметров задаёт сама фабрика; встроенная фабрика этот блок игнорирует.

enable_phdr_cache

Тип: bool
Значение по умолчанию: true

Controller

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

Параметр

Описание

controller_threads

Тип: int
Значение по умолчанию: 5
Число тредов.

orchid_update_period

Тип: TDuration
Значение по умолчанию: 1s
Период пересчёта orchid.

warm_up_time

Тип: TDuration
Значение по умолчанию: 5s
Период разогрева при старте контроллера.

scheduler_period

Тип: TDuration
Значение по умолчанию: 5s
Период запуска планировщика.

cache_period

Тип: TDuration
Значение по умолчанию: 1s

feedback_period

Тип: TDuration
Значение по умолчанию: 1s

metrics_period

Тип: TDuration
Значение по умолчанию: 5s

write_own_retryable_errors_period

Тип: TDuration
Значение по умолчанию: 5s

publish_retry_period

Тип: TDuration
Значение по умолчанию: 5s
Период, в течение которого контроллер пытается передать в YTsaurus информацию, что является лидером.

publish_timeout

Тип: TDuration
Значение по умолчанию: 2h

election_manager

Тип: NYT::NYTree::TPolymorphicYsonStruct<NYT::NYTree::NDetail::TPolymorphicMapping<&NYT::NFlow::NController::ElectionBackendDiscriminator.<char const at offset 0>, NYT::NFlow::NController::EElectionBackend, NYT::NYTree::NDetail::TOptionalValue<NYT::NFlow::NController::EElectionBackend, (NYT::NFlow::NController::EElectionBackend)0>, NYT::NFlow::NController::TElectionBackendConfigBase, NYT::NYTree::NDetail::TLeafTag<(NYT::NFlow::NController::EElectionBackend)0, NYT::NFlow::NController::TCypressElectionBackendConfig>, NYT::NYTree::NDetail::TLeafTag<(NYT::NFlow::NController::EElectionBackend)1, NYT::NFlow::NController::TDyntableElectionBackendConfig>, NYT::NYTree::NDetail::TLeafTag<(NYT::NFlow::NController::EElectionBackend)2, NYT::NFlow::NController::TChaosElectionBackendConfig>>>
Значение по умолчанию:

{
    "backend" = "cypress";
    "leader_cache_update_period" = 1000;
    "leader_lease_ping_period" = 1000;
    "leader_lease_ttl" = 5000;
    "lock_acquisition_period" = 1000;
}

Настройки выбора лидера среди нескольких контроллеров.

persisted_state_manager

Тип: NYT::TIntrusivePtr<NYT::NFlow::NController::TPersistedStateManagerConfig>
Значение по умолчанию: {}
Настройки PersistedStateManager.

lease_manager

Тип: NYT::TIntrusivePtr<NYT::NFlow::NController::TLeaseManagerConfig>
Значение по умолчанию: {}
Настройки LeaseManager.

controller_service

Тип: NYT::TIntrusivePtr<NYT::NFlow::NController::TControllerServiceConfig>
Значение по умолчанию: {}
Настройки ControllerService.

bus

Тип: NYT::TIntrusivePtr<NYT::NBus::NTcp::TBusConfig>
Значение по умолчанию: {}
Дополнительные параметры

publish_request_timeout

Тип: TDuration
Значение по умолчанию: 1s
Таймаут одного запроса попытки публикации. Неудачная попытка повторяется, так что этот параметр определяет только то, как долго одна попытка может висеть до повтора.

PersistedStateManager

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

Параметр

Описание

timeout

Тип: TDuration
Значение по умолчанию: 5s
Таймаут сохранения в YTsaurus.

max_reads_per_transaction

Тип: long
Значение по умолчанию: 10000

max_writes_per_transaction

Тип: long
Значение по умолчанию: 10000

LeaseManager

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

Параметр

Описание

lease_timeout

Тип: TDuration
Значение по умолчанию: 10m
Сколько джоба остаётся отгороженной без обновления фенса контроллером. С кипарисным бэкендом выборов это таймаут мастер-транзакции лиза джобы, с хаосным — таймаут его хаос-лиза, с дин-табличным — время жизни общей для пайплайна строки дедлайна, через которую проходит каждый коммит воркера.

lease_ping_period

Тип: TDuration
Значение по умолчанию: 30s
Период, с которым лидер продлевает лизы. Продлевает всегда лидер, а не воркер, но механика у бэкендов разная: хаосные лизы пингует один периодик, кипарисные — по клиентскому пингеру на транзакцию лиза, а общий дин-табличный дедлайн просто переписывается.

max_concurrent_requests

Тип: long
Значение по умолчанию: 500
Максимальное количество одновременных запросов, которые менеджер держит в полёте при прицеплении лизов и их прекращении. Не увеличивайте, чтобы избежать перегрузки мастера. На продление не распространяется: хаосные пинги рассылаются одним раундом без ожидания, а кипарисные принадлежат самим транзакциям лизов.

ControllerService

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

Параметр

Описание

set_spec_retry_count

Тип: int
Значение по умолчанию: 3
Количество попыток обновления спеки в случае ошибок.

set_spec_retry_period

Тип: TDuration
Значение по умолчанию: 5s
Время между попытками.

tables_throttler

Тип: NYT::TIntrusivePtr<NYT::NFlow::TLoadThroughputThrottlerSpec>
Значение по умолчанию: {}

Плюс небольшой запас для кейса, что в некоторые живые партиции тоже не будет записи.

Runner config

Конфиг для запуска пайплайна. Формат этого конфига не регламентирован, пользователь может использовать такой конфиг, какой удобнее для его задач.

Flow предоставляет простой вариант этого конфига NYT::NFlow::TSimpleRunnerConfig и программу NYT::NFlow::TSimpleRunnerProgram для запуска пайплайна с использованием этого конфига. И именно на такой формат конфига рассчитывает тестовый фреймворк Flow, чтобы подставить в него кластер и путь к пайплайну в локальном YTsaurus и чтобы применить тестовые дефолтные параметры для более быстрой работы теста.

SimpleRunnerConfig

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

Параметр

Описание

logging

Тип: NYT::TIntrusivePtr<NYT::NLogging::TLogManagerConfig>
Значение по умолчанию:

{
    "high_backlog_watermark" = 100000;
    "low_backlog_watermark" = 100000;
    "min_disk_space" = 0;
    "rules" = [
        {
            "exclude_categories" = [];
            "max_level" = "maximum";
            "min_level" = "info";
            "writers" = [
                "Stderr";
            ];
        };
    ];
    "writers" = {
        "Stderr" = {
            "common_fields" = {};
            "enable_host_field" = %false;
            "enable_native_tags" = %false;
            "enable_source_location" = %false;
            "enable_span_id_field" = %false;
            "enable_system_fields" = %true;
            "format" = "plain_text";
            "type" = "stderr";
            "yson_format" = "text";
        };
    };
}

Настройки логирования.

jaeger

Тип: NYT::TIntrusivePtr<NYT::NTracing::TJaegerTracerConfig>
Значение по умолчанию: {}

pipe_io_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NPipeIO::TPipeIODispatcherConfig>
Значение по умолчанию: {}

solomon_registry

Тип: NYT::TIntrusivePtr<NYT::NProfiling::TSolomonRegistryConfig>
Значение по умолчанию: {}

error_backtrace_enricher

Тип: NYT::TIntrusivePtr<NYT::NFlow::TBacktraceEnricherSpec>
Значение по умолчанию: {}

query_engine_config

Тип: NYT::TIntrusivePtr<NYT::NQueryClient::TQueryEngineConfig>
Значение по умолчанию: {}

cluster_url

Тип: std::string
Обязательный параметр
Имя кластера, на котором находится пайплайн. Хвост после последнего / трактуется как прокси-роль и имеет приоритет над proxy_role — например, hahn/flow означает кластер hahn и роль flow.

proxy_role

Тип: std::optional<std::string>
Обязательный параметр
Прокси-роль, которую нужно использовать для выполнения операций над пайплайном.

path

Тип: TString
Обязательный параметр
Путь к пайплайну на YTsaurus.

spec

Тип: NYT::TIntrusivePtr<NYT::NFlow::TPipelineSpec>
Обязательный параметр
Статическая спека, с которой нужно запустить пайплайн.

dynamic_spec

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDynamicPipelineSpec>
Значение по умолчанию: {}
Динамическая спека, с которой нужно запустить пайплайн.

abort_on_unrecognized_options

Тип: bool
Значение по умолчанию: false

abort_on_specs_parseability_error

Тип: bool
Значение по умолчанию: false

vanilla

Тип: NYT::TIntrusivePtr<NYT::NFlow::TVanillaConfig>
Если задан и enable=%true, раннер вместо запуска контроллера/воркеров локально поднимет vanilla-операцию в YTsaurus и запустит федерацию в ней.

set_flow_core_target

Тип: bool
Значение по умолчанию: true

direct_controller_commands

Тип: NYT::TIntrusivePtr<NYT::NFlow::TDirectControllerCommandsConfig>
Значение по умолчанию: {}
Дополнительные параметры

address_resolver

Тип: NYT::TIntrusivePtr<NYT::NNet::TAddressResolverConfig>
Значение по умолчанию: {}

grpc_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NRpc::NGrpc::TDispatcherConfig>
Значение по умолчанию: {}

tcmalloc

Тип: NYT::TIntrusivePtr<NYT::NTCMalloc::TTCMallocConfig>
Значение по умолчанию: {}

fiber_manager

Тип: NYT::TIntrusivePtr<NYT::NConcurrency::TFiberManagerConfig>
Значение по умолчанию: {}

resource_tracker

Тип: NYT::TIntrusivePtr<NYT::NProfiling::TResourceTrackerConfig>
Значение по умолчанию: {}

tcp_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NBus::NTcp::TDispatcherConfig>
Значение по умолчанию: {}

rpc_dispatcher

Тип: NYT::TIntrusivePtr<NYT::NRpc::TDispatcherConfig>
Значение по умолчанию: {}

protobuf_interop

Тип: NYT::TIntrusivePtr<NYT::NYson::TProtobufInteropConfig>
Значение по умолчанию: {}

clients_cache

Тип: NYT::TIntrusivePtr<NYT::NClient::NCache::TClientsCacheConfig>
Значение по умолчанию: {}
Настройки соединений кэша YT-клиентов, через который раннер общается с кластером пайплайна. На vanilla launcher не распространяются: он создаёт своих клиентов отдельно.

clients_cache_factory

Тип: NYT::TIntrusivePtr<NYT::NYTree::INode>
Параметры фабрики корневого кэша YT-клиентов, установленной через SetRootClientsCacheFactory(). Схему параметров задаёт сама фабрика; встроенная фабрика этот блок игнорирует.

См. также

Предыдущая
Distributed Throttler
Следующая
Быстрый старт