Computation in YTsaurus Flow

Use a Computation as the main building block of a pipeline. You get messages from input streams, process them, and send the results to output streams.

Computation types

Flow implements four basic Computation types. You’ll find each type described in the sections below.

Classes with Swift in their name follow the Swift principle. This is an approach to data processing without full materialization, while preserving exactly-once guarantees and requiring deterministic transformations.

In C++, these classes define execution modes and are used by built-in adapters. Implement new user logic only as a process function, and select the appropriate TProcessFunction*Computation in the spec. Don’t create user subclasses of the Computation base classes.

TTransformComputation

This mode handles arbitrary transformations of input data. The processing result is stored in YTsaurus, so the transformation needn’t be deterministic. It supports timers, states, and Sink. Run user process functions in this mode with TProcessFunctionComputation; for a passthrough variant without business logic, use TPassthroughComputation.

TSwiftMapComputation

This mode implements a deterministic Map without materializing results in YTsaurus. It doesn’t support timers, Source, or Sink. The process function must be strictly deterministic — the system recomputes the result if needed. Use TProcessFunctionSwiftMapComputation for user logic and TSwiftPassthroughComputation for passthrough.

TSwiftOrderedSourceComputation

This mode reads data from external sources. It requires that the data stream from each instance is ordered. It supports WatermarkStrategy to estimate watermarks. Use TProcessFunctionSourceComputation for user logic and TSwiftPassthroughOrderedSourceComputation for passthrough.

TTransformOrderedSourceComputation

This mode processes Source data with arbitrary custom logic: parsing, filtering, or expanding one message into several. Run the user process function with TProcessFunctionTransformOrderedSourceComputation instead of chaining TSwiftPassthroughOrderedSourceComputation → TProcessFunctionComputation.

The processing result is materialized in YTsaurus, as in TTransformComputation, so there are no determinism requirements: after a restart, Flow delivers the already materialized messages with the MessageId values previously assigned to them instead of recomputing them. The source offset, the materialized output messages, and the states are committed in a single YTsaurus transaction — processing of each source message is applied exactly once, including state updates.

A process function keeps its state in a TMutableStateKeyClient<T> field, initializes it through initContext->InitClient(...) in Init(const IRuntimeInitContextPtr&), and reads it through GetState(message->Key) during processing. For an example, see Process functions (C++).

Supported: source_streams (exactly one ordered Source), several output streams, watermark_strategy (watermark_generator estimates the source watermarks, watermark_alignment aligns reading, event_timestamp_assigner assigns event_timestamp), skip_if_expression, and messages with distribute = false. A non-empty group_by_schema, input streams, timers, and key-visitor streams cause a spec validation error.

Passthrough Computation

A passthrough Computation doesn’t include custom business logic. Incoming messages are converted to the output stream schema and passed on unchanged. Use it to simply align schemas between streams, for example, when reading a queue and moving data to another stream without any processing.

Flow implements three C++ classes:

Class Base class Purpose
TPassthroughComputation TTransformComputation Converts input messages to the output stream schema
TSwiftPassthroughComputation TSwiftMapComputation Does the same, without materialization (Swift)
TSwiftPassthroughOrderedSourceComputation TSwiftOrderedSourceComputation Converts source messages to the output stream

Passthrough is natively implemented in Flow using C++ and doesn’t require a Java or Python companion. To enable it, specify the corresponding C++ class in the computation_class_name field in the Computation’s static spec:

"passthrough" = {
    "computation_class_name" = "NYT::NFlow::TPassthroughComputation";
    "group_by_schema" = [...];
    "input_stream_ids" = [...];
    "output_stream_ids" = [...];
};

For more details, see Computation (C++).

Common properties

  • All execution within a single partition is strictly single-threaded. You achieve multithreading by increasing the number of partitions.
  • All Computations handle filling the message and timer metadata fields.
  • The OutputCollector object collects output messages and timers.
  • The SetParents method lets you manage the lineage of messages to correctly calculate metadata fields.

Implementation in different languages

Each language offers its own set of interfaces for implementing a Computation:

  • C++: implement a process function (IProcessFunction, IBatchProcessFunction, or IKeyedBatchProcessFunction) and select its built-in mode adapter in the spec. Learn more →
  • Java: implement the RowFunction or BatchFunction interfaces with the onMessage/onTimer methods. Learn more →
  • Python: inherit from RowFunction or BatchFunction and use the on_message/on_timer methods. Learn more →
  • Go: implement flow.RowFunction (OnMessage) or flow.BatchFunction (OnMessages); timers use flow.RowTimerFunction or flow.BatchTimerFunction. Learn more →
  • YQL: Computations are generated automatically based on a declarative description. Learn more →

See also

Input filtering

skip_if_expression filters input messages before deduplication and user processing. Skipped messages do not enter the deduplication store. Matching messages are acknowledged without being passed to the handler, including when the entire batch is filtered out. Input statistics, including key distribution and heavy hitters, describe the original stream before filtering.

Previous
The Pipeline object
Next
Swift