--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/connectors/sorted-dynamic-table.md - https://ytsaurus.tech/docs/ru/flow/connectors/sorted-dynamic-table.md - href: https://ytsaurus.tech/docs/en/flow/connectors/sorted-dynamic-table.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Sorted Dynamic Table Sink in YTsaurus Flow Use this connector to write data to [sorted dynamic tables in YTsaurus](https://ytsaurus.tech/docs/en/user-guide/dynamic-tables/sorted-dynamic-tables.md). {% note info "Attention" %} In most cases, use [StateManager](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md) to work with sorted dynamic tables, not this connector. Key differences: - **StateManager** supports read-modify-write operations, row deletion based on non-trivial rules, and guarantees a single modification point for each row. The `group_by_schema` computation must match (or be a prefix of) the state’s key columns. - **This connector** is a low-level primitive that supports only write-only operations (inserting or deleting rows). There are no restrictions on `group_by_schema`, which lets you write to the table from multiple computations with different keys. However, the order of modifications to a single row isn’t controlled. {% endnote %} You can find the connector code [here](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/connectors/sorted_dynamic_table). ## Write modes The sink accepts messages and writes their rows to a sorted dynamic table. The message schema must be compatible with the target table’s schema. The synchronous sink (`NYT::NFlow::NSortedDynamicTable::TSyncSink`) writes rows within the main epoch transaction. The write is atomic together with the other epoch changes, but the table must reside on the main processing cluster — the same cluster as the pipeline. The asynchronous sink (`NYT::NFlow::NSortedDynamicTable::TAsyncSink`) writes each batch in a separate tablet transaction. The target table may reside on another cluster specified in `table_path`. This write is not atomic with the main epoch transaction: the sink acknowledges messages after the separate transaction commits successfully. The asynchronous sink reports the latest YT write error in `/async_write` and retries every error except `Canceled` until the write succeeds or the job is cancelled. Retry delays use exponential backoff with jitter. The delay before jitter starts at `backoff_duration` and is capped at one minute (or at `backoff_duration` when it is greater). Pending messages remain accounted for in the output buffers, so the regular output backpressure applies while the sink is retrying. Within one sink instance, batches are written sequentially. This ordering does not extend across job restarts or partition handoffs: a tablet transaction already started by the old job may commit after the replacement job writes the replayed batch and subsequent batches. As a result, an older modification may overwrite a newer modification of the same key. Use the asynchronous sink only when both replay and this reordering are safe, for example when every key is always written with the same value or the logical version is part of the key. Aggregate writes are unsupported. The number of receiver channels automatically adjusts to match the number of tablets in the target table. For a chaos replicated table, the controller reads the tablet count from the enabled data replica with the lexicographically smallest replica id; its sync/async mode does not affect the choice. Resharding that replica may change the number of receiver channels. ### Synchronous sink parameters Sink class: `NYT::NFlow::NSortedDynamicTable::TSyncSink`. #### Static spec: Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h) #| || **Parameter** | **Description** || || `table_path` | **Type**: `NYT::NYPath::TRichYPath` **Required parameter** Path to the target sorted dynamic table with the cluster specified. || || `column_filter` | **Type**: `std::optional>` Which columns from the message to write to the table (default — all). When using the `delete_rows` parameter, you must specify the key columns of the target table, otherwise deletions will fail. || || `aggregate_columns` | **Type**: `std::optional>` List of columns for which the aggregation function is considered during writing. By default, the value in the column is overwritten. || || `delete_rows` | **Type**: `bool` **Default value**: `false` If set to `true`, rows will be deleted instead of inserted. In this mode, it is important that the key columns of the target table are selected in `column_filter`. {% note info %} A Sink can work in either insert or delete mode, but not simultaneously. If you need both insert and delete functionality, use StateManager. {% endnote %} || || `require_sync_replica` | **Type**: `bool` **Default value**: `true` If set to `false`, the presence of a synchronous replica is not checked when writing rows to the target table. The default value is `true` — the check is performed. || |# {% cut "**Additional parameters**" %} #| || `update_partition_count_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration) **Default value**: `1m` How often the controller should update the number of tablets in the table (the controller changes the number of receiver channels accordingly). For a chaos replicated table, the controller uses the enabled data replica with the lexicographically smallest replica id; its current sync/async mode does not affect the choice. Resharding that replica may change the number of receiver channels. || |# {% endcut %} #### Dynamic spec: Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h) **This structure has no main parameters.** ### Asynchronous sink parameters Sink class: `NYT::NFlow::NSortedDynamicTable::TAsyncSink`. #### Static spec: Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h) #| || **Parameter** | **Description** || || `table_path` | **Type**: `NYT::NYPath::TRichYPath` **Required parameter** Path to the target sorted dynamic table with the cluster specified. || || `column_filter` | **Type**: `std::optional>` Which columns from the message to write to the table (default — all). When using the `delete_rows` parameter, you must specify the key columns of the target table, otherwise deletions will fail. || || `aggregate_columns` | **Type**: `std::optional>` Aggregate writes are unsupported: omit this parameter or provide an empty list. An asynchronous write may replay an already committed batch after recovery, which makes non-idempotent aggregation unsafe. || || `delete_rows` | **Type**: `bool` **Default value**: `false` If set to `true`, rows will be deleted instead of inserted. In this mode, make sure that `column_filter` selects the key columns of the target table. || || `require_sync_replica` | **Type**: `bool` **Default value**: `true` If set to `false`, the presence of a synchronous replica is not checked when writing rows to the target table. The default value is `true`. || |# {% cut "**Additional parameters**" %} #| || `update_partition_count_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration) **Default value**: `1m` How often the controller should update the number of tablets in the table (the controller changes the number of receiver channels accordingly). For a chaos replicated table, the controller uses the enabled data replica with the lexicographically smallest replica id; its current sync/async mode does not affect the choice. Resharding that replica may change the number of receiver channels. || |# {% endcut %} #### Dynamic spec: Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h) #| || **Parameter** | **Description** || || `max_rows_per_batch` | **Type**: `long` **Default value**: `1000` Maximum number of Flow messages in a batch: the batching sink flushes the accumulated batch when it reaches this threshold. The valid range is 1 through 2000. || || `max_bytes_per_batch` | **Type**: `long` **Default value**: `5242880` Maximum total size of Flow messages in a batch: the batching sink flushes the accumulated batch when it reaches this threshold. The valid range is 1 byte through 10 MiB. || |# {% cut "**Additional parameters**" %} #| || `backoff_duration` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration) **Default value**: `3s` Initial retry delay before jitter. Retry delays use exponential backoff with jitter; the delay before jitter is capped at one minute (or at this value when it is greater). The sink keeps retrying until the write succeeds or the job is cancelled, and reports the latest error in its status. || |# {% endcut %}