StarRocks SQL connector - Realtime Compute for Apache Flink

Background

StarRocks is a next-generation Massively Parallel Processing (MPP) data warehouse that delivers extremely fast performance across all scenarios and a unified analytics experience. StarRocks offers the following advantages:

StarRocks is compatible with the MySQL protocol, enabling you to use MySQL clients and common Business Intelligence (BI) tools to connect and analyze data.
StarRocks uses a distributed architecture:
- It horizontally partitions data tables and stores them with multiple replicas.
- The cluster can be scaled flexibly and can analyze up to 10 petabytes (PB) of data.
- It uses an MPP framework to accelerate parallel computations.
- It supports multiple replicas to provide fault tolerance.

The Flink connector caches data and uses Stream Load to write it in batches to sink tables. It reads from source tables by fetching data in batches. The following table lists the capabilities of the StarRocks connector.

Category	Description
Supported types	source tables, dimension tables, sink tables, and data ingestion targets
Execution mode	streaming mode and batch mode
Data format	CSV
Connector-specific metrics	None
API types	DataStream, SQL, and YAML for data ingestion
Support for updates/deletions in sink tables	Yes

Prerequisites

You must have a StarRocks cluster deployed on EMR or a self-managed one on ECS.

Limitations

Only Ververica Runtime (VVR) 11.1 or later supports joins with dimension tables.
To avoid network access restrictions, add the following StarRocks cluster ports to a security group or firewall whitelist: 9030, 8030, 8040, 9060, 8060, 9020.
When the target StarRocks table contains hidden generated columns, declare only the actual physical columns to be written in the Flink job, and exclude the generated column fields. Hidden generated columns automatically created by StarRocks expression partitioning (such as __generated_partition_column_0) do not accept external writes. By default, the connector constructs write requests based on the full schema, which causes the job to fail. For more information about generated columns in StarRocks, see Generated columns.

Features

StarRocks on E-MapReduce supports CREATE TABLE AS SELECT (CTAS) and CREATE DATABASE AS SELECT (CDAS) statements. CTAS synchronizes the schema and data of a single table, whereas CDAS synchronizes an entire database or multiple tables within the same database. For more information, see Use CTAS and CDAS statements in Realtime Compute for Apache Flink to synchronize data from a MySQL database to StarRocks.

Syntax

CREATE TABLE USER_RESULT(
 name VARCHAR,
 score BIGINT
 ) WITH (
 'connector' = 'starrocks',
 'jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx',
 'load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port',
 'database-name' = 'xxx',
 'table-name' = 'xxx',
 'username' = 'xxx',
 'password' = 'xxx'
 );

Parameters

Type	Parameter	Description	Type	Required	Default	Remarks
General	connector	Specifies the connector to use.	String	Yes	—	The value must be `starrocks`.
	jdbc-url	The Java Database Connectivity (JDBC) URL.	String	Yes	—	Specify the IP address and JDBC port of the FE in the format `jdbc:mysql://ip:port`.
	database-name	The name of the StarRocks database.	String	Yes	—	—
	table-name	The name of the StarRocks table.	String	Yes	—	—
	username	The username to connect to StarRocks.	String	Yes	—	—
	password	The password to connect to StarRocks.	String	Yes	—	—
	starrocks.create.table.properties	Specifies properties for automatic table creation.	String	No	—	Specifies initial table properties, such as the engine and replica count. For example: 'starrocks.create.table.properties' = 'buckets 8' or 'starrocks.create.table.properties' = 'replication_num=1'.
Source-specific	scan-url	The data scanning URL.	String	No	—	Specifies the IP address and HTTP port of the FE. Format: `fe_ip:http_port;fe_ip:http_port`. Note To specify multiple IP addresses and ports, separate them with commas (,).
	scan.connect.timeout-ms	The timeout for `flink-connector-starrocks` to connect to StarRocks. The connector reports an error if a connection is not established within this timeout.	String	No	1000	Unit: milliseconds.
	scan.params.keep-alive-min	The keep-alive duration for the query task.	String	No	10	—
	scan.params.query-timeout-s	The timeout for a query task. If no result is returned within this period, the system stops the query task.	String	No	600	Unit: seconds.
	scan.params.mem-limit-byte	The memory limit for a single query on a BE node.	String	No	1073741824 (1 GB)	Unit: bytes.
	scan.max-retries	The maximum number of retries for a failed query. The connector reports an error if this limit is exceeded.	String	No	1	—
Sink-specific	load-url	The data import URL.	String	Yes	—	Specify the IP addresses and HTTP ports of the FEs in the format `fe_ip:http_port;fe_ip:http_port`. Note To specify multiple IP addresses and ports, separate them with semicolons (;).
	sink.semantic	The delivery semantics for writes.	String	No	at-least-once	Valid values: `at-least-once` (default): Guarantees that data is delivered at least once. `exactly-once`: Guarantees that data is delivered exactly once.
	sink.buffer-flush.max-bytes	The maximum amount of data to buffer before flushing.	String	No	94371840 (90 MB)	Valid range: 64 MB to 10 GB.
	sink.buffer-flush.max-rows	The maximum number of rows to buffer before flushing.	String	No	500000	Valid range: 1,000 to 5,000,000.
	sink.buffer-flush.interval-ms	The buffer flush interval.	String	No	300000	Valid range: 1,000 ms to 3,600,000 ms.
	sink.max-retries	The maximum number of retries for failed writes.	String	No	3	Valid range: 0 to 1000.
	sink.connect.timeout-ms	The timeout for connecting to StarRocks.	String	No	1000	Valid range: 100 to 60,000. Unit: milliseconds.
	sink.properties.*	Additional Stream Load properties for the sink.	String	No	—	These parameters control the behavior of Stream Load. For example, `sink.properties.format` specifies the format of the imported data, such as CSV. For more parameters, see Stream Load.
