StarRocks YAML connector

Configuration

Reuse a built-in catalog

VVR 11.5 and later lets you reference a built-in StarRocks catalog created on the Data Management page directly in a Flink CDC data ingestion job. This simplifies configuration by reducing the number of properties you need to set manually.

sink:
  type: starrocks
  using.built-in-catalog: starrocks_catalog

Data ingestion jobs can automatically reuse the following StarRocks catalog options:

• jdbc-url

• http-url

• username

• password

• table.num-buckets

To override these values, you can explicitly set the corresponding YAML options, which take precedence.

Supported events

• CREATE TABLE EVENT

  Note

  If the downstream StarRocks table already exists, the connector does not attempt to create it again. Ensure that the downstream table schema is compatible with the upstream schema.

• ADD COLUMN EVENT

  Note

  StarRocks requires primary key columns to appear first in a table. Any new columns must be added after them.

• ALTER COLUMN TYPE EVENT

  Note

  For supported schema change paths, refer to the official StarRocks documentation.

• RENAME COLUMN EVENT

  Note

  Requires Ververica Runtime (VVR) 11.9 or later, and StarRocks 3.3.2 or later.

• DROP COLUMN EVENT

• TRUNCATE TABLE EVENT

• DROP TABLE EVENT

LENIENT (default)

In LENIENT mode, schema changes are handled as follows:

• Add a nullable column: The corresponding column is automatically added to the end of the sink table schema, and the data of the new column is synchronized automatically.

• Add a non-nullable column: A corresponding column is automatically added to the end of the sink table schema and is set to nullable. For data that existed before the column was added, the value is automatically set to NULL.

• Drop a column: The column is not dropped from the sink table. If the original column was non-nullable, it is changed to nullable. If it was already nullable, no change is made.

• Rename a column: This operation is treated as adding a new column while keeping the old one. The renamed nullable column is added to the end of the sink table. If the original column was non-nullable, it is changed to nullable.

• Reorder columns: Ignored. The change is not synchronized to the downstream table.

• Change a column type: Not automatically ignored by LENIENT mode. The change is still passed to StarRocks, and only compatible change paths are supported. For the allowed paths, see ALTER TABLE. For VVR 11.1 and later, the supported scope remains the same as that described in the existing documentation.

• Drop a table or truncate a table: Excluded by default. The change is not synchronized to the downstream.

EVOLVE

In EVOLVE mode, schema changes are handled as follows:

• Add a column: Supported. The StarRocks executor ignores the upstream insertion position and appends the column to the end of the existing columns.

• Drop a column: Supported. The column is actually dropped from the sink table.

• Rename a column: Supported (requires StarRocks 3.3.2 or later). The name of the downstream column is changed.

• Change a column type: Supported. Only compatible change paths are supported.

• Reorder columns: Supported. The primary key columns must still come first and keep their original order.

• Drop a table or truncate a table: Actually executed when the events are not excluded.

Synchronize a single table

Synchronize a single MySQL table to StarRocks. If the destination database and table do not exist, the connector automatically creates a primary key table.

pipeline:
          name: MySQL to StarRocks Pipeline
      source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.test_source_table
        server-id: 5401-5499
        # (Optional) Synchronize data from newly added tables during the incremental phase without restarting the job.
        scan.binlog.newly-added-table.enabled: true
        # (Optional) Synchronize table and column comments to the destination.
        include-comments.enabled: true
        # (Optional) Deserialize binlogs only for captured tables to improve read performance.
        scan.only.deserialize.captured.tables.changelog.enabled: true
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
       
        # (Optional) For jobs with a small amount of data, reduce the flush interval to prevent long persistence delays. Default: 300000, or 5 minutes.
        sink.buffer-flush.interval-ms: 5000
        # (Optional) If the upstream character set is utf8mb4, set this parameter to 4 to prevent text truncation. Default: 3.
        unicode-char.max-bytes: 4
        # (Optional) The number of buckets for automatically created tables. This parameter is required for StarRocks versions earlier than 2.5.7. Later versions can infer the value automatically.
        table.create.num-buckets: 8
        # (Optional) The number of replicas for automatically created tables. Configure this value based on your cluster.
        table.create.properties.replication_num: 3
        # (Optional) For StarRocks 3.2 and later, enable this option to accelerate schema changes.
        table.create.properties.fast_schema_evolution: true
        # Note: If you use a transform to change the primary key, you must also set sink.ignore.update-before: false.
        # Otherwise, rows associated with the old primary key remain in the destination.
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Synchronize an entire database

