StarRocks 连接器提供了对 StarRocks 数据仓库的读写能力，支持做数据源表、结果表和维表。

使用限制

• StarRocks 连接器目前仅支持在 Flink 1.16-volcano 引擎版本中使用 。
• StarRocks 维表场景建议使用 JDBC 维表替换，效果是相同的，同时 JDBC 使用更广泛，更稳定，参考 JDBC Connector。

DDL 定义

CREATE TABLE starrocks_table(
 name VARCHAR,
 score BIGINT
 ) WITH (
 'connector' = 'starrocks',
 'jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port',
 'load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port',
 'database-name' = 'doc_db',
 'table-name' = 'table1',
 'username' = 'flinkuser',
 'password' = 'flinkpw'
 );

WITH 参数

通用参数

源表参数

结果表参数

维表参数

注意：StarRocks 维表场景建议都使用 JDBC 维表替换，效果是相同的，同时 JDBC 使用更广泛，更稳定，参考 JDBC Connector。

数据类型映射

• 源表

  StarRocks字段类型	Flink字段类型
  NULL	NULL
  BOOLEAN	BOOLEAN
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  INT	INT
  BIGINT	BIGINT
  LARGEINT	STRING
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DATE	DATE
  DATETIME	TIMESTAMP
  DECIMAL	DECIMAL
  DECIMALV2	DECIMAL
  DECIMAL32	DECIMAL
  DECIMAL64	DECIMAL
  DECIMAL128	DECIMAL
  CHAR	CHAR
  VARCHAR	STRING

• 结果表

  Flink 字段类型	StarRocks 字段类型
  BOOLEAN	BOOLEAN
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  INTEGER	INTEGER
  BIGINT	BIGINT
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DECIMAL	DECIMAL
  BINARY	INT
  CHAR	STRING
  VARCHAR	STRING
  STRING	STRING
  DATE	DATE
  TIMESTAMP_WITHOUT_TIME_ZONE(N)	DATETIME
  TIMESTAMP_WITH_LOCAL_TIME_ZONE(N)	DATETIME
  ARRAY	ARRAY
  MAP<KT,VT>	JSON STRING
  ROW	JSON STRING

示例代码

• 源表

  CREATE TABLE starrocks_source (
      id INTEGER,
      name STRING,
      score INTEGER,
      PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://172.28.*.*:9030', 
    'scan-url' = '172.28.*.*:8030', 
    'database-name' = 'doc_db',
    'table-name' = 'table1',
    'username' = 'docuser',
    'password' = 'docpw'
  );
  
  CREATE TABLE print_sink (
      id INTEGER,
      name STRING,
      score INTEGER
  ) WITH (
    'connector' = 'print'
  );
  
  INSERT INTO print_sink
  SELECT * FROM starrocks_source;
  

• 结果表

  CREATE TABLE datagen_source(
    id INTEGER,
    name STRING,
    score INTEGER
  ) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '1',
    'fields.name.length' = '9',
    'fields.id.min' = '1',
    'fields.id.max' = '1000',
    'fields.score.min' = '1',
    'fields.score.max' = '1000'
  );
  
  CREATE TABLE starrocks_sink(
    id INTEGER,
    name STRING,
    score INTEGER,
    PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://172.28.*.*:9030', 
    'load-url' = '172.28.*.*:8030', 
    'database-name' = 'doc_db',
    'table-name' = 'table1',
    'username' = 'docuser',
    'password' = 'docpw',
    'sink.parallelism' = '1'
  );
  
  
  INSERT INTO starrocks_sink
  SELECT * FROM datagen_source;
  

• 维表（使用 JDBC 维表的方式读取 SR 表）

  CREATE TABLE datagen_source (
      id1 INTEGER,
      name1 STRING,
      score1 INTEGER,
      proc_time as PROCTIME ()
    )
  WITH
    (
      'connector' = 'datagen',
      'rows-per-second' = '1',
      'fields.name1.length' = '9',
      'fields.id1.min' = '1',
      'fields.id1.max' = '1000',
      'fields.score1.min' = '1',
      'fields.score1.max' = '1000'
    );
  
  
  CREATE TABLE starrocks_lookup (
      id INTEGER,
      name STRING,
      score INTEGER,
      PRIMARY KEY (id) NOT ENFORCED
    )
  WITH
    (
      'connector' = 'jdbc',
      'url' = 'jdbc:mysql://192.*.*.*:3306/doc_db',     
      'table-name' = '<yourtablename>',
      'username' = 'admin',          
      'password' = 'MyPa$$w0rd',      
      'lookup.cache.max-rows' = '100',
      'lookup.max-retries' = '3'
    );
  
  
  CREATE TABLE print_sink (
      id1 INTEGER,
      name1 STRING,
      score1 INTEGER,
      id INTEGER,
      name STRING,
      score INTEGER
    )
  WITH
    ('connector' = 'print');
  
  
  INSERT INTO
    print_sink
  SELECT
    datagen_source.id1,
    datagen_source.name1,
    datagen_source.score1,
    starrocks_lookup.id,
    starrocks_lookup.name,
    starrocks_lookup.score
  FROM
    datagen_source
    LEFT JOIN starrocks_lookup FOR SYSTEM_TIME AS OF proc_time ON datagen_source.id1 = starrocks_lookup.id;