This connector works with YTsaurus queues (QYT).
You can find the connector code here.
Reading from a Queue
When you read from a queue using the simplest method, messages arrive in Computation with the same schema as the queue table rows.
By default, the message’s event_time and system_time are calculated based on the $timestamp column, which contains the YT timestamp of the row record (for more details, see the documentation for ordered dynamic tables).
But if the queue has a column with metadata (see the static spec parameters for the source), the message’s event_time can be taken from the corresponding field. You can also get stream watermark information from the metadata.
The source class is NYT::NFlow::TQueueSource.
Below are the source settings.
Note
The source is defined by the cluster and the queue path (queue_path). If you change them, the partitions are recreated and read again — see Changing the Source.
Managing Consumer Lag
When you register a consumer for a queue, the consumer lag is set to the last stored data in the queue. So, after registration, if the policy for deleting old rows in the queue is 1 day, the lag will also be 1 day.
This behavior can lead to the following scenario: when you roll out a release, the data lag will be 1 day, and message processing will pause until the lag is processed.
You can avoid this in several ways:
1. Using the Standard YT CLI
Use the standard YT CLI to set the consumer offsets to the desired values for each partition. This isn’t very convenient for the first read of a topic, because you need to know the value of the latest offset and set it for each partition.
Here’s an example command:
yt advance-queue-consumer --proxy <cluster-name> //home/service/stable/consumer <cluster-name>://home/source_service/Data/queue --partition-index 0 --new-offset 486482013
For more advanced options, check the YT documentation or use the -h parameter.
Static spec
Source: yt/yt/flow/library/cpp/common/registry-inl.h
|
Parameter
|
Description
|
|
finite
|
Type: bool
Default value: false
Treat the source as finite: transition the stream to the completed state after reading all available messages. How the end of the data is determined depends on the connector.
|
|
queue_path
|
Type: NYT::NYPath::TRichYPath
Required parameter
Path to the queue including the cluster.
|
|
consumer_path
|
Type: NYT::NYPath::TRichYPath
Required parameter
Path to the queue consumer with the cluster specified.
|
|
try_parse_flow_queue_meta
|
Type: bool
Default value: true
Whether to retrieve and parse meta from the queue (the meta contains watermarks provided by the writer to the queue).
|
|
flow_queue_meta_column
|
Type: std::string
Default value: flow_queue_meta
Column from which to retrieve meta.
|
|
ignore_malformed_flow_queue_meta
|
Type: bool
Default value: false
Whether to ignore invalid meta or fail.
|
|
partition_filter
|
Type: std::optional<std::vector<std::pair<int, int>>>
|
Additional parameters
|
update_info_period
|
Type: TDuration
Default value: 15s
Period for refreshing the source's auxiliary partition information. A connector may use this tick for status requests and read-session liveness checks.
|
|
byte_size_alpha
|
Type: double
Default value: 0.05
Exponential smoothing coefficient for the average byte total and message count per offset: larger values make the estimate react to new data faster.
|
|
update_partition_count_period
|
Type: TDuration
Default value: 1m
How often the controller should update the partition count in the queue (the controller changes the computation partition count according to the queue partition count).
|
|
update_partition_count_retry_min_backoff
|
Type: TDuration
Default value: 1s
Initial delay before retrying a failed partition count update, subject to jitter. The value is capped by update_partition_count_period; subsequent delays increase exponentially.
|
Dynamic spec
Source: yt/yt/flow/library/cpp/common/registry-inl.h
|
Parameter
|
Description
|
|
pull_queue_timeout
|
Type: TDuration
Default value: 1m
Timeout for a read request from the queue.
|
Additional parameters
|
unavailable_threshold
|
Type: TDuration
Default value: 5m
How long the source must be continuously unavailable for the partition to count as stably unavailable. Only time during which the job was running and seeing the failure counts: a gap between restarts is not charged, and any successful answer from the source resets the total.
|
Writing to a Queue
To write messages to a queue in a sink, send messages with the same schema as the table where you plan to write. You can also configure writing metadata to a special column to pass information about the message’s event_time and the stream watermark to queue readers.
There are two sink options: synchronous (NYT::NFlow::TSyncQueueSink) and asynchronous (NYT::NFlow::TAsyncQueueSink). Writing to a synchronous sink happens in the main transaction of the epoch. This is efficient, but you can only write to a queue on the main processing cluster. Writing to an asynchronous sink happens after the main epoch transaction, using messages stored in output messages. This is more resource-intensive, but you can write to a queue on any cluster.
Parameters for synchronous sink specs
Static spec
Source: yt/yt/flow/library/cpp/common/registry-inl.h
|
Parameter
|
Description
|
|
queue_path
|
Type: NYT::NYPath::TRichYPath
Required parameter
Path to the queue including the cluster.
|
|
write_flow_queue_meta
|
Type: bool
Default value: false
Whether to write meta to the queue (watermarks are written in meta).
|
|
flow_queue_meta_column
|
Type: std::string
Default value: flow_queue_meta
Column to which to write meta.
|
|
tablet_index_expression
|
Type: std::optional<std::string>
Verbatim routing mode. A QL expression (builtins only, e.g. farm_hash) over the message columns whose value is written to the system column $tablet_index verbatim. Mutually exclusive with tablet_index_routing_hash_expression; when neither expression is set, routing is off ($tablet_index is not written and the tablet is chosen by the driver). Sharded queues are inconvenient to operate (object→tablet affinity is tightly coupled and a reshard remaps keys), so enable routing only when genuinely needed. Routing is supported only on sync queue sinks; async sinks reject the routing params at spec load (YTFLOW-766).
|
|
tablet_index_routing_hash_expression
|
Type: std::optional<std::string>
Hash routing mode. A QL expression (builtins only, e.g. farm_hash) yielding a uint64 hash that is reduced to a tablet index by tablet_index_routing_hash_policy over tablet_count. Mutually exclusive with tablet_index_expression. Supported only on sync queue sinks; async sinks reject the routing params (YTFLOW-766).
|
|
tablet_index_routing_hash_policy
|
Type: std::optional<NYT::NFlow::EQueueTabletIndexRoutingHashPolicy>
The policy for reducing the tablet_index_routing_hash_expression hash to a tablet index. Required together with it. range — contiguous equal-width hash ranges (rangeSize = 2^64 / tablet_count), recommended: a consumer range-partitioned by the same key reads only its own tablet. modulo — hash % tablet_count; discouraged: Flow computations are range-partitioned by key, so a modulo-partitioned queue forces every reader to read every tablet (a full mesh on read).
|
|
tablet_count
|
Type: std::optional<long>
The number of tablets used to reduce tablet_index_routing_hash_expression to a tablet index. Optional. When unset, the sink periodically re-reads the target queue's @tablet_count (from the master cache) and follows a reshard without a restart. When set explicitly, the value is fixed (no queue read) and a reshard is reflected only by changing the config.
|
|
column_filter
|
Type: std::optional<THashSet<std::string>>
Which columns from the message to write to the queue (default — all).
|
Additional parameters
|
update_partition_count_period
|
Type: TDuration
Default value: 1m
How often the controller should update the partition count in the queue (the controller changes the computation partition count according to the queue partition count).
|
|
update_partition_count_retry_min_backoff
|
Type: TDuration
Default value: 1s
Initial delay before retrying a failed partition count update, subject to jitter. The value is capped by update_partition_count_period; subsequent delays increase exponentially.
|
Dynamic spec
Source: yt/yt/flow/library/cpp/common/registry-inl.h
This structure has no main parameters.
Additional parameters
|
flow_queue_meta_heartbeat_period
|
Type: TDuration
Default value: 10s
How often the controller will write heartbeats with meta containing watermarks to all queue partitions.
|
Parameters for asynchronous sink specs
Static spec
Source: yt/yt/flow/library/cpp/common/registry-inl.h
|
Parameter
|
Description
|
|
queue_path
|
Type: NYT::NYPath::TRichYPath
Required parameter
Path to the queue including the cluster.
|
|
write_flow_queue_meta
|
Type: bool
Default value: false
Whether to write meta to the queue (watermarks are written in meta).
|
|
flow_queue_meta_column
|
Type: std::string
Default value: flow_queue_meta
Column to which to write meta.
|
|
producer_path
|
Type: NYT::NYPath::TRichYPath
Required parameter
Path to the queue producer with the cluster specified.
|
|
require_sync_replica
|
Type: bool
Default value: true
The eponymous parameter when writing to the queue. Whether write to the queue without synchronous replicas is allowed.
|
|
tablet_index_expression
|
Type: std::optional<std::string>
Verbatim routing mode. A QL expression (builtins only, e.g. farm_hash) over the message columns whose value is written to the system column $tablet_index verbatim. Mutually exclusive with tablet_index_routing_hash_expression; when neither expression is set, routing is off ($tablet_index is not written and the tablet is chosen by the driver). Sharded queues are inconvenient to operate (object→tablet affinity is tightly coupled and a reshard remaps keys), so enable routing only when genuinely needed. Routing is supported only on sync queue sinks; async sinks reject the routing params at spec load (YTFLOW-766).
|
|
tablet_index_routing_hash_expression
|
Type: std::optional<std::string>
Hash routing mode. A QL expression (builtins only, e.g. farm_hash) yielding a uint64 hash that is reduced to a tablet index by tablet_index_routing_hash_policy over tablet_count. Mutually exclusive with tablet_index_expression. Supported only on sync queue sinks; async sinks reject the routing params (YTFLOW-766).
|
|
tablet_index_routing_hash_policy
|
Type: std::optional<NYT::NFlow::EQueueTabletIndexRoutingHashPolicy>
The policy for reducing the tablet_index_routing_hash_expression hash to a tablet index. Required together with it. range — contiguous equal-width hash ranges (rangeSize = 2^64 / tablet_count), recommended: a consumer range-partitioned by the same key reads only its own tablet. modulo — hash % tablet_count; discouraged: Flow computations are range-partitioned by key, so a modulo-partitioned queue forces every reader to read every tablet (a full mesh on read).
|
|
tablet_count
|
Type: std::optional<long>
The number of tablets used to reduce tablet_index_routing_hash_expression to a tablet index. Optional. When unset, the sink periodically re-reads the target queue's @tablet_count (from the master cache) and follows a reshard without a restart. When set explicitly, the value is fixed (no queue read) and a reshard is reflected only by changing the config.
|
|
at_most_once_strategy
|
Type: NYT::TIntrusivePtr<NYT::NFlow::TAtMostOnceStrategyParameters>
Optional at-most-once delivery strategy. Connector support varies; check the connector documentation before enabling it.
|
|
column_filter
|
Type: std::optional<THashSet<std::string>>
Which columns from the message to write to the queue (default — all).
|
Additional parameters
|
update_partition_count_period
|
Type: TDuration
Default value: 1m
How often the controller should update the partition count in the queue (the controller changes the computation partition count according to the queue partition count).
|
|
update_partition_count_retry_min_backoff
|
Type: TDuration
Default value: 1s
Initial delay before retrying a failed partition count update, subject to jitter. The value is capped by update_partition_count_period; subsequent delays increase exponentially.
|
Dynamic spec
Source: yt/yt/flow/library/cpp/common/registry-inl.h
|
Parameter
|
Description
|
|
at_most_once_strategy
|
Type: NYT::TIntrusivePtr<NYT::NFlow::TAtMostOnceStrategyDynamicParameters>
Default value: {}
Dynamic parameters for at_most_once_strategy. Connector support varies; check the connector documentation before configuring them.
|
Additional parameters
|
flow_queue_meta_heartbeat_period
|
Type: TDuration
Default value: 10s
How often the controller will write heartbeats with meta containing watermarks to all queue partitions.
|
|
write_period
|
Type: TDuration
Default value: 100ms
|
|
max_rows_per_write
|
Type: long
Default value: 1000
|
|
max_bytes_per_write
|
Type: long
Default value: 1048576
|
|
backoff_duration
|
Type: TDuration
Default value: 1s
|
See also