E-MapReduce:Flink コネクタ

Flink クラスターへの JAR のアップロード

Flink クラスター上の flink-{flink_version}/lib ディレクトリに JAR をアップロードします。

EMR-5.19.0 を実行している EMR クラスターの場合は、JAR を /opt/apps/FLINK/flink-current/lib に配置します。

パラメーター

データ型マッピング

フラッシュポリシー

コネクタはレコードをメモリ内にバッファリングし、Stream Load を介して StarRocks にフラッシュします。フラッシュがトリガーされるタイミングは、配信セマンティクスによって異なります。

`at-least-once` の場合、以下のいずれかの条件が満たされると、フラッシュがトリガーされます：

• バッファリングされたデータサイズが sink.buffer-flush.max-bytes に達した場合。

• バッファリングされた行数が sink.buffer-flush.max-rows に達した場合（シンクバージョン V1 のみ）。

• 最後のフラッシュからの経過時間が sink.buffer-flush.interval-ms に達した場合。

exactly-once の場合、メモリ内のデータは Flink のチェックポイントがトリガーされたときにのみフラッシュされます。この場合、sink.buffer-flush.max-bytes および sink.buffer-flush.interval-ms パラメーターは、しきい値に達してもデータが自動的にフラッシュされないため、効果がありません。

1 回限りのセマンティクス

Flink コネクタ 1.2.4 以降では、StarRocks 2.4 で導入された Stream Load トランザクションインターフェイスを用いて 1 回限りのセマンティクスが実装されています。このアプローチは、従来の非トランザクション方式と比較して、メモリ使用量が少なく、チェックポイントのオーバーヘッドも低減されます。

Flink コネクタ 1.2.8 以降で 1 回限りのセマンティクスを使用する場合は、sink.label-prefix を設定してください。コネクタはこのプレフィックスを使用して、チェックポイント中に Flink ジョブが失敗した際にオープン状態のトランザクションを特定およびクリーンアップします。

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 クラスターのマスターノードにログインします。詳細については、「クラスターへのログイン」をご参照ください。

3. Flink SQL クライアントを起動します。

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

4. StarRocks テーブルに対応する Flink テーブルを作成し、データを挿入します。

   対象が StarRocks のプライマリキーを持つテーブルの場合、Flink テーブルの DDL でプライマリキーを明示的に定義する必要があります（上記の例を参照）。Duplicate Key テーブルなどの他のテーブルタイプでは、プライマリキーの定義は任意です。

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

Flink DataStream API を使用したデータ書き込み

レコードの形式に応じて、以下のいずれかのアプローチを選択してください。すべての例では、Flink SQL の例で使用したのと同じ score_board テーブルに書き込みます。

CSV 文字列レコードの書き込み

レコードが CSV 形式の文字列である場合、sink.properties.format を csv に設定し、列区切り文字を指定します。GitHub 上の完全な例をご参照ください。

/**
 * CSV 形式のレコードを生成します。各レコードは "\t" で区切られた 3 つの値から構成されます。
 * これらの値は、StarRocks テーブルの列 `id`、`name`、`score` に対応します。
 */
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);

JSON 文字列レコードの書き込み

レコードが JSON 形式の文字列である場合、sink.properties.format を json に設定し、外側の配列を除去するために strip_outer_array を有効にします。GitHub 上の完全な例をご参照ください。

/**
 * JSON 形式のレコードを生成します。
 * 各レコードは、列 id、name、score に対応する 3 つのキーと値のペアから構成されます。
 */
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);

カスタム Java オブジェクトの書き込み

レコードがカスタム Java オブジェクトである場合、StarRocks テーブルと一致するスキーマを定義し、各オブジェクトを Object[] に変換する StarRocksSinkRowBuilder を実装します。GitHub 上の完全な例をご参照ください。

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

シンクの設定：

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

// StarRocks テーブルのレイアウトと一致するようにスキーマを定義します。
TableSchema schema = TableSchema.builder()
        .field("id", DataTypes.INT().notNull())  // プライマリキー列には notNull() が必要です
        .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);

トランスフォーマーの実装：

private static class RowDataTransformer implements StarRocksSinkRowBuilder<RowData> {

    /**
     * RowData の各フィールドを Object[] の対応する位置にマップします。
     * 配列のレイアウトは、上記で定義したスキーマと一致している必要があります。
     */
    @Override
    public void accept(Object[] internalRow, RowData rowData) {
        internalRow[0] = rowData.id;
        internalRow[1] = rowData.name;
        internalRow[2] = rowData.score;
        // プライマリキーを持つテーブルの場合、最後の要素を UPSERT または DELETE を示すために設定します。
        internalRow[internalRow.length - 1] = StarRocksSinkOP.UPSERT.ordinal();
    }
}

プライマリキーを持つテーブルへのインポート

プライマリキーを持つテーブルでは、細かい粒度での書き込み制御を実現するために、部分更新および条件付き更新がサポートされています。

StarRocks テーブルのセットアップ

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

Flink SQL クライアントを起動します。

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

Bitmap 列へのインポート

Bitmap 列は、ユニークビジター（UV）の計算など、正確な重複排除カウントを高速化します。Flink にはネイティブの Bitmap 型がないため、Flink テーブルでは BIGINT 型の列を使用し、to_bitmap() 関数で変換します。

1. SQL エディターで、Bitmap 列を持つ集計テーブルを作成します。

   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 および visit_date は集計キーです。visit_users には集計関数として BITMAP_UNION が使用されます。

2. 対応する Flink テーブルを作成し、sink.properties.columns を使用して visit_user_id（BIGINT）を 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 クライアントでデータを挿入します。

   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 エディターでページごとの UV カウントを照会します。

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

   

HLL 列へのインポート

HLL（HyperLogLog）列は、Bitmap よりも少ないストレージオーバーヘッドで、大規模な近似重複排除カウントをサポートします。設定方法は同様で、Flink では BIGINT 型の列を使用し、hll_hash() 関数で変換します。

1. SQL エディターで、HLL 列を持つ集計テーブルを作成します。

   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 テーブルを作成し、sink.properties.columns を使用して visit_user_id（BIGINT）を hll_hash() 関数で 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 クライアントでデータを挿入します。

   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 エディターでページごとの UV カウントを照会します。

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