Static Table Join in YTsaurus Flow (Java and Kotlin)

This pipeline loads a static reference table into external state and enriches queue events by the shared key. Java and Kotlin use the same SDK and pipeline topology.

Java source

Kotlin source

Pipeline structure

  • reference_reader reads the static reference table;
  • reference_loader normalizes each reference value and writes it to /reference_state through external_state_managers;
  • event_reader reads events from a queue;
  • enricher reads /reference_state through external_state_joiners and publishes only successful matches to enriched.

The manager and joiner point to the same dynamic state table. Their keys must match the group_by_schema of their computations. See the Java pipeline spec or Kotlin pipeline spec.

Load the reference state

@Override
public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
    String name = message.get("name", String.class);
    String normalizedName = name.strip().toLowerCase();

    ExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
    stateAccessor.set(stateAccessor.getOrDefault().toBuilder()
            .set("normalized_name", normalizedName)
            .finish());
}

override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
    val normalizedName = message.get("name", String::class.java)?.trim()?.lowercase() ?: return

    val stateAccessor = ctx.getState(REFERENCE_STATE, message)
    stateAccessor.set(
        stateAccessor.orDefault.toBuilder()
            .set("normalized_name", normalizedName)
            .finish()
    )
}

Enrich events

The read-only descriptor selects the joined row by the current message key. If the row or normalized_name is absent, the example produces no output.

@Override
public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
    ReadOnlyExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
    String normalizedName = stateAccessor.getOrDefault().get("normalized_name", String.class);
    if (normalizedName == null) {
        return;
    }

    output.addMessage(ctx.createMessageBuilder("enriched")
            .set("key", message.get("key", Long.class))
            .set("name", normalizedName)
            .finish());
}

override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
    val stateAccessor = ctx.getState(REFERENCE_STATE, message)
    val normalizedName = stateAccessor.orDefault.get("normalized_name", String::class.java) ?: return

    output.addMessage(
        ctx.createMessageBuilder("enriched")
            .set("key", message.get("key", Long::class.java))
            .set("name", normalizedName)
            .finish()
    )
}

Both computations are discovered through @FlowComputation; the pipeline spec declares the three stream schemas used by the native sources and companion functions.

See also

Previous
URL Downloader
Next
External State Join