StarRocksDest

class

class StarRocksDest(
    tables: StarRocksTableNamesSpec,
    if_table_exists: IfTableExistsSpec = 'append',
    schema_evolution: SchemaEvolutionSpec = 'strict',
    dest_cfg: DestCfgSpec | None = None,
)

Bases: Dest

Categories: destination

StarRocks destination for @subscriber -- one output slot per table.

The target tables must already exist; this destination does not create them (StarRocks table creation needs physical layout -- a key definition and a DISTRIBUTED BY clause -- that cannot be inferred from the data). Writing to a missing table fails with a StarRocks table-not-found error.

Examples

Append to an existing table -- row count grows each run:

@subscriber(
    destination=StarRocksDest(
        tables=["wld_country"], if_table_exists="append"
    ),
    input_tables=["subs_tables/wld_country"],
)
def out_append(country: TableFrameSpec) -> TableFrameSpec:
    return country

Replace truncates the target table, then reloads it:

@subscriber(
    destination=StarRocksDest(
        tables=["wld_city"], if_table_exists="replace"
    ),
    input_tables=["subs_tables/wld_city"],
)
def out_replace(city: TableFrameSpec) -> TableFrameSpec:
    return city

Parameters

parameter

tablesStarRocksTableNamesSpec (list[str])

Target table names, one write slot per entry; each must reference an existing table. Each is a 1- or 2-part name (table or database.table); backtick-quote a part for special characters.

parameter

if_table_existsIfTableExistsSpec (Literal['append', 'replace'])

"append" (the default) issues INSERT INTO; "replace" issues TRUNCATE + INSERT INTO against the existing table.

parameter

schema_evolutionSchemaEvolutionSpec (Literal['strict', 'iceberg'])

"strict" (the default) rejects schema changes; "iceberg" applies Iceberg-style schema deltas.

parameter

dest_cfgDestCfgSpec | None (Mapping[Literal['starrocks.logging_level'], Any] | None)

Optional connector config; the only supported key is starrocks.logging_level.