Dimension-specific	lookup.cache.enabled	Specifies whether to enable caching for the dimension table.	Boolean	No	true	Valid values: `true`: Enables caching. After the table data is read for the first time, the system caches it in memory. Subsequent requests use the cached data within its validity period to reduce I/O overhead. `false`: Disables caching. Each query directly accesses the data source. Important This feature requires Realtime Compute for Apache Flink engine VVR 11.1 and later. We recommend disabling this feature in the following scenarios: The dimension table data is frequently updated, and real-time data is required. The table contains a large amount of data, posing a risk of memory overflow.

Data type mapping

StarRocks data type	Flink data type
NULL	NULL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
BIGINT UNSIGNED Note Requires Realtime Compute for Apache Flink engine VVR 8.0.10 and later.	DECIMAL(20,0)
LARGEINT	DECIMAL(20,0)
FLOAT	FLOAT
DOUBLE	DOUBLE
DATE	DATE
DATETIME	TIMESTAMP
DECIMAL	DECIMAL
DECIMALV2	DECIMAL
DECIMAL32	DECIMAL
DECIMAL64	DECIMAL
DECIMAL128	DECIMAL
CHAR(m) Note VVR 8.0.10 automatically extends the CHAR length threefold (m=n3, where n<=85) to accommodate encoding differences between MySQL and StarRocks. VVR 8.0.11 and later automatically extend the CHAR length fourfold (m=n4, where n<=63) to accommodate encoding differences between MySQL and StarRocks. The maximum length for the StarRocks CHAR type is 255. Therefore, Flink maps a CHAR type to a StarRocks CHAR type only if its automatically extended length does not exceed 255.	CHAR(n)
VARCHAR(m) Note VVR 8.0.10 automatically extends the VARCHAR length threefold (m=n3, where n>85) to accommodate encoding differences between MySQL and StarRocks. VVR 8.0.11 and later automatically extend the VARCHAR length fourfold (m=n4, where n>63) to accommodate encoding differences between MySQL and StarRocks. The maximum length for the StarRocks CHAR type is 255. Therefore, if the automatically extended length of a Flink CHAR type exceeds 255, Flink maps the type to the StarRocks VARCHAR type.	CHAR(n)
VARCHAR	STRING
VARBINARY Note Requires Realtime Compute for Apache Flink engine VVR 8.0.10 and later.	VARBINARY

Examples

The source table, sink table, and dimension table in the following examples can be used together. The runoob_tbl_source table declared in the source table example also serves as the data source for the sink table and dimension table examples. Before you use these examples, prepare tables with matching field types in StarRocks, and replace the endpoints, database and table names, and secret variables.

Source table

CREATE TEMPORARY TABLE runoob_tbl_source (
    runoob_id BIGINT NOT NULL,
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    proc_time AS PROCTIME() --Processing time column, used by the temporal join in the dimension table example
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'scan-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with commas (,).
    'database-name' = '<database_name>',
    'table-name' = '<source_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'scan.params.query-timeout-s' = '600' --Timeout for a single query, in seconds
);

Note

The source table is read in batch mode. After the job starts, it scans the current data of the table and then stops producing output. It does not continuously produce incremental changes.

Sink table

Note

StarRocks allows a primary key column to be NULLABLE. However, Flink does not support a primary key that contains a nullable column. Flink's data consistency model requires that a primary key is unique and non-nullable. Otherwise, Flink throws the error Invalid primary key. Column 'xxx' is nullable. For more information, see "Invalid primary key. Column 'xxx' is nullable." error.

VVR 11+

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Primary key columns must be declared as NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'load-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with semicolons (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'sink.version' = 'V2', --Transactional Stream Load. Requires StarRocks 2.4 or later.
    'sink.semantic' = 'at-least-once', --Optional. exactly-once is also supported and requires checkpointing.
    'sink.buffer-flush.interval-ms' = '5000' --Batch flush interval, in milliseconds
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

VVR 8+

In VVR 8.x, the default value of sink.version is V1, and V1, V2, and AUTO are supported. Parameters that are registered only in VVR 11.x, such as sink.ignore.update-before, are not supported:

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Primary key columns must be declared as NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'load-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with semicolons (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'sink.semantic' = 'at-least-once', --Optional. exactly-once is also supported and requires checkpointing.
    'sink.buffer-flush.interval-ms' = '5000', --Batch flush interval, in milliseconds
    'sink.max-retries' = '3' --Number of retries after a Stream Load failure
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

Dimension table

Dimension table joins are supported only in VVR 11.1 and later. The dimension table must declare a primary key as the join key, and the temporal join uses processing time. The following example builds on the runoob_tbl_source table from the source table example, which includes the proc_time processing time column:

CREATE TEMPORARY TABLE sr_dim (
    runoob_id BIGINT NOT NULL,
    runoob_author STRING,
    PRIMARY KEY (runoob_id) NOT ENFORCED --The dimension table must declare a primary key as the join key
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'scan-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with commas (,).
    'database-name' = '<database_name>',
    'table-name' = '<dim_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'lookup.cache.ttl-ms' = '5000', --TTL of the dimension table cache, in milliseconds
    'lookup.cache.enabled' = 'true' --Default: true. Set to false to query the table directly without the cache.
);

SELECT o.runoob_id, o.runoob_title, d.runoob_author
FROM runoob_tbl_source AS o
JOIN sr_dim FOR SYSTEM_TIME AS OF o.proc_time AS d --Temporal join that uses the processing time column of the source table
ON o.runoob_id = d.runoob_id;