Flink Connector

参数说明

数据类型映射

获取Flink Connector JAR并上传至Flink集群

1. 您可以通过以下方式获取Flink Connector JAR包。

   方式1：直接下载

   从Maven Central Repository获取不同版本的Flink Connector JAR文件。

   方式2：Maven依赖

   在Maven项目的pom.xml文件中，根据以下格式将Flink Connector添加为依赖项。

   • 适用于Flink 1.15版本及以后的Flink Connector。

     <dependency>
         <groupId>com.starrocks</groupId>
         <artifactId>flink-connector-starrocks</artifactId>
         <version>${connector_version}_flink-${flink_version}</version>
     </dependency>

   • 适用于Flink 1.15版本之前的Flink Connector。

     <dependency>
         <groupId>com.starrocks</groupId>
         <artifactId>flink-connector-starrocks</artifactId>
         <version>${connector_version}_flink-${flink_version}_${scala_version}</version>
     </dependency>

   方式3：手动编译

   1. 下载Flink Connector代码。

   2. 执行以下命令，将Flink Connector的源代码编译成一个JAR文件。

      sh build.sh <flink_version>

      例如，如果您的环境中的Flink版本为1.17，您需要执行以下命令。

      sh build.sh 1.17

   3. 编译完成后，在 target/ 目录下找到生成的JAR文件。

      例如，文件名通常为flink-connector-starrocks-1.2.7_flink-1.17-SNAPSHOT.jar格式。

      说明

      非正式发布的Flink Connector版本会带有SNAPSHOT后缀。

   Flink Connector JAR文件的命名格式如下：

   • 适用于Flink 1.15版本及以后的Flink Connector命名格式为flink-connector-starrocks-${connector_version}_flink-${flink_version}.jar。例如您安装了Flink 1.17，并且想要使用1.2.8版本的Flink Connector，则您可以使用flink-connector-starrocks-1.2.8_flink-1.17.jar。

   • 适用于Flink 1.15版本之前的Flink Connector命名格式为flink-connector-starrocks-${connector_version}_flink-${flink_version}_${scala_version}.jar。例如您安装了Flink 1.14和Scala 2.12，并且您想要使用1.2.7版本的Flink Connector，您可以使用flink-connector-starrocks-1.2.7_flink-1.14_2.12.jar。

     说明

     请根据实际情况替换以下信息：

     • flink_version：Flink的版本号。

     • scala_version：Scala的版本号。

     • connector_version：Flink Connector的版本号。

2. 将获取到的Flink Connector JAR文件上传至Flink集群的flink-{flink_version}/lib目录下。

   例如，如果您使用的是EMR集群，且集群版本为EMR-5.19.0，则JAR文件应放置于/opt/apps/FLINK/flink-current/lib目录下。

启动Flink集群

1. 登录Flink集群的Master节点，详情请参见登录集群。

2. 执行以下命令，启动Flink集群。

   /opt/apps/FLINK/flink-current/bin/start-cluster.sh

使用Flink SQL写入数据

1. 在StarRocks中创建一个名为test的数据库，并在其中创建一张名为score_board的主键表。

   CREATE DATABASE test;
   
   CREATE TABLE test.score_board(
       id int(11) NOT NULL COMMENT "",
       name varchar(65533) NULL DEFAULT "" COMMENT "",
       score int(11) NOT NULL DEFAULT "0" COMMENT ""
   )
   ENGINE=OLAP
   PRIMARY KEY(id)
   DISTRIBUTED BY HASH(id);

2. 登录Flink集群的Master节点，详情请参见登录集群。

3. 执行以下命令，启动Flink SQL。

   /opt/apps/FLINK/flink-current/bin/sql-client.sh

