E-MapReduce:Flink コネクタ

パラメーターの説明

データ型マッピング

Flink コネクタ JAR ファイルの取得と Flink クラスターへのアップロード

1. Flink コネクタ JAR ファイルは、次のいずれかの方法で取得できます：

   方法 1：直接ダウンロード

   Maven 中央リポジトリ から、さまざまなバージョンの Flink コネクタ JAR ファイルをダウンロードできます。

   方法 2：Maven 依存関係

   次の形式で、Maven プロジェクトの pom.xml ファイルに Flink コネクタを依存関係として追加します：

   • Flink 1.15 以降と互換性のある Flink コネクタの場合：

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

   • 1.15 より前の Flink バージョンと互換性のある Flink コネクタの場合：

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

   方法 3：手動コンパイル

   1. Flink コネクタのソースコード をダウンロードします。

   2. 次のコマンドを実行して、Flink コネクタのソースコードを JAR ファイルにコンパイルします：

      sh build.sh <flink_version>

      たとえば、Flink バージョンが 1.17 の場合は、次のコマンドを実行します：

      sh build.sh 1.17

   3. コンパイル後、生成された JAR ファイルは target/ ディレクトリにあります。

      たとえば、ファイル名は通常 flink-connector-starrocks-1.2.7_flink-1.17-SNAPSHOT.jar のような形式です。

      説明

      未リリースの Flink コネクタのバージョンには、SNAPSHOT サフィックスが含まれます。

   Flink コネクタ JAR ファイル名は、次の形式です：

   • Flink 1.15 以降の場合： flink-connector-starrocks-${connector_version}_flink-${flink_version}.jar 。 たとえば、Flink 1.17 と Flink コネクタ 1.2.8 を使用する場合、ファイル名は flink-connector-starrocks-1.2.8_flink-1.17.jar です。

   • 1.15 より前の Flink バージョンの場合： flink-connector-starrocks-${connector_version}_flink-${flink_version}_${scala_version}.jar 。 たとえば、Flink 1.14、Scala 2.12、Flink コネクタ 1.2.7 を使用する場合、ファイル名は flink-connector-starrocks-1.2.7_flink-1.14_2.12.jar です。

     説明

     必要に応じて、次のプレースホルダーを置き換えてください：

     • flink_version ：お使いの Flink のバージョン番号。

     • scala_version ：お使いの Scala のバージョン番号。

     • connector_version ：お使いの Flink コネクタのバージョン番号。

2. 取得した Flink コネクタ JAR ファイルを、お使いの Flink クラスターの flink-{flink_version}/lib ディレクトリにアップロードします。

   たとえば、バージョン EMR-5.19.0 の EMR クラスターを使用する場合、JAR ファイルを /opt/apps/FLINK/flink-current/lib ディレクトリに配置します。

Flink クラスターの起動

1. お使いの Flink クラスターのマスターノードにログインします。 詳細については、「クラスターへのログイン」をご参照ください。

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

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 でプライマリキーを明示的に指定する必要があります。Duplicate Key テーブルなど、他のタイプの StarRocks テーブルの場合、プライマリキーの指定は任意です。

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

入力レコードのタイプに基づいて、Flink DataStream ジョブを作成します。

• CSV 形式の文字列データの書き込み

  入力レコードが CSV 形式の文字列の場合は、「LoadCsvRecords」で完全な Flink DataStream ジョブコードをご参照ください。

  /**
   * 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);
  /**
   * 必要なプロパティを使用して Flink コネクタを設定します。
   * また、プロパティ "sink.properties.format" と "sink.properties.column_separator" を追加して、
   * 入力レコードが CSV 形式であり、列区切り文字が "\t" であることを Flink コネクタに指定する必要があります。
   * CSV 形式のレコードで他の列区切り文字を使用することもできますが、
   * その場合は "sink.properties.column_separator" を適宜変更してください。
   */
  StarRocksSinkOptions options = StarRocksSinkOptions.builder()
          .withProperty("jdbc-url", jdbcUrl)
          .withProperty("load-url", loadUrl)
          .withProperty("database-name", "test")
          .withProperty("table-name", "score_board")
          .withProperty("username", "admin")
          .withProperty("password", "<password>")
          .withProperty("sink.properties.format", "csv")
          .withProperty("sink.properties.column_separator", "\t")
          .build();
  // オプションを使用してシンクを作成します。
  SinkFunction<String> starRockSink = StarRocksSink.sink(options);
  source.addSink(starRockSink);

