Computation in YTsaurus Flow (Java)

Computation types

In Flow, you have two types of Computation: Swift and Transform. Your choice affects how you ensure exactly-once guarantees and what transformations you can implement.

For more on processing guarantees, see the Processing guarantees section.

For Java and Kotlin pipelines, you choose between Swift or Transform by setting computation_class_name in the static spec:

• NYT::NFlow::NCompanion::TTransformCompanionComputation — for Transform.
• NYT::NFlow::NCompanion::TSwiftMapCompanionComputation — for Swift.
• NYT::NFlow::NCompanion::TTransformOrderedSourceCompanionComputation — for Transform source.
• NYT::NFlow::NCompanion::TSwiftOrderedSourceCompanionComputation — for Swift source.

Creating a Computation

In Java and Kotlin code, you create a Computation using Computation.Builder and register it in PipelineContext.

var join = Computation.builder()
        .setComputationId("join")
        .setProcessFunction(new JoinProcessFunction())
        .build();

val join = Computation.builder()
        .setComputationId("join")
        .setProcessFunction(JoinProcessFunction())
        .build()

In the static spec, you create a Computation with the same id (in this example, join):

"join" = {
    "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
    "group_by_schema" = [
        ...
    ];
    "input_stream_ids" = [...];
    "output_stream_ids" = [...];
    "parameters" = {
        ...
    };
    "timers" = {};
};

For more on specs, see the Spec, DynamicSpec, and Config section.

SourceComputation

SourceComputation is the top node in the pipeline graph that reads data from external sources. On the worker side, it corresponds to TSwiftOrderedSourceComputation or TTransformOrderedSourceComputation.

In Java, SourceComputation extends Computation. Like Computation, it requires the processFunction parameter.

In the static spec, for deterministic processing without custom state, you specify TSwiftOrderedSourceCompanionComputation. If SourceComputation uses internal state or non-deterministic logic, you specify TTransformOrderedSourceCompanionComputation: the worker materializes the output and records it together with the state and the source offset. The internal state key in such a computation is the source partition key.

Parameters

Creating a SourceComputation

var reader = SourceComputation.builder()
        .setComputationId("hit_reader")
        .setProcessFunction(new HitParsingFunction())
        .build();

val reader = SourceComputation.builder()
        .setComputationId("hit_reader")
        .setProcessFunction(HitParsingFunction())
        .build()

For a passthrough Source, don’t use Java. Specify NYT::NFlow::TSwiftPassthroughOrderedSourceComputation in computation_class_name in the spec and leave the computation unregistered in the Java companion. For more details, see Passthrough Computation.

Interaction with Worker

When you initialize Worker, it requests information about registered Computation and SourceComputation objects from the Java companion. The source computation on the worker side sends input messages to the Java companion, which applies ProcessFunction to them and returns the result.

Process Function

You implement the business logic for data processing with a Process Function. Choose one of these two interfaces: RowFunction or BatchFunction.

RowFunction

Source code

RowFunction receives messages and timers one at a time. The interface provides two methods:

• onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) — called for each input message.
• onTimer(Timer timer, OutputCollector output, RuntimeContext ctx) — called when a timer fires.

Example of a stateless function

public class X2Mapper implements RowFunction {
    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        var messageBuilder = ctx.createMessageBuilder("x2_numbers"); //1
        Long number = message.get("number", Long.class);             //2
        messageBuilder.set("number_x2", number * 2);                 //3
        output.addMessage(messageBuilder.finish());                  //4
    }
}

class X2Mapper : RowFunction {
    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val messageBuilder = ctx.createMessageBuilder("x2_numbers") //1
        val number: Long? = message.get("number", Long::class.java)  //2
        messageBuilder.set("number_x2", number!! * 2)                //3
        output.addMessage(messageBuilder.finish())                   //4
    }
}

Let’s walk through the code line by line:

1. ctx.createMessageBuilder("x2_numbers") — you create a MessageBuilder for the output stream with id = x2_numbers. The stream with this identifier must be present in the output_stream_ids list in the computation’s static spec.
2. message.get("number", Long.class) — you get the value of the number field from the incoming message. You must pass the value’s class to the Message#get method to unambiguously convert the serialized form to a Java object.
3. messageBuilder.set("number_x2", number * 2) — you write the value to the number_x2 field. This field must be present in the schema of the x2_numbers stream in the static spec.
4. output.addMessage(messageBuilder.finish()) — the finish method returns the completed message, which you add to the OutputCollector.

BatchFunction

Source code

BatchFunction receives the entire list of messages and timers that come from the worker. The interface provides two methods:

• onMessages(List<ExtendedMessage> messages, OutputCollector output, RuntimeContext ctx) — called for a batch of messages.
• onTimers(List<Timer> timers, OutputCollector output, RuntimeContext ctx) — called for a batch of timers.

A batch corresponds to one worker request and may contain messages with different keys; per-key grouping, if needed, is done in user code (see Companion).

Example of a batch function

public class X2BatchMapper implements BatchFunction {
    @Override
    public void onMessages(List<ExtendedMessage> messages, OutputCollector output, RuntimeContext ctx) {
        var messageBuilder = ctx.createMessageBuilder("x2_numbers"); //1
        for (var message : messages) {                               //2
            Long number = message.get("number", Long.class);         //3
            messageBuilder.set("number_x2", number * 2);             //4
            output.addMessage(messageBuilder.finish());              //5
        }
    }
}