Synchronize all tables in a MySQL database to StarRocks at the same time. The connector automatically creates the destination database and primary key tables, so you do not need to create each table in advance.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        # Use a regular expression to match all tables in the database. To match multiple databases, separate patterns with commas.
        tables: test_db.\.*
        server-id: 5401-5499
        # (Optional) Synchronize data from newly added tables during the incremental phase without restarting the job.
        scan.binlog.newly-added-table.enabled: true
        # (Optional) Synchronize table and column comments to the destination.
        include-comments.enabled: true
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Optional) For jobs with a small amount of data, reduce the flush interval to prevent long persistence delays. Default: 300000, or 5 minutes.
        sink.buffer-flush.interval-ms: 5000
        # (Optional) If the upstream character set is utf8mb4, set this parameter to 4 to prevent text truncation. Default: 3.
        unicode-char.max-bytes: 4
        # (Optional) The number of buckets for automatically created tables. This parameter is required for StarRocks versions earlier than 2.5.7. Later versions can infer the value automatically.
        table.create.num-buckets: 8
        # (Optional) The number of replicas for automatically created tables. Configure this value based on your cluster.
        table.create.properties.replication_num: 3
        # (Optional) For StarRocks 3.2 and later, enable this option to accelerate schema changes.
        table.create.properties.fast_schema_evolution: true
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Exclude specific tables during full database synchronization

When synchronizing an entire database, use a regular expression to skip tables that you do not want to synchronize to the destination, such as temporary or sensitive tables.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.\.*
        # Tables that match this regular expression are not synchronized.
        tables.exclude: test_db.tmp_.\*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Optional) The load interface version. V2 requires StarRocks 2.4 or later. For EMR Serverless StarRocks, use V1 if streaming issues occur.
        sink.version: V2
        # (Optional) For jobs with a small amount of data, reduce the flush interval to prevent long persistence delays. Default: 300000, or 5 minutes.
        sink.buffer-flush.interval-ms: 5000
        # (Optional) The number of buckets for automatically created tables. This parameter is required for StarRocks versions earlier than 2.5.7.
        table.create.num-buckets: 8
        # (Optional) For StarRocks 3.2 and later, enable this option to accelerate schema changes.
        table.create.properties.fast_schema_evolution: true
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Synchronize to specified databases and tables

If the destination StarRocks database or table name must differ from the upstream name, such as when writing to an ODS-layer database, use a route to rename it.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.\.*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Optional) For jobs with a small amount of data, reduce the flush interval to prevent long persistence delays. Default: 300000, or 5 minutes.
        sink.buffer-flush.interval-ms: 5000
        # (Optional) The number of buckets for automatically created tables. This parameter is required for StarRocks versions earlier than 2.5.7.
        table.create.num-buckets: 8
      
      route:
        # Synchronize all tables in the MySQL test_db database to the StarRocks test_db2 database without changing table names.
        # <> is a placeholder that is replaced by the matched source table name.
        - source-table: test_db.\.*
          sink-table: test_db2.<>
          replace-symbol: <>
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Merge sharded tables

Merge multiple sharded tables with identical schemas into a single StarRocks table for unified queries and analytics.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        # Match all sharded tables, such as user_0 and user_1.
        tables: test_db.user\.*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Optional) For jobs with a small amount of data, reduce the flush interval to prevent long persistence delays. Default: 300000, or 5 minutes.
        sink.buffer-flush.interval-ms: 5000
        # (Optional) For merged tables, explicitly specify the number of buckets based on the total data volume.
        table.create.num-buckets: 8
      
      route:
        # Merge all sharded tables into the StarRocks test_db.user table.
        - source-table: test_db.user\.*
          sink-table: test_db.user
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Enable EVOLVE mode

By default, LENIENT mode does not synchronize schema changes such as dropping a column, dropping a table, or truncating a table to the destination. If you require strict schema synchronization, enable EVOLVE mode. This mode has significant limitations, so review the following notes before you use it.

Limitations and usage notes

• Column renaming is not supported. The job fails if an upstream column rename event occurs.

• Dropping a column or table, or truncating a table, is applied to the destination. An accidental upstream operation directly affects the destination table. The default LENIENT mode is safer because it does not synchronize drop-table or truncate-table events.

• If you restart the job without state and do not delete the sink table, an upstream and sink schema mismatch may cause the job to fail. You must manually adjust the downstream table schema.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.test_source_table
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
      
      pipeline:
        name: MySQL to StarRocks Pipeline
        # Enable EVOLVE mode to strictly synchronize schema changes. The job fails on unsupported changes, such as column renaming.
        schema.change.behavior: evolve