E-MapReduce:Flink Connector

Upload the JAR to your Flink cluster

Upload the JAR to the flink-{flink_version}/lib directory on the Flink cluster.

For EMR clusters running EMR-5.19.0, place the JAR in /opt/apps/FLINK/flink-current/lib.

Parameters

Data type mapping

Flush policy

The connector buffers records in memory and flushes them to StarRocks via Stream Load. When a flush is triggered depends on the delivery semantic:

For `at-least-once`, a flush is triggered when any of the following conditions are met:

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

• The number of buffered rows reaches sink.buffer-flush.max-rows (sink version V1 only).

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

For `exactly-once`, data in memory is flushed only when a Flink checkpoint is triggered. In this case, the sink.buffer-flush.max-bytes and sink.buffer-flush.interval-ms parameters have no effect because data is not automatically flushed when the threshold is reached.

Exactly-once semantics

From Flink Connector 1.2.4, exactly-once is implemented using the Stream Load transaction interface introduced in StarRocks 2.4. This approach uses less memory and reduces checkpoint overhead compared to the earlier non-transactional implementation.

Set sink.label-prefix when using exactly-once with Flink Connector 1.2.8 or later. The connector uses this prefix to identify and clean up transactions left open when a Flink job fails during a checkpoint.

Write data using Flink SQL

1. Create a database named test in StarRocks and create a primary key table 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. Log on to the master node of the Flink cluster. See Log on to a cluster.

3. Start the Flink SQL client:

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

4. Create a Flink table mapped to the StarRocks table and insert data:

   When the target is a StarRocks primary key table, you must explicitly define the primary key in the Flink table DDL (as shown above). For other table types such as Duplicate Key tables, the primary key definition is optional.

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

Write data using the Flink DataStream API

Choose the approach based on your record format. All examples write to the same score_board table used in the Flink SQL example.

Write CSV string records

If your records are CSV-formatted strings, set sink.properties.format to csv and specify the column separator. See the full example on GitHub.

/**
 * Generate CSV-format records. Each record has three values separated by "\t".
 * These values map 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);

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

SinkFunction<String> starRockSink = StarRocksSink.sink(options);
source.addSink(starRockSink);

Write JSON string records

If your records are JSON-formatted strings, set sink.properties.format to json and enable strip_outer_array to strip the outermost array. See the full example on GitHub.

/**
 * Generate JSON-format records.
 * Each record has three key-value pairs mapping to columns id, name, and score.
 */
String[] records = new String[]{
        "{\"id\":1, \"name\":\"starrocks-json\", \"score\":100}",
        "{\"id\":2, \"name\":\"flink-json\", \"score\":100}"
};
DataStream<String> source = env.fromElements(records);

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

SinkFunction<String> starRockSink = StarRocksSink.sink(options);
source.addSink(starRockSink);

Write custom Java objects

If your records are custom Java objects, define a schema that matches the StarRocks table and implement a StarRocksSinkRowBuilder to convert each object to an Object[]. See the full example on GitHub.

Define the POJO:

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

Set up the sink:

RowData[] records = new RowData[]{
        new RowData(1, "starrocks-rowdata", 100),
        new RowData(2, "flink-rowdata", 100)
};
DataStream<RowData> source = env.fromElements(records);

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

// Define the schema to match the StarRocks table layout.
TableSchema schema = TableSchema.builder()
        .field("id", DataTypes.INT().notNull())  // notNull() required for primary key columns
        .field("name", DataTypes.STRING())
        .field("score", DataTypes.INT())
        .primaryKey("id")
        .build();

RowDataTransformer transformer = new RowDataTransformer();
SinkFunction<RowData> starRockSink = StarRocksSink.sink(schema, options, transformer);
source.addSink(starRockSink);

Implement the transformer:

private static class RowDataTransformer implements StarRocksSinkRowBuilder<RowData> {

    /**
     * Map each field of RowData to the corresponding position in the Object[].
     * The array layout must match the schema defined above.
     */
    @Override
    public void accept(Object[] internalRow, RowData rowData) {
        internalRow[0] = rowData.id;
        internalRow[1] = rowData.name;
        internalRow[2] = rowData.score;
        // For primary key tables, set the last element to indicate UPSERT or DELETE.
        internalRow[internalRow.length - 1] = StarRocksSinkOP.UPSERT.ordinal();
    }
}

Import to primary key tables

Primary key tables support partial updates and conditional updates for fine-grained write control.

Set up the StarRocks table

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

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

Start the Flink SQL client:

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

Import to Bitmap columns

Bitmap columns accelerate exact deduplication counts, such as calculating unique visitors (UV). Because Flink does not have a native Bitmap type, use a BIGINT column in the Flink table and convert it with the to_bitmap() function.

1. Create an Aggregate table with a Bitmap column in SQL Editor:

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

   page_id and visit_date are the Aggregate Keys. visit_users uses BITMAP_UNION as the aggregate function.

2. Create the corresponding Flink table, using sink.properties.columns to convert visit_user_id (BIGINT) to Bitmap via to_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. Insert data in the Flink SQL client:

   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 per page in SQL Editor:

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

   

Import to HLL columns

HLL (HyperLogLog) columns support approximate deduplication counting at scale with lower storage overhead than Bitmap. The setup is similar: use a BIGINT column in Flink and convert it with hll_hash().

1. Create an Aggregate table with an HLL column in SQL Editor:

   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 the corresponding Flink table, using sink.properties.columns to convert visit_user_id (BIGINT) to HLL via hll_hash():

   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 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 per page in SQL Editor:

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