4. 执行以下命令，创建一个名为score_board的表，并向其中插入数据。

   CREATE TABLE `score_board` (
       `id` INT,
       `name` STRING,
       `score` INT,
       PRIMARY KEY (id) NOT ENFORCED
   ) WITH (
       'connector' = 'starrocks',
       'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
       'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
       'database-name' = 'test',
       'table-name' = 'score_board',
       'username' = 'admin',
       'password' = '<password>',
   );
   
   INSERT INTO `score_board` VALUES (1, 'starrocks', 100), (2, 'flink', 100);

   如果目标是将数据导入StarRocks主键表，则必须在Flink表的DDL中明确指定主键。对于其他类型的StarRocks表（如Duplicate Key表），定义主键则为可选项。

使用Flink DataStream写入数据

根据输入记录（input records）的不同类型，编写对应Flink DataStream作业。

• 写入CSV格式的字符串数据

  如果输入记录为CSV格式的字符串，对应的Flink DataStream作业的主要代码如下所示，完整代码请参见LoadCsvRecords。

  /**
   * Generate CSV-format records. Each record has three values separated by "\t". 
   * These values will be loaded to the columns `id`, `name`, and `score` in the StarRocks table.
   */
  String[] records = new String[]{
          "1\tstarrocks-csv\t100",
          "2\tflink-csv\t100"
  };
  DataStream<String> source = env.fromElements(records);
  
  /**
   * Configure the Flink connector with the required properties.
   * You also need to add properties "sink.properties.format" and "sink.properties.column_separator"
   * to tell the Flink connector the input records are CSV-format, and the column separator is "\t".
   * You can also use other column separators in the CSV-format records,
   * but remember to modify the "sink.properties.column_separator" correspondingly.
   */
  StarRocksSinkOptions options = StarRocksSinkOptions.builder()
          .withProperty("jdbc-url", jdbcUrl)
          .withProperty("load-url", loadUrl)
          .withProperty("database-name", "test")
          .withProperty("table-name", "score_board")
          .withProperty("username", "root")
          .withProperty("password", "")
          .withProperty("sink.properties.format", "csv")
          .withProperty("sink.properties.column_separator", "\t")
          .build();
  // Create the sink with the options.
  SinkFunction<String> starRockSink = StarRocksSink.sink(options);
  source.addSink(starRockSink);

• 写入JSON格式的字符串数据

  如果输入记录为JSON格式的字符串，对应的Flink DataStream作业的主要代码如下所示，完整代码请参见LoadJsonRecords。

  /**
   * Generate JSON-format records. 
   * Each record has three key-value pairs corresponding to the columns id, name, and score in the StarRocks table.
   */
  String[] records = new String[]{
          "{\"id\":1, \"name\":\"starrocks-json\", \"score\":100}",
          "{\"id\":2, \"name\":\"flink-json\", \"score\":100}",
  };
  DataStream<String> source = env.fromElements(records);
  
  /** 
   * Configure the Flink connector with the required properties.
   * You also need to add properties "sink.properties.format" and "sink.properties.strip_outer_array"
   * to tell the Flink connector the input records are JSON-format and to strip the outermost array structure. 
   */
  StarRocksSinkOptions options = StarRocksSinkOptions.builder()
          .withProperty("jdbc-url", jdbcUrl)
          .withProperty("load-url", loadUrl)
          .withProperty("database-name", "test")
          .withProperty("table-name", "score_board")
          .withProperty("username", "root")
          .withProperty("password", "")
          .withProperty("sink.properties.format", "json")
          .withProperty("sink.properties.strip_outer_array", "true")
          .build();
  // Create the sink with the options.
  SinkFunction<String> starRockSink = StarRocksSink.sink(options);
  source.addSink(starRockSink);

