Flink Connector

Parameter description

Data type mapping

Obtain the Flink Connector JAR and upload it to your Flink cluster

1. You can obtain the Flink Connector JAR file using one of the following methods.

   Method 1: Direct download

   You can download different versions of the Flink Connector JAR file from the Maven Central Repository.

   Method 2: Maven dependency

   Add the Flink Connector as a dependency to the pom.xml file of your Maven project in the following format.

   • For Flink Connectors compatible with Flink 1.15 and later:

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

   • For Flink Connectors compatible with Flink versions earlier than 1.15:

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

   Method 3: Manual compilation

   1. Download the Flink Connector source code.

   2. Run the following command to compile the Flink Connector source code into a JAR file.

      sh build.sh <flink_version>

      For example, if your Flink version is 1.17, run the following command.

      sh build.sh 1.17

   3. After compilation, you can find the generated JAR file in the target/ directory.

      For example, the filename is typically in the format flink-connector-starrocks-1.2.7_flink-1.17-SNAPSHOT.jar.

      Note

      Unreleased Flink Connector versions include the SNAPSHOT suffix.

   The Flink Connector JAR filename follows these formats:

   • For Flink 1.15 and later: flink-connector-starrocks-${connector_version}_flink-${flink_version}.jar. For example, if you use Flink 1.17 and Flink Connector 1.2.8, the filename is flink-connector-starrocks-1.2.8_flink-1.17.jar.

   • For Flink versions earlier than 1.15: flink-connector-starrocks-${connector_version}_flink-${flink_version}_${scala_version}.jar. For example, if you use Flink 1.14, Scala 2.12, and Flink Connector 1.2.7, the filename is flink-connector-starrocks-1.2.7_flink-1.14_2.12.jar.

     Note

     Replace the following placeholders as needed:

     • flink_version: Your Flink version number.

     • scala_version: Your Scala version number.

     • connector_version: Your Flink Connector version number.

2. Upload the Flink Connector JAR file that you obtained to the flink-{flink_version}/lib directory of your Flink cluster.

   For example, if you use an EMR cluster of version EMR-5.19.0, place the JAR file in the /opt/apps/FLINK/flink-current/lib directory.

Start the Flink cluster

1. Log on to the master node of your Flink cluster. For more information, see Log on to a cluster.

2. Run the following command to start the Flink cluster.

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

Write data using Flink SQL

1. Create a database named test in StarRocks, and then create a primary key table named score_board in the database.

   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. Log on to the master node of your Flink cluster. For more information, see Log on to a cluster.

3. Run the following command to start Flink SQL.

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

4. Run the following command to create a table named score_board and insert data into it.

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

   If you are importing data into a StarRocks primary key table, you must explicitly specify the primary key in the Flink table DDL. For other types of StarRocks tables, such as Duplicate Key tables, specifying a primary key is optional.

Write data using Flink DataStream

Write a Flink DataStream job based on the type of input records.

• Write CSV-formatted string data

  If your input records are CSV-formatted strings, see the complete Flink DataStream job code in 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);

• Write JSON-formatted string data

  If your input records are JSON-formatted strings, see the complete Flink DataStream job code in 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);

• Write custom Java object data

  If your input records are custom Java objects, see the complete Flink DataStream job code in LoadCustomJavaRecords.

  • In this example, a simple POJO class RowData is defined to represent each record.

    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;
            }
        }
    

  • The main code is as follows.

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

    The RowDataTransformer is defined as follows.

    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();
        }
    }  

Synchronize data using Flink CDC 3.0

The Flink CDC 3.0 framework simplifies building streaming ELT pipelines from CDC data sources, such as MySQL or Kafka, to StarRocks. With this pipeline, you can:

• Automatically create databases and tables

• Synchronize full and incremental data

• Synchronize schema changes

Starting from StarRocks Flink Connector v1.2.9, the connector is integrated into the Flink CDC 3.0 framework and renamed StarRocks Pipeline Connector. This connector supports all the preceding features. We recommend that you use it with StarRocks v3.2.1 or later to leverage the `fast_schema_evolution` feature, which further accelerates column additions and deletions and reduces resource consumption.

Import into a primary key table

1. Create a database named test in StarRocks, and then create a primary key table named 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. Insert data into the StarRocks table.

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

3. Run the following command to start the Flink SQL client.

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

