Pipeline Object in YTsaurus

This section describes the Cypress object pipeline — the deployment unit of YTsaurus Flow. It covers the data model, creation methods, internal tables, and management tools.

Data Model

In YTsaurus, a pipeline is a special Cypress object of type pipeline. Structurally, it’s a map node — a directory where a set of service internal tables required for Flow operation are automatically created.

Internal Tables

Table 1 — Pipeline internal tables

Table name	Purpose
`input_messages`	Index of input messages for transform computations, used for deduplication.
`compact_input_messages`	Compact index of input messages. It’s used by default for all computations, except when `experimental_enable_non_uint_key` is enabled; the behavior is overridden by the `use_compact_input_messages` parameter in TComputationSpec.
`compact_output_messages`	Not used.
`compact_partition_output_messages`	Output messages of transform computations, physically grouped by partitions and chunked by `stream_id` / `chunk_id` for optimal reading.
`states`	User and service states stored by key.
`partition_states`	States stored by partition.
`timers`	Timers of the user code.
`controller_logs`	Event logs of the Controller in the PublicFlowController category.
`flow_state`	Current Flow state.
`flow_state_obsolete`	Flow KV storage for named objects (spec, dynamic_spec, etc.).
`flow_control`	Published controller address; also the leader lease with the Dyntable election backend.
`key_visitor_states`	Durable key-range scan cursor and pass state.
`partition_transactions`	Service table for safe retry of transactions.
`leases`	Dyntable election backend: partition lease owners and the pipeline-wide lease deadline that fences stale workers.
`leader_election_lock`	Controller election lock for the Chaos backend.

After you create a pipeline, these tables appear under the path <pipeline_path>/<table_name> and are automatically mounted.

The table set, schemas, and base attributes for create pipeline are defined in definitions.yson.

The yt_sync_mini library also applies physical-attribute presets.

The input_messages and compact_input_messages tables hold temporary records used for exactly-once deduplication, not a history of input. The controller advances their SystemWatermark, and dynamic table cleanup removes processed records, including after the pipeline finishes. An empty input_messages table is therefore normal and does not mean input was lost.

Attention

The internal tables are service tables, and their structure may change between Flow releases. Don’t read from or write to them directly from your user code. To read debugging data (for example, controller logs), use yt flow show-logs and other commands from the yt flow family.

Creating a Pipeline

Via yt_sync_mini

The recommended way to create a pipeline in open source is the Python library yt_sync_mini. It creates a map node of type pipeline, all internal tables with correct schemas and physical attributes, and mounts them immediately. The operation is idempotent — running it again on an existing pipeline is a no-op.

import yt.wrapper as yt

from yt.yt.flow.library.python.yt_sync_mini import create_pipeline

client = yt.YtClient(proxy="<cluster>")
create_pipeline(client, "<pipeline_path>")

Low-level Creation of a Cypress Node

To integrate with an existing deployment system, you can create the pipeline using the standard create mechanism, as with other Cypress object types (table, map_node, queue_consumer, etc.). By default, create pipeline creates and mounts the internal tables. Pass initialize_tables=%false when creating the object to disable initialization; you are then responsible for creating and mounting the tables with the correct schemas and attributes.

Via YTsaurus CLI

yt --proxy <cluster> create pipeline <pipeline_path>

To create the internal tables yourself:

yt --proxy <cluster> create pipeline <pipeline_path> --attributes '{initialize_tables=%false}'

Via Python (YTsaurus wrapper)

import yt.wrapper as yt

client = yt.YtClient(proxy="<cluster>")
client.create(
    "pipeline",
    "<pipeline_path>"
)

Via C++ (YTsaurus native client)

#include <yt/yt/flow/library/cpp/native_client/pipeline_init.h>

NYT::NApi::TCreateNodeOptions options;

auto nodeId = NYT::NFlow::CreatePipelineNode(client, pipelinePath, options);

External State

When the schema of the internal tables changes in a new Flow version, the format upgrade is performed via a separate migration — see Pipeline internal tables and Basic rollout rules.

If the pipeline uses External State (user tables outside the node), creating them and evolving their schemas is your responsibility. You perform the operations using standard commands like yt create table ... --attributes '{dynamic=%true; schema=...}' and yt mount-table — see examples in the Create command section.