• 写入自定义Java对象数据

  如果输入记录为自定义的Java对象，对应的Flink DataStream作业的主要代码如下所示，完整代码请参见LoadCustomJavaRecords。

  • 本示例中，定义一个简单的POJO类 RowData，用于表示每条记录。

    public static class RowData {
            public int id;
            public String name;
            public int score;
      
            public RowData() {}
      
            public RowData(int id, String name, int score) {
                this.id = id;
                this.name = name;
                this.score = score;
            }
        }
    

  • 主要代码如下所示。

    // Generate records which use RowData as the container.
    RowData[] records = new RowData[]{
            new RowData(1, "starrocks-rowdata", 100),
            new RowData(2, "flink-rowdata", 100),
        };
    DataStream<RowData> source = env.fromElements(records);
    
    // Configure the Flink connector with the required properties.
    StarRocksSinkOptions options = StarRocksSinkOptions.builder()
            .withProperty("jdbc-url", jdbcUrl)
            .withProperty("load-url", loadUrl)
            .withProperty("database-name", "test")
            .withProperty("table-name", "score_board")
            .withProperty("username", "root")
            .withProperty("password", "")
            .build();
    
    /**
     * The Flink connector will use a Java object array (Object[]) to represent a row to be loaded into the StarRocks table,
     * and each element is the value for a column.
     * You need to define the schema of the Object[] which matches that of the StarRocks table.
     */
    TableSchema schema = TableSchema.builder()
            .field("id", DataTypes.INT().notNull())
            .field("name", DataTypes.STRING())
            .field("score", DataTypes.INT())
            // When the StarRocks table is a Primary Key table, you must specify notNull(), for example, DataTypes.INT().notNull(), for the primary key `id`.
            .primaryKey("id")
            .build();
    // Transform the RowData to the Object[] according to the schema.
    RowDataTransformer transformer = new RowDataTransformer();
    // Create the sink with the schema, options, and transformer.
    SinkFunction<RowData> starRockSink = StarRocksSink.sink(schema, options, transformer);
    source.addSink(starRockSink);
    

    其中RowDataTransformer定义如下所示。

    private static class RowDataTransformer implements StarRocksSinkRowBuilder<RowData> {
    
        /**
         * Set each element of the object array according to the input RowData.
         * The schema of the array matches that of the StarRocks table.
         */
        @Override
        public void accept(Object[] internalRow, RowData rowData) {
            internalRow[0] = rowData.id;
            internalRow[1] = rowData.name;
            internalRow[2] = rowData.score;
            // When the StarRocks table is a Primary Key table, you need to set the last element to indicate whether the data loading is an UPSERT or DELETE operation.
            internalRow[internalRow.length - 1] = StarRocksSinkOP.UPSERT.ordinal();
        }
    }  

使用Flink CDC 3.0同步数据

Flink CDC 3.0框架能够轻松构建从CDC数据源（如MySQL、Kafka）到StarRocks的流式ELT管道。通过该管道，您可以实现以下功能：

• 自动创建数据库和表

• 同步全量和增量数据

• 同步Schema Change

自StarRocks Flink Connector v1.2.9版本起，该连接器已整合至Flink CDC 3.0框架中，并被命名为StarRocks Pipeline Connector。该连接器具备上述所有功能，建议与StarRocks v3.2.1及以上版本配合使用，以充分利用fast_schema_evolution特性，进一步提升列的增减速度并降低资源消耗。

导入至主键表

1. 在StarRocks中创建一个名为test的数据库，并在其中创建一个主键表score_board。

   CREATE DATABASE `test`;
   
   CREATE TABLE `test`.`score_board`
   (
       `id` int(11) NOT NULL COMMENT "",
       `name` varchar(65533) NULL DEFAULT "" COMMENT "",
       `score` int(11) NOT NULL DEFAULT "0" COMMENT ""
   )
   ENGINE=OLAP
   PRIMARY KEY(`id`)
   COMMENT "OLAP"
   DISTRIBUTED BY HASH(`id`);

2. 向StarRocks表中插入数据。

   INSERT INTO `test`.`score_board` VALUES (1, 'starrocks', 100), (2, 'flink', 100);

3. 执行以下命令，启动Flink SQL客户端。

   /opt/apps/FLINK/flink-current/bin/sql-client.sh