• JSON 形式の文字列データの書き込み

  入力レコードが JSON 形式の文字列の場合は、「LoadJsonRecords」で完全な Flink DataStream ジョブコードをご参照ください。

  /**
   * JSON 形式のレコードを生成します。
   * 各レコードには、StarRocks テーブルの 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);
  /** 
   * 必要なプロパティを使用して Flink コネクタを設定します。
   * また、プロパティ "sink.properties.format" と "sink.properties.strip_outer_array" を追加して、
   * 入力レコードが JSON 形式であり、最も外側の配列構造を削除することを Flink コネクタに指定する必要があります。
   */
  StarRocksSinkOptions options = StarRocksSinkOptions.builder()
          .withProperty("jdbc-url", jdbcUrl)
          .withProperty("load-url", loadUrl)
          .withProperty("database-name", "test")
          .withProperty("table-name", "score_board")
          .withProperty("username", "admin")
          .withProperty("password", "<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 オブジェクトの場合は、「LoadCustomJavaRecords」で完全な Flink DataStream ジョブコードをご参照ください。

  • この例では、各レコードを表すために、シンプルな 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;
            }
        }
    

  • メインコードは次のとおりです。

    // RowData をコンテナとして使用するレコードを生成します。
    RowData[] records = new RowData[]{
            new RowData(1, "starrocks-rowdata", 100),
            new RowData(2, "flink-rowdata", 100),
        };
    DataStream<RowData> source = env.fromElements(records);
    // 必要なプロパティを使用して Flink コネクタを設定します。
    StarRocksSinkOptions options = StarRocksSinkOptions.builder()
            .withProperty("jdbc-url", jdbcUrl)
            .withProperty("load-url", loadUrl)
            .withProperty("database-name", "test")
            .withProperty("table-name", "score_board")
            .withProperty("username", "admin")
            .withProperty("password", "<password>")
            .build();
    /**
     * Flink コネクタは、StarRocks テーブルにロードされる行を表すために Java オブジェクト配列 (Object[]) を使用します。
     * 各要素は列の値です。
     * StarRocks テーブルのスキーマと一致する Object[] のスキーマを定義する必要があります。
     */
    TableSchema schema = TableSchema.builder()
            .field("id", DataTypes.INT().notNull())
            .field("name", DataTypes.STRING())
            .field("score", DataTypes.INT())
            // StarRocks テーブルがプライマリキーテーブルの場合、プライマリキー `id` に notNull()、例えば DataTypes.INT().notNull() を指定する必要があります。
            .primaryKey("id")
            .build();
    // スキーマに従って RowData を Object[] に変換します。
    RowDataTransformer transformer = new RowDataTransformer();
    // スキーマ、オプション、トランスフォーマーを使用してシンクを作成します。
    SinkFunction<RowData> starRockSink = StarRocksSink.sink(schema, options, transformer);
    source.addSink(starRockSink);
    

    RowDataTransformer は次のように定義されています。

    private static class RowDataTransformer implements StarRocksSinkRowBuilder<RowData> {
        /**
         * 入力 RowData に従ってオブジェクト配列の各要素を設定します。
         * 配列のスキーマは StarRocks テーブルのスキーマと一致します。
         */
        @Override
        public void accept(Object[] internalRow, RowData rowData) {
            internalRow[0] = rowData.id;
            internalRow[1] = rowData.name;
            internalRow[2] = rowData.score;
            // StarRocks テーブルがプライマリキーテーブルの場合、データロードがアップサートまたは DELETE 操作であるかを示すために、最後の要素を設定する必要があります。
            internalRow[internalRow.length - 1] = StarRocksSinkOP.UPSERT.ordinal();
        }
    }  

Flink CDC 3.0 を使用したデータの同期

Flink CDC 3.0 フレームワークは、MySQL や Kafka などの CDC データソースから StarRocks へのストリーミング ELT パイプラインの構築を簡素化します。このパイプラインを使用すると、次のことが可能になります：

• データベースとテーブルの自動作成

• 全量データと増分データの同期

• スキーマ変更の同期

