URL Downloader in YTsaurus Flow (Java)

The pipeline groups incoming URLs by host, accumulates them in the internal state, and processes them in batches based on a timer. The timer triggers 5 seconds after a URL is added to the queue and emits the results to the output stream.

Source code (Java)

Source code (Kotlin)

Components

UrlDownloadFunction

This is the main process function that implements the logic for accumulating and processing URLs. The onMessage method adds a URL to the host’s internal state and sets a timer. The onTimer method processes the accumulated URLs and emits the results.

Java

Kotlin

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        UrlMessage urlMessage = message.getPayload();
        String host = urlMessage.getHost();
        String url = urlMessage.getUrl();

        StateAccessor<HostState> accessor = ctx.getState(HOST_STATE, message);
        HostState state = accessor.getOrDefault(new HostState(host));
        state.getPendingUrls().add(url);

        output.addTimer(System.currentTimeMillis() / 1000 + 5, 0L);
        log.debug("Queued url (Host: {}, Url: {})", host, url);
    }

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val urlMessage: UrlMessage = message.getPayload()
        val host = urlMessage.host
        val url = urlMessage.url

        val accessor = ctx.getState(HOST_STATE, message)
        val state = accessor.getOrDefault(HostState(host))
        state.pendingUrls!!.add(url)

        output.addTimer(System.currentTimeMillis() / 1000 + 5, 0L)
        log.debug("Queued url (Host: {}, Url: {})", host, url)
    }

Java

Kotlin

    @Override
    public void onTimer(Timer timer, OutputCollector output, RuntimeContext ctx) {
        StateAccessor<HostState> accessor = ctx.getState(HOST_STATE, timer);
        HostState state = accessor.get();
        if (state == null || state.getPendingUrls().isEmpty()) {
            accessor.clear();
            return;
        }
        String host = state.getHost();

        for (String url : new ArrayList<>(state.getPendingUrls())) {
            String data = processUrl(url);
            ProcessedUrl processedUrl = new ProcessedUrl(host, url, data);
            output.addMessage(new Message("processed_urls", processedUrl));
            log.debug("Processed url (Host: {}, Url: {})", host, url);
        }
        accessor.clear();
    }

    override fun onTimer(timer: Timer, output: OutputCollector, ctx: RuntimeContext) {
        val accessor = ctx.getState(HOST_STATE, timer)
        val state = accessor.get()
        if (state == null || state.pendingUrls.isNullOrEmpty()) {
            accessor.clear()
            return
        }
        val host = state.host

        for (url in ArrayList(state.pendingUrls)) {
            val data = processUrl(url!!)
            val processedUrl = ProcessedUrl(host, url, data)
            output.addMessage(Message("processed_urls", processedUrl))
            log.debug("Processed url (Host: {}, Url: {})", host, url)
        }
        accessor.clear()
    }

HostState

This is the internal state model that is serialized to YSON. It stores the host name and the list of URLs that are waiting to be processed.

Java

Kotlin

@Entity
public class HostState {
    private String host;

    @Column(name = "pending_urls")
    private List<String> pendingUrls = new ArrayList<>();

    public HostState() {
    }

    public HostState(String host) {
        this.host = host;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public List<String> getPendingUrls() {
        return pendingUrls;
    }

    public void setPendingUrls(List<String> urls) {
        this.pendingUrls = urls;
    }
}

@Entity
class HostState {
    var host: String? = null

    @Column(name = "pending_urls")
    var pendingUrls: MutableList<String?>? = ArrayList()

    constructor()

    constructor(host: String?) {
        this.host = host
    }
}

Registering the computation and streams

You register the url_downloader computation with the @FlowComputation annotation on the process function class.

Java

Kotlin

@FlowComputation(id = "url_downloader")

@FlowComputation(id = "url_downloader")

You declare typed streams via ComputationProvider (the getStreams() method).

Java

Kotlin

@Configuration
public class UrlDownloaderComputationContext implements ComputationProvider {

    @Override
    public List<FlowStream<?>> getStreams() {
        return List.of(
                FlowStreams.typed("urls", UrlMessage.class),
                FlowStreams.typed("processed_urls", ProcessedUrl.class)
        );
    }
}

@Configuration
open class UrlDownloaderComputationContext : ComputationProvider {

    override fun getStreams(): List<FlowStream<*>> {
        return listOf(
            FlowStreams.typed("urls", UrlMessage::class.java),
            FlowStreams.typed("processed_urls", ProcessedUrl::class.java),
        )
    }
}

PipelineMain

This single entry point starts the pipeline or serves as its companion, depending on YT_FLOW_MODE.

Java

Kotlin

    public static void main(String[] args) throws Exception {
        new SpringApplicationBuilder(PipelineMain.class).run(args);
    }

        @JvmStatic
        fun main(args: Array<String>) {
            SpringApplicationBuilder(PipelineMain::class.java)
                .run(*args)
        }

Key patterns

Grouping by key: you create a separate state for each host. Flow automatically routes messages with the same key to a single computation instance.
Wall-clock time-based timer: output.addTimer(System.currentTimeMillis() / 1000 + 5, 0L) starts processing 5 seconds after a URL is added to the queue. Multiple addTimer calls with the same triggerTimestamp are deduplicated.
Batch processing in onTimer: all accumulated URLs are processed at once when the timer triggers. This reduces the number of calls to downstream services.
State cleanup: after processing, you delete the state via accessor.clear() to prevent memory leaks.
YsonStateAccessor: the internal state is serialized to YSON and stored on the C++ worker side. You get the Java object via getOrDefault.

URL Downloader in YTsaurus Flow (Java)

Components

UrlDownloadFunction

HostState

Registering the computation and streams

PipelineMain

Key patterns

See also