4. 更新数据。

   部分更新

   部分更新允许您仅更新指定列（如name），而不影响其他列（如score）。

   1. 在Flink SQL客户端创建表score_board，并启用部分更新功能。

      CREATE TABLE `score_board` (
          `id` INT,
          `name` STRING,
          PRIMARY KEY (id) NOT ENFORCED
      ) WITH (
          'connector' = 'starrocks',
          'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
          'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
          'database-name' = 'test',
          'table-name' = 'score_board',
          'username' = 'admin',
          'password' = '<password>',
          'sink.properties.partial_update' = 'true',
          -- only for Flink connector version <= 1.2.7
          'sink.properties.columns' = 'id,name,__op'
      ); 

      • sink.properties.partial_update：启用部分更新。

      • sink.properties.columns：指定需要更新的列。如果 Flink Connector版本小于等于1.2.7，则还需要将选项sink.properties.columns设置为id,name,__op，以告诉 Flink connector 需要更新的列。请注意，您需要在末尾附加字段__op。字段__op表示导入是 UPSERT 还是 DELETE 操作，其值由 Flink connector 自动设置。

   2. 插入更新数据。

      插入两行数据，主键与现有数据相同，但name列的值被修改。

      INSERT INTO score_board VALUES (1, 'starrocks-update'), (2, 'flink-update');

   3. 在SQL Editor中查询StarRocks表。

      SELECT * FROM `test`.`score_board`;

      您会看到只有name列的值发生了变化，而score列保持不变。

      

   条件更新

   本示例展示如何根据score列的值进行条件更新。只有导入的数据行中score列值大于等于StarRocks表当前值时，该数据行才会更新。

   1. 在Flink SQL客户端按照以下方式创建表score_board。

      CREATE TABLE `score_board` (
          `id` INT,
          `name` STRING,
          `score` INT,
          PRIMARY KEY (id) NOT ENFORCED
      ) WITH (
          'connector' = 'starrocks',
          'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
          'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
          'database-name' = 'test',
          'table-name' = 'score_board',
          'username' = 'admin',
          'password' = '<password>',
          'sink.properties.merge_condition' = 'score',
          'sink.version' = 'V1'
      );

      • sink.properties.merge_condition：设置为score，表示在数据写入时，Flink Connector会以 score 列作为更新条件。

      • sink.version：设置为V1，表示Flink Connector使用Stream Load接口导入数据。

   2. 在Flink SQL客户端插入两行数据到表中。

      数据行的主键与StarRocks表中的行相同。第一行数据score列中具有较小的值，而第二行数据score列中具有较大的值。

      INSERT INTO `score_board` VALUES (1, 'starrocks-update', 99), (2, 'flink-update', 101);

   3. 在SQL Editor中查询StarRocks表。

      SELECT * FROM `test`.`score_board`;

      您会看到只有第二行数据发生了变化，而第一行数据未发生变化。

      

导入至Bitmap列

Bitmap类型常用于加速精确去重计数场景，例如计算独立访客数（UV）。以下是一个完整的示例，展示如何通过Flink SQL将数据导入到StarRocks表的Bitmap列中，并在StarRocks中查询UV数。

1. 在SQL Editor中创建StarRocks聚合表。

   在数据库test中创建一个聚合表page_uv，其中：

   • visit_users列被定义为 BITMAP 类型，并配置聚合函数BITMAP_UNION。

   • page_id和visit_date作为聚合键（AGGREGATE KEY），用于分组和去重。

   CREATE TABLE `test`.`page_uv` (
     `page_id` INT NOT NULL COMMENT 'page ID',
     `visit_date` datetime NOT NULL COMMENT 'access time',
     `visit_users` BITMAP BITMAP_UNION NOT NULL COMMENT 'user ID'
   ) ENGINE=OLAP
   AGGREGATE KEY(`page_id`, `visit_date`)
   DISTRIBUTED BY HASH(`page_id`);