StarRocks Flink コネクタ v1.2.9 以降、このコネクタは Flink CDC 3.0 フレームワークに統合され、StarRocks Pipeline Connector に名前が変更されました。このコネクタは、前述のすべての機能をサポートしています。fast_schema_evolution 機能を活用して列の追加と削除をさらに高速化し、リソース消費を削減するために、StarRocks v3.2.1 以降と併用することを推奨します。

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

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. データを更新します。

   部分更新

   部分更新では、score などの他の列に影響を与えることなく、name などの特定の列のみを更新できます。

   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',
          -- Flink コネクタのバージョンが 1.2.7 以下の場合のみ
          'sink.properties.columns' = 'id,name,__op'
      ); 

      • sink.properties.partial_update：部分更新を有効にします。

      • sink.properties.columns：更新する列を指定します。Flink コネクタのバージョンが 1.2.7 以前の場合は、Flink コネクタに更新対象の列を通知するため、sink.properties.columns オプションも id,name,__op に設定する必要があります。末尾に __op フィールドを追加する必要がある点に注意してください。__op フィールドは、インポートがアップサートまたは削除のどちらであるかを示し、その値は Flink コネクタによって自動的に設定されます。

   2. 更新データを挿入します。

      既存データと同じプライマリキーを持ち、name 列の値のみを変更した 2 行を挿入します。

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

   3. SQL エディターで StarRocks テーブルをクエリします。

      SELECT * FROM `test`.`score_board`;

      name 列の値のみが変更され、score 列は変更されていないことを確認できます。

      クエリは、id=1, name=starrocks-update, score=100 と id=2, name=flink-update, score=100 の 2 レコードを返します。

   条件付き更新

   この例では、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 コネクタはデータ書き込み時の更新条件として score 列を使用します。

      • sink.version：V1 に設定します。これにより、Flink コネクタはストリームロードインターフェイスを使用してデータをインポートします。

   2. Flink SQL クライアントで 2 行のデータをテーブルに挿入します。

      これらの行は、StarRocks テーブル内の既存の行と同じプライマリキーを持っています。1 行目の score の値は小さく、2 行目の score の値は大きいです。

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

   3. SQL エディターで StarRocks テーブルをクエリします。

      SELECT * FROM `test`.`score_board`;

      2 行目のみが変更され、1 行目は変更されていないことを確認できます。

      クエリは、1 行目 id=1, name=starrocks, score=100 (変更なし) と、2 行目 id=2, name=flink-update, score=101 (更新済み) の 2 レコードを返します。

ビットマップ列へのインポート

ビットマップデータ型は、ユニークビジター (UV) の算出など、正確な個別カウントのシナリオを高速化するためによく使用されます。次の例では、Flink SQL を使用して StarRocks テーブルのビットマップ列にデータをインポートし、StarRocks で UV 数をクエリする方法を示します。

1. SQL エディターで 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 'ページ ID',
     `visit_date` datetime NOT NULL COMMENT 'アクセス時間',
     `visit_users` BITMAP BITMAP_UNION NOT NULL COMMENT 'ユーザー ID'
   ) ENGINE=OLAP
   AGGREGATE KEY(`page_id`, `visit_date`)
   DISTRIBUTED BY HASH(`page_id`);

2. Flink SQL クライアントでテーブルを作成します。

   Flink はビットマップ型をサポートしていないため、次のようにカラムマッピングと型変換を実装する必要があります：

   • Flink テーブルでは、StarRocks テーブルの visit_users 列を表すために、visit_user_id 列を BIGINT として定義します。

   • sink.properties.columns 設定を使用して、visit_user_id 列のデータを 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. Flink SQL クライアントでデータを挿入します。

   異なるユーザーが異なる時刻にページにアクセスする状況をシミュレートするため、page_uv テーブルに複数の行を挿入します。

   visit_user_id は BIGINT 型です。Flink によってビットマップ型に自動的に変換されます。

   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 数をクエリします。

   StarRocks の集約機能を使用し、COUNT(DISTINCT visit_users) でページごとのユニークビジター (UV) 数を算出します。

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

   page_id が 2 の場合、count(DISTINCT visit_users) は 1 です。page_id が 1 の場合、count(DISTINCT visit_users) は 3 です。

HyperLogLog (HLL) 列へのインポート

HyperLogLog (HLL) は、近似的な個別カウントに使用されるデータ型であり、大規模データのシナリオでユニークビジター (UV) を算出するのに適しています。次の例では、Flink SQL を使用して StarRocks テーブルの HLL 列にデータをインポートし、StarRocks で UV 数をクエリする方法を示します。

1. SQL エディターで 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 'ページ ID',
     `visit_date` DATETIME NOT NULL COMMENT 'アクセス時間',
     `visit_users` HLL HLL_UNION NOT NULL COMMENT 'ユーザー ID'
   ) ENGINE=OLAP
   AGGREGATE KEY(`page_id`, `visit_date`)
   DISTRIBUTED BY HASH(`page_id`);

2. Flink SQL クライアントでテーブルを作成します。

   Flink は HLL 型をサポートしていないため、次のようにカラムマッピングと型変換を実装する必要があります：

   • Flink テーブルでは、StarRocks テーブルの visit_users 列を表すために、visit_user_id 列を BIGINT として定義します。

   • sink.properties.columns 設定を使用して、列をマッピングし、BIGINT 型の visit_user_id データを 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 クライアントでデータを挿入します。

   異なるユーザーが異なる時刻にページにアクセスする状況をシミュレートするため、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 エディターで UV 数をクエリします。

   StarRocks の集約機能を使用し、COUNT(DISTINCT visit_users) でページごとのユニークビジター (UV) 数を算出します。

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

   結果は次のとおりです。

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