Static Table Join in YTsaurus Flow (Python)

This pipeline loads a static reference table into external state and enriches queue events by the shared key.

Pipeline structure

reference_reader reads the static table;
reference_loader normalizes reference values and writes them to /reference_state;
event_reader reads queue events;
enricher performs a read-only lookup through the joiner named /reference_state and writes successful matches to enriched.

The manager and joiner point to the same dynamic state table. Their keys must match the computations' group_by_schema. See the pipeline spec.

Load and join

The loader uses ctx.external_state(...) and persists the normalized value. The enricher uses ctx.joined_external_state(...); when normalized_name is absent, it emits nothing.

class ReferenceLoader(RowFunction):
    """Trims and lowercases the reference name, storing it in external state."""

    def on_message(self, message, output, ctx):
        name = message.payload["name"]
        normalized_name = name.strip().lower()

        state = ctx.external_state("/reference_state", message)
        builder = state.to_builder()
        builder.set("normalized_name", normalized_name)
        state.set(builder.finish())

class Enricher(RowFunction):
    """Joins each event against the reference state and emits the enriched row."""

    def on_message(self, message, output, ctx):
        state = ctx.joined_external_state("/reference_state", message)
        normalized_name = state.get("normalized_name")
        if normalized_name is None:
            return

        builder = ctx.message_builder("enriched")
        builder.set("key", message.payload["key"])
        builder.set("name", normalized_name)
        output.add_message(builder.finish())

Static Table Join in YTsaurus Flow (Python)

Pipeline structure

Load and join

See also