E-MapReduce:Flinkコネクタ

プロジェクトの pom.xml ファイルに次のコードを追加します。

<dependency>
    <groupId>com.starrocks</groupId>
    <artifactId>flink-connector-starrocks</artifactId>
    <!-- flink-1.11、flink-1.12の場合 -->
    <version>x.x.x_flink-1.11</version>
    <!-- flink-1.13の場合 -->
    <version>x.x.x_flink-1.13</version>
</dependency>

サンプルコード:

• 方法 1

  // -------- 生のJSON文字列ストリームを持つシンク --------
  fromElements(new String[]{
      "{\"score\": \"99\", \"name\": \"stephen\"}",
      "{\"score\": \"100\", \"name\": \"lebron\"}"
  }).addSink(
      StarRocksSink.sink(
          // シンクオプション
          StarRocksSinkOptions.builder()
              .withProperty("jdbc-url", "jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx")
              .withProperty("load-url", "fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port")
              .withProperty("username", "xxx")
              .withProperty("password", "xxx")
              .withProperty("table-name", "xxx")
              .withProperty("database-name", "xxx")
              .withProperty("sink.properties.format", "json")
              .withProperty("sink.properties.strip_outer_array", "true")
              .build()
      )
  );
  
  
  // -------- ストリーム変換を持つシンク --------
  class RowData {
      public int score;
      public String name;
      public RowData(int score, String name) {
          ......
      }
  }
  fromElements(
      new RowData[]{
          new RowData(99, "stephen"),
          new RowData(100, "lebron")
      }
  ).addSink(
      StarRocksSink.sink(
          // テーブル構造
          TableSchema.builder()
              .field("score", DataTypes.INT())
              .field("name", DataTypes.VARCHAR(20))
              .build(),
          // シンクオプション
          StarRocksSinkOptions.builder()
              .withProperty("jdbc-url", "jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx")
              .withProperty("load-url", "fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port")
              .withProperty("username", "xxx")
              .withProperty("password", "xxx")
              .withProperty("table-name", "xxx")
              .withProperty("database-name", "xxx")
              .withProperty("sink.properties.column_separator", "\\x01")
              .withProperty("sink.properties.row_delimiter", "\\x02")
              .build(),
          // streamRowDataでスロットを設定する
          (slots, streamRowData) -> {
              slots[0] = streamRowData.score;
              slots[1] = streamRowData.name;
          }
      )
  );

• 方法 2

  // `structure`と`properties`でテーブルを作成する
  // 必要：`com.starrocks.connector.flink.table.StarRocksDynamicTableSinkFactory` を `src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory` に追加する
  tEnv.executeSql(
      "CREATE TABLE USER_RESULT(" +
          "name VARCHAR," +
          "score BIGINT" +
      ") WITH ( " +
          "'connector' = 'starrocks'," +
          "'jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx'," +
          "'load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port'," +
          "'database-name' = 'xxx'," +
          "'table-name' = 'xxx'," +
          "'username' = 'xxx'," +
          "'password' = 'xxx'," +
          "'sink.buffer-flush.max-rows' = '1000000'," +
          "'sink.buffer-flush.max-bytes' = '300000000'," +
          "'sink.buffer-flush.interval-ms' = '5000'," +
          "'sink.properties.column_separator' = '\\x01'," +
          "'sink.properties.row_delimiter' = '\\x02'," +
          "'sink.max-retries' = '3'," +
          "'sink.properties.*' = 'xxx'" + // `'sink.properties.columns' = 'k1, v1'` のようなストリームロードプロパティ
      ")"
  );

次の表に、Flinkシンクのパラメータを示します。

Flink Change Data Capture（CDC）コネクタとStarRocks移行ツールを使用することで、MySQLデータベースからStarRocksクラスターに数秒以内にデータを同期できます。

このトピックで提供されているサンプルコードを実行する場合は、次の項目に注意してください。

• 複数のルールのセットを構成する必要がある場合は、データベース名とテーブル名を照合するために使用されるルールを構成し、各ルールのセットに対してFlink StarRocksコネクタを構成する必要があります。

  サンプルコード：

  [table-rule.1]
  # プロパティを設定するためのデータベースに一致するパターン
  database = ^console_19321.*$
  # プロパティを設定するためのテーブルに一致するパターン
  table = ^.*$
  
  ############################################
  ### flink シンク構成
  ### `connector`、`table-name`、`database-name` は設定しないでください。これらは自動生成されます
  ############################################
  flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
  flink.starrocks.load-url= 192.168.**.**:8030
  flink.starrocks.username=root
  flink.starrocks.password=
  flink.starrocks.sink.properties.column_separator=\x01
  flink.starrocks.sink.properties.row_delimiter=\x02
  flink.starrocks.sink.buffer-flush.interval-ms=15000
  
  [table-rule.2]
  # プロパティを設定するためのデータベースに一致するパターン
  database = ^database2.*$
  # プロパティを設定するためのテーブルに一致するパターン
  table = ^.*$
  
  ############################################
  ### flink シンク構成
  ### `connector`、`table-name`、`database-name` は設定しないでください。これらは自動生成されます
  ############################################
  flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
  flink.starrocks.load-url= 192.168.**.**:8030
  flink.starrocks.username=root
  flink.starrocks.password=
  # CSV形式でデータをインポートするための適切な区切り文字を選択できない場合は、JSON形式でデータをインポートできます。ただし、インポートのパフォーマンスが低下する可能性があります。JSON形式でデータをインポートするには、flink.starrocks.sink.properties.column_separatorパラメータとflink.starrocks.sink.properties.row_delimiterパラメータを次のパラメータに置き換えます。
  flink.starrocks.sink.properties.strip_outer_array=true
  flink.starrocks.sink.properties.format=json

  説明

  flink.starrocks.sinkパラメータを使用して、各ルールのセットのプロパティ（インポート頻度など）を構成できます。

• シャーディングが実行されると、大きなテーブルのデータが複数のテーブルに分割されたり、複数のデータベースに分散されたりする可能性があります。この場合、複数のテーブルから1つのテーブルにデータを同期するためのルールのセットを構成できます。

  たとえば、edu_db_1データベースとedu_db_2データベースの両方に、course_1テーブルとcourse_2テーブルの2つのテーブルがあるとします。すべてのテーブルのスキーマは同じです。次のコードを実行して、これらのテーブルからStarRocksのテーブルにデータを同期し、さらに分析することができます。

  [table-rule.3]
  # プロパティを設定するためのデータベースに一致するパターン
  database = ^edu_db_[0-9]*$
  # プロパティを設定するためのテーブルに一致するパターン
  table = ^course_[0-9]*$
  
  ############################################
  ### flink シンク構成
  ### `connector`、`table-name`、`database-name` は設定しないでください。これらは自動生成されます
  ############################################
  flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
  flink.starrocks.load-url= 192.168.**.**:8030
  flink.starrocks.username=root
  flink.starrocks.password=
  flink.starrocks.sink.properties.column_separator=\x01
  flink.starrocks.sink.properties.row_delimiter=\x02
  flink.starrocks.sink.buffer-flush.interval-ms=5000

  コードを実行すると、多対1の同期関係が自動的に生成されます。デフォルトでは、StarRocksのテーブルの名前はcourse__auto_shardです。生成されたSQLファイルでテーブル名を変更できます。

• SQLクライアントのCLIでテーブルを作成し、データを同期する必要がある場合は、バックスラッシュ（\）をエスケープする必要があります。

  サンプルコード：

  'sink.properties.column_separator' = '\\x01'
  'sink.properties.row_delimiter' = '\\x02'