class X2BatchMapper : BatchFunction {
    override fun onMessages(messages: List<ExtendedMessage>, output: OutputCollector, ctx: RuntimeContext) {
        val messageBuilder = ctx.createMessageBuilder("x2_numbers") //1
        for (message in messages) {                                  //2
            val number: Long? = message.get("number", Long::class.java) //3
            messageBuilder.set("number_x2", number!! * 2)           //4
            output.addMessage(messageBuilder.finish())               //5
        }
    }
}

The key differences from RowFunction are:

• You create MessageBuilder once for the entire batch (line 1).
• The finish() method returns the completed message and resets MessageBuilder to its initial state, so you can reuse it for the next message (line 5).

Registering in PipelineContext

You must register all Computation objects and typed streams (created via FlowStreams.typed) in PipelineContext before you run GrpcServerExecution.
You don’t need to register untyped streams (created via FlowStreams.raw). Flow creates them automatically based on the streams block in the static spec.

Learn more about Typed Streams.

var context = new PipelineContext();

// Register Computation objects.
Computation join = Computation.builder()
        .setComputationId("join")
        .setProcessFunction(new JoinProcessFunction())
        .build();
context.registerComputation(join);

SourceComputation reader = SourceComputation.builder()
        .setComputationId("hit_reader")
        .setProcessFunction(new HitParsingFunction())
        .build();
context.registerComputation(reader);

// Register typed streams.
context.registerStream(FlowStreams.typed("hit", Hit.class));
context.registerStream(FlowStreams.typed("action", Action.class));
context.registerStream(FlowStreams.typed("joined_action", JoinedAction.class));

val context = PipelineContext()

// Register Computation objects.
val join: Computation = Computation.builder()
        .setComputationId("join")
        .setProcessFunction(JoinProcessFunction())
        .build()
context.registerComputation(join)

val reader: SourceComputation = SourceComputation.builder()
        .setComputationId("hit_reader")
        .setProcessFunction(HitParsingFunction())
        .build()
context.registerComputation(reader)

// Register typed streams.
context.registerStream(FlowStreams.typed("hit", Hit::class.java))
context.registerStream(FlowStreams.typed("action", Action::class.java))
context.registerStream(FlowStreams.typed("joined_action", JoinedAction::class.java))

RuntimeContext

RuntimeContext source code

StatefulContext source code

RuntimeContext gives you access to the computation’s execution context. Key methods:

Learn more about working with states in the Working with States (Java) section.

OutputCollector

Source code

Use OutputCollector to send processing results:

For a Swift computation, MessageIdSuffix.payloadHash() or MessageIdSuffix.userDefined(value) makes message identity independent of emission order:

output.addMessage(
        message,
        AddMessageOptions.builder()
                .setMessageIdSuffix(MessageIdSuffix.payloadHash())
                .build());

Default options use the message sequence number. Payload-hash and user-defined suffixes are supported only by Swift computations.

Spring Boot

When you use Spring Boot, you register the computation with the @FlowComputation annotation (or @FlowSourceComputation for a source) directly on the ProcessFunction class. The annotation is meta-annotated with @Component, so the class automatically becomes a Spring bean:

@FlowComputation(id = "mapper")
public class WordCountMapper implements RowFunction {
    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        // process the message
    }
}

@FlowComputation(id = "mapper")
class WordCountMapper : RowFunction {
    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        // process the message
    }
}

Declare streams as Spring beans of type FlowStream<?> (or via ComputationProvider.getStreams()):

@Configuration
public class StreamConfiguration {

    @Bean
    public FlowStream<Word> wordsStream() {
        return FlowStreams.typed("words", Word.class);
    }
}

@Configuration
class StreamConfiguration {

    @Bean
    fun wordsStream(): FlowStream<Word> = FlowStreams.typed("words", Word::class.java)
}

FlowStreams.typed(...) creates a typed stream that automatically serializes and deserializes messages into Java objects of the specified type. Learn more in the Typed Streams section.

Learn more about registration via annotations and ComputationProvider in the Spring Boot Integration section.

CompanionManager Resource Configuration

To run a companion in Java or Kotlin, you must declare the CompanionManager resource in the static spec:

"CompanionManager" = {
    "resource_class_name" = "NYT::NFlow::NCompanion::TJavaCompanionManager";
    "parameters" = {
        "timeout" = "10s";
        "jdk_bin_path" = "/app/ytflow/jdk/bin/java";
        "main_class" = "tech.ytsaurus.flow.examples.waitclickjoin.PipelineMain";
        "classpath" = "/app/ytflow/lib/*";
    };
    "dependencies" = {};
};

The resource_class_name parameter specifies the resource class that will run the companion.
For a Java or Kotlin companion, resource_class_name must always be NYT::NFlow::NCompanion::TJavaCompanionManager (it supports both languages via the JVM).

Learn more about the spec in the Spec, DynamicSpec and Config section.

See also

• Computation (concept)
• Working with States (Java)
• Quick Start (Java)
• Companion