Realtime Compute for Apache Flink:StarRocks connector

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'
);

Connector options

General options

Source-specific options

Sink-specific options

Dimension table options

Flush policy

The connector buffers data in memory and flushes it to StarRocks via Stream Load. Flush behavior differs by delivery semantics:

• at-least-once: A flush is triggered when any of the following conditions is met:

  • The buffered data size reaches sink.buffer-flush.max-bytes.

  • The buffered row count reaches sink.buffer-flush.max-rows.

  • The time since the last flush reaches sink.buffer-flush.interval-ms.

  • A Flink checkpoint is triggered.

• exactly-once: Flush only happens when a Flink checkpoint is triggered. The sink.buffer-flush.* options have no effect.

Data type mappings

The following table lists the data type mappings between StarRocks and Realtime Compute for Apache Flink.

Sample code

The following example creates a source table and a sink table, then reads from the source and writes to the sink.

CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_source` (
  `runoob_id`      BIGINT  NOT NULL,
  `runoob_title`   STRING  NOT NULL,
  `runoob_author`  STRING  NOT NULL,
  `submission_date` DATE   NULL
) WITH (
  'connector'     = 'starrocks',
  'jdbc-url'      = 'jdbc:mysql://ip:9030',
  'scan-url'      = 'ip:18030',
  'database-name' = 'db_name',
  'table-name'    = 'table_name',
  'username'      = 'xxxxx',
  'password'      = 'xxxxxxx'
);

CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_sink` (
  `runoob_id`      BIGINT  NOT NULL,
  `runoob_title`   STRING  NOT NULL,
  `runoob_author`  STRING  NOT NULL,
  `submission_date` DATE   NULL,
  PRIMARY KEY(`runoob_id`) NOT ENFORCED
) WITH (
  'connector'                    = 'starrocks',
  'jdbc-url'                     = 'jdbc:mysql://ip:9030',
  'load-url'                     = 'ip:18030',
  'database-name'                = 'db_name',
  'table-name'                   = 'table_name',
  'username'                     = 'xxxx',
  'password'                     = 'xxxxxxx',
  'sink.buffer-flush.interval-ms' = '5000'
);

INSERT INTO runoob_tbl_sink SELECT * FROM runoob_tbl_source;

StarRocks allows nullable primary key columns, but Flink requires primary keys to be non-nullable and unique for data consistency. A nullable primary key in StarRocks causes the error: Invalid primary key. Column 'xxx' is nullable. For more information, see Why do I get the error "Invalid primary key. Column 'xxx' is nullable." when writing to a table?.

Usage notes

• The pipeline connector supports only at-least-once semantics. It uses primary key tables to ensure idempotent writes, so duplicate records caused by retries are automatically deduplicated in StarRocks.

• Every source table must have a primary key. For tables without a primary key, specify one in the TRANSFORM statement block before writing to downstream databases:

  transform:
    - source-table: ...
      primary-keys: id, ...

• The bucket key of an automatically created table must match the primary key, and the table cannot contain partition keys.

• New columns are appended to the end of existing columns. Lenient mode is used by default for schema evolution: columns inserted at other positions are automatically moved to the end.

• StarRocks earlier than 2.5.7: explicitly set table.create.num-buckets. StarRocks 2.5.7 and later: the bucket count is determined automatically. For more information, see Data distribution.

• StarRocks 3.2 and later: set table.create.properties.fast_schema_evolution to true to speed up schema changes.

• Stream data issues may arise when ingesting data to EMR Serverless StarRocks using Flink CDC. To resolve this, use one of the following options:

  • Use SQL jobs and set sink.version=V1 in the job draft.

  • Continue to use Flink CDC, but enable FE emr_internal_redirect.

  • Use an EMR Serverless StarRocks instance with built-in Private Zone instead of Server Load Balancer (SLB) for load balancing.

Syntax

source:
  ...

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030
  load-url: 127.0.0.1:8030
  username: root
  password: pass
  sink.buffer-flush.interval-ms: 5000  # Flush buffered data every 5 seconds.

Connector options

Required options

Optional options

Data type mappings

StarRocks does not support all CDC YAML data types. Writing data of an unsupported type causes the job to fail. Use the built-in CAST function in the transform component to convert unsupported types, or use a projection statement to exclude them from the sink table. For more information, see Data ingestion with Flink CDC.

Schema evolution

As a data ingestion sink, the StarRocks connector supports the following schema evolution events:

• CREATE TABLE: If the table already exists, the connector skips creation. Make sure the existing table schema is compatible with the source schema.

• ADD COLUMN: StarRocks requires primary key columns to appear first. Newly added columns must follow this restriction.

• MODIFY COLUMN TYPE: For supported type conversions, see the StarRocks ALTER TABLE documentation.

• DROP COLUMN

• TRUNCATE TABLE

• DROP TABLE