2. 在Flink SQL客户端中创建表。

   由于Flink不支持Bitmap类型，需要通过以下方式实现列映射和类型转换：

   • 在Flink表中，将visit_user_id列定义为BIGINT类型，以代表StarRocks表中的visit_users列。

   • 使用sink.properties.columns配置，将visit_user_id列的数据通过to_bitmap函数转换为Bitmap类型。

   CREATE TABLE `page_uv` (
       `page_id` INT,
       `visit_date` TIMESTAMP,
       `visit_user_id` BIGINT
   ) WITH (
       'connector' = 'starrocks',
       'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
       'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
       'database-name' = 'test',
       'table-name' = 'page_uv',
       'username' = 'admin',
       'password' = '<password>',
       'sink.properties.columns' = 'page_id,visit_date,visit_user_id,visit_users=to_bitmap(visit_user_id)'
   );

3. 在Flink SQL客户端中插入数据。

   向表page_uv插入多行数据，模拟不同用户在不同时间访问页面的行为。

   visit_user_id是BIGINT类型，Flink会将其自动转换为Bitmap类型。

   INSERT INTO `page_uv` VALUES
      (1, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 13),
      (1, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 23),
      (1, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 33),
      (1, CAST('2020-06-23 02:30:30' AS TIMESTAMP), 13),
      (2, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 23);

4. 在SQL Editor中查询UV数。

   通过StarRocks的聚合能力，使用COUNT(DISTINCT visit_users)计算每个页面的独立访客数（UV）。

   SELECT page_id, COUNT(DISTINCT visit_users) FROM page_uv GROUP BY page_id;

   返回结果如下所示。

   

导入至HLL列

HLL（HyperLogLog）是一种用于近似去重计数的数据类型，适合处理大规模数据的独立访客数（UV）计算。以下是一个完整的示例，展示如何通过Flink SQL将数据导入到StarRocks表的HL 列中，并在StarRocks中查询UV数。

1. 在SQL Editor中创建StarRocks聚合表。

   在数据库test中创建一个聚合表hll_uv，其中：

   • visit_users列被定义为HLL类型，并配置聚合函数HLL_UNION。

   • page_id和visit_date作为聚合键（AGGREGATE KEY），用于分组和去重。

   CREATE TABLE `test`.`hll_uv` (
     `page_id` INT NOT NULL COMMENT 'page ID',
     `visit_date` DATETIME NOT NULL COMMENT 'access time',
     `visit_users` HLL HLL_UNION NOT NULL COMMENT 'user ID'
   ) ENGINE=OLAP
   AGGREGATE KEY(`page_id`, `visit_date`)
   DISTRIBUTED BY HASH(`page_id`);

2. 在Flink SQL客户端中创建表。

   由于Flink不支持HLL类型，需要通过以下方式实现列映射和类型转换：

   • 在Flink表中，将visit_user_id列定义为BIGINT类型，以代表StarRocks表中的visit_users列。

   • 使用sink.properties.columns配置列映射，并通过hll_hash函数将BIGINT类型的visit_user_id数据转换为HLL类型。

   CREATE TABLE `hll_uv` (
       `page_id` INT,
       `visit_date` TIMESTAMP,
       `visit_user_id` BIGINT
   ) WITH (
       'connector' = 'starrocks',
       'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
       'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
       'database-name' = 'test',
       'table-name' = 'hll_uv',
       'username' = 'admin',
       'password' = '<password>',
       'sink.properties.columns' = 'page_id,visit_date,visit_user_id,visit_users=hll_hash(visit_user_id)'
   );

3. 在Flink SQL客户端中插入数据。

   向表hll_uv插入多行数据，模拟不同用户在不同时间访问页面的行为。

   visit_user_id是BIGINT类型，Flink会将其自动转换为HLL类型。

   INSERT INTO `hll_uv` VALUES
      (3, CAST('2023-07-24 12:00:00' AS TIMESTAMP), 78),
      (4, CAST('2023-07-24 13:20:10' AS TIMESTAMP), 2),
      (3, CAST('2023-07-24 12:30:00' AS TIMESTAMP), 674);

4. 在SQL Editor中查询UV数。

   通过StarRocks的聚合能力，使用COUNT(DISTINCT visit_users)计算每个页面的独立访客数（UV）。

   SELECT `page_id`, COUNT(DISTINCT `visit_users`) FROM `hll_uv` GROUP BY `page_id`;

   返回结果如下所示。