Sink data from RisingWave to StarRocks

This guide describes how to sink data from RisingWave to StarRocks.

StarRocks is an open-source, massively parallel processing (MPP) database. For details on how to get started with StarRocks, see the Quick start guide.

The StarRocks stream load does not support sinking struct type.

Prerequisites

Before sinking data from RisingWave to StarRocks, please ensure the following:

• The StarRocks database you want to sink to is accessible from RisingWave.
• Ensure you have an upstream materialized view or source in RisingWave that you can sink data from.

Syntax

CREATE SINK [ IF NOT EXISTS ] sink_name
[FROM sink_from | AS select_query]
WITH (
   connector='starrocks',
   connector_parameter = 'value', ...
);

Parameters

All parameters are required unless specified otherwise.

Parameter names	Description
starrocks.host	The StarRocks host address.
starrocks.query_port	The port to the MySQL server of the StarRocks frontend.
starrocks.http_port	The port to the HTTP server of the StarRocks frontend.
starrocks.user	The user name used to access the StarRocks database.
starrocks.password	The password associated with the user.
starrocks.database	The StarRocks database where the target table is located
starrocks.table	The StarRocks table you want to sink data to.
starrocks.partial_update	Optional. Set it to true to improve performance when you need to update many rows but only change a few columns in each row.
starrocks.max_batch_size_bytes	Optional. The maximum size in bytes for each StarRocks stream load request payload. Set it to a positive integer to have RisingWave split requests by encoded JSON payload size before sending them to StarRocks.
type	Data format. Allowed formats: append-only: Output data with insert operations. upsert: Output data as a chagelog stream. In StarRocks, Primary Key table must be selected.
force_append_only	If true, forces the sink to be append-only, even if it cannot be.
primary_key	Required if type is upsert. The primary key of the downstream table.
commit_checkpoint_interval	Optional. Commit every N checkpoints (N > 0). Default value is 10. The behavior of this field also depends on the sink_decouple setting: If sink_decouple is true (the default), the default value of commit_checkpoint_interval is 10. If sink_decouple is set to false, the default value of commit_checkpoint_interval is 1. If sink_decouple is set to false and commit_checkpoint_interval is set to larger than 1, an error will occur.

Examples

Assume we have a materialized view, bhv_mv.

CREATE SINK bhv_starrocks_sink
FROM bhv_mv WITH (
    connector = 'starrocks',
    type = 'append-only',
    starrocks.host = 'starrocks-fe',
    starrocks.mysqlport = '9030',
    starrocks.httpport = '8030',
    starrocks.user = 'users',
    starrocks.password = '123456',
    starrocks.database = 'demo',
    starrocks.table = 'demo_bhv_table',
    force_append_only='true'
);

Data type mapping

Use this table to choose StarRocks column types compatible with the RisingWave types in your source table or materialized view. For details on native RisingWave data types, see Overview of data types.

StarRocks type	RisingWave type
BOOLEAN	BOOLEAN
SMALLINT	SMALLINT
INT	INTEGER
BIGINT	BIGINT
FLOAT	REAL
DOUBLE	DOUBLE
DECIMAL	DECIMAL
DATE	DATE
VARCHAR	VARCHAR, TIMESTAMP WITH TIME ZONE
No support	TIME
DATETIME	TIMESTAMP WITHOUT TIME ZONE, TIMESTAMP WITH TIME ZONE
CHAR	TIMESTAMP WITH TIME ZONE
STRING	TIMESTAMP WITH TIME ZONE
No support	INTERVAL
No support	STRUCT
ARRAY	ARRAY
No support	BYTEA
JSON	JSONB
BIGINT	SERIAL

When sinking TIMESTAMP WITH TIME ZONE to StarRocks, RisingWave uses the session timezone captured when the sink is created. Set it before CREATE SINK with SET TIME ZONE, for example SET TIME ZONE 'Asia/Shanghai';. The output format is %Y-%m-%d %H:%M:%S%.6f (year-month-day, hour:minute:second with six fractional-second digits), without a timezone suffix. For example, with Asia/Shanghai, 2018-01-26T18:30:09.453Z is written as 2018-01-27 02:30:09.453000.

If a decimal value is out of bounds or represents inf, -inf, or nan, RisingWave will insert null values.