Service Log in YTsaurus Flow

The connector code is available here.

Use the service log to regularly scan sorted tables. Unlike other connectors, it’s inherently infinite: after a table is fully read, it starts reading it again.

You can also scan multiple tables in parallel, joining them by key columns.

Working with replicated tables

The service log supports replicated tables.

To access them directly, you need to specify all replica clusters in the table’s clusters attribute.

To access them via the meta-cluster RPC, it’s enough to specify the meta-cluster in the cluster attribute.

Fullstates

You can specify multiple fetchers in the service log. This can be used for the “fullstate” scenario. A fullstate is a table external to the pipeline that contains data for the same keys for which there is a state in the pipeline. This table is periodically loaded into the pipeline. You can do this efficiently: when generating the service log, join the pipeline’s state table and the fullstate table by keys right away. This way, events are generated for keys that exist in at least one of the fetchers. This approach gives you two useful properties:

• During a single joined pass of the service log, each state is touched once. Instead of twice — once for the main service log event and once for loading the fullstate row.
• The pipeline can react to events where a key exists in the pipeline’s state but not in the external fullstate.

You can distinguish the case where “all non-key columns are null” from the case where “there is no key match” using the ispresent system column.

How to write tests

A common mistake is writing tests for the service log in finite mode (finite = true) while expecting it to see data generated by processing messages from other sources. With this approach, you can’t guarantee that the service log will see the expected rows during the scan: it might scan a certain key range before the rows appear there. Instead, use one of these approaches to write tests correctly:

1. Manually load the rows you want the service log to see before starting the test. With this approach, make sure the test result doesn’t depend on whether the service log sees the other rows.
2. Run two executions: in the first, load data from other sources; in the second, run only the service log source.
3. Make the test non-finite (finite = false).

Source settings

The source class is NYT::NFlow::TServiceLogSource.

Static spec

NYT::NFlow::TUnitedParameters<NYT::NFlow::TServiceLogSource>

Source: yt/yt/flow/library/cpp/common/registry-inl.h

In the static spec of the service log, you define a list of table fetchers to scan, joining them by key columns. Most often, there’s just one fetcher, and the source simply scans a single table.

NYT::NFlow::TTableJoinerSpec

Source: yt/yt/flow/library/cpp/connectors/servicelog/joiner.h

NYT::NFlow::TFetcherInJoinerSpec

Source: yt/yt/flow/library/cpp/connectors/servicelog/joiner.h

In the fetcher settings, you specify which table to scan, which columns to take, and what prefix to use for these columns in the join result (if a join is performed).

Minimal spec example

"table_joiner" = {
    "fetchers" = [
        {
            "table_path" = <path to the table>;
        };
    ];
};

NYT::NFlow::EFetchType

Dynamic spec

NYT::NFlow::TDynamicUnitedParameters<NYT::NFlow::TServiceLogSource>

Source: yt/yt/flow/library/cpp/common/registry-inl.h

See also

• List of connectors
• Stateful processing