4. Update data.

   Partial update

   Partial update lets you update only specific columns, such as name, without affecting others, such as score.

   1. Create the score_board table in the Flink SQL client and enable partial updates.

      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: Enables partial updates.

      • sink.properties.columns: Specifies the columns to update. If the Flink Connector version is 1.2.7 or earlier, you must also set the sink.properties.columns option to id,name,__op to inform the Flink connector which columns to update. Note that you must append the __op field at the end. The __op field indicates whether the import is an UPSERT or DELETE operation, and its value is automatically set by the Flink connector.

   2. Insert updated data.

      Insert two rows that have the same primary keys as existing data but with modified values in the name column.

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

   3. Query the StarRocks table in the SQL Editor.

      SELECT * FROM `test`.`score_board`;

      You will see that only the values in the name column have changed, while the score column remains unchanged.

      The query returns two records: id=1, name=starrocks-update, score=100 and id=2, name=flink-update, score=100.

   Conditional update

   This example shows how to perform conditional updates based on the value of the score column. A row is updated only if the score value in the incoming data is greater than or equal to the current value in the StarRocks table.

   1. Create the score_board table in the Flink SQL client as follows.

      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: Set to score. This specifies that the Flink Connector uses the score column as the update condition during data writes.

      • sink.version: Set to V1. This specifies that the Flink Connector uses the Stream Load interface to import data.

   2. Insert two rows of data into the table in the Flink SQL client.

      The rows have the same primary keys as existing rows in the StarRocks table. The first row has a smaller score value, and the second row has a larger score value.

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

   3. Query the StarRocks table in the SQL Editor.

      SELECT * FROM `test`.`score_board`;

      You will see that only the second row has changed, while the first row remains unchanged.

      The query returns two records: the first row id=1, name=starrocks, score=100 (unchanged) and the second row id=2, name=flink-update, score=101 (updated).

Import into a Bitmap column

The Bitmap data type is commonly used to accelerate exact distinct count scenarios, such as calculating unique visitors (UV). The following example shows how to import data into a Bitmap column of a StarRocks table using Flink SQL and query the UV count in StarRocks.

1. Create a StarRocks aggregate table in the SQL Editor.

   Create an aggregate table named page_uv in the test database with the following specifications:

   • The visit_users column is defined as the BITMAP type and configured with the `BITMAP_UNION` aggregate function.

   • page_id and visit_date serve as aggregate keys (AGGREGATE KEY) for grouping and deduplication.

   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. Create a table in the Flink SQL client.

   Because Flink does not support the Bitmap type, you must implement column mapping and type conversion as follows:

   • In the Flink table, define the visit_user_id column as BIGINT to represent the visit_users column in the StarRocks table.

   • Use the sink.properties.columns configuration to convert the data in the visit_user_id column to the Bitmap type using the `to_bitmap` function.

   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. Insert data in the Flink SQL client.

   Insert multiple rows into the page_uv table to simulate different users accessing pages at different times.

   visit_user_id is of the BIGINT type. Flink automatically converts it to the Bitmap type.

   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. Query the UV count in the SQL Editor.

   Use the aggregation capability of StarRocks to calculate the unique visitor count (UV) per page with COUNT(DISTINCT visit_users).

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

   For page_id 2, count(DISTINCT visit_users) is 1. For page_id 1, count(DISTINCT visit_users) is 3.

Import into an HLL column

HyperLogLog (HLL) is a data type used for approximate distinct counting and is suitable for calculating unique visitors (UV) in large-scale data scenarios. The following example shows how to import data into an HLL column of a StarRocks table using Flink SQL and query the UV count in StarRocks.

1. Create a StarRocks aggregate table in the SQL Editor.

   Create an aggregate table named hll_uv in the test database with the following specifications:

   • The visit_users column is defined as the HLL type and configured with the `HLL_UNION` aggregate function.

   • page_id and visit_date serve as aggregate keys (AGGREGATE KEY) for grouping and deduplication.

   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. Create a table in the Flink SQL client.

   Because Flink does not support the HLL type, you must implement column mapping and type conversion as follows:

   • In the Flink table, define the visit_user_id column as BIGINT to represent the visit_users column in the StarRocks table.

   • Use the sink.properties.columns configuration to map columns and convert the BIGINT-type visit_user_id data to the HLL type using the `hll_hash` function.

   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. Insert data in the Flink SQL client.

   Insert multiple rows into the hll_uv table to simulate different users accessing pages at different times.

   visit_user_id is of the BIGINT type. Flink automatically converts it to the HLL type.

   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. Query the UV count in the SQL Editor.

   Use the aggregation capability of StarRocks to calculate the unique visitor count (UV) per page with COUNT(DISTINCT visit_users).

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

   The result is as follows.

     page_id  count(DISTINCT visit_users)
   0       4                            1
   1       3                            2