E-MapReduce:Flink コネクタ

方法 1：DataStream API

// -------- 生の JSON 文字列ストリームを使用した sink -------- 
fromElements(new String[]{
    "{\"score\": \"99\", \"name\": \"stephen\"}",
    "{\"score\": \"100\", \"name\": \"lebron\"}"
}).addSink(
    StarRocksSink.sink(
        // 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()
    )
);


// -------- ストリーム変換を使用した sink --------
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(),
        // 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.column_separator", "\\x01")
            .withProperty("sink.properties.row_delimiter", "\\x02")
            .build(),
        // streamRowData でスロットを設定
        (slots, streamRowData) -> {
            slots[0] = streamRowData.score;
            slots[1] = streamRowData.name;
        }
    )
);

方法 2：SQL テーブル作成

この方法を実行する前に、com.starrocks.connector.flink.table.StarRocksDynamicTableSinkFactory を src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory に追加します。

// `structure` および `properties` を指定してテーブルを作成
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'" + // Stream Load プロパティ（例：`'sink.properties.columns' = 'k1, v1'`）
    ")"
);

前提条件

開始する前に、以下の条件を満たしていることを確認してください。

• Apache Flink 1.13（推奨）または 1.11 / 1.12

• MySQL データベースおよび StarRocks クラスターへのアクセス権限

環境のセットアップ

1. Apache Flink をダウンロードします。推奨バージョンは Apache Flink 1.13 です。対応する最小バージョンは 1.11 です。

2. MySQL 向けの Flink CDC コネクタ をダウンロードします。ご使用の Flink バージョンに対応するパッケージを選択してください。

3. Flink StarRocks コネクタ をダウンロードします。

   重要

   Flink StarRocks コネクタは Apache Flink 1.13、1.11、1.12 それぞれに対応した異なるバージョンが存在します。正しいバージョンをダウンロードしてください。

4. flink-sql-connector-mysql-cdc-xxx.jar および flink-connector-starrocks-xxx.jar の両方を flink-xxx/lib/ ディレクトリにコピーします。

5. smt.tar.gz パッケージをダウンロードして解凍し、接続の詳細を含む構成ファイルを編集します。構成ファイルでは次のセクションを使用します: サンプル構成:

   • db — MySQL データベース接続情報

   • be_num — StarRocks クラスター内のノード数

   • [table-rule.N] — データベース名およびテーブル名の正規表現によるマッチングルール。ルールセットごとに 1 つのセクションを構成します。

   • flink.starrocks.* — StarRocks クラスターの構成

   [db]
   host = 192.168.**.**
   port = 3306
   user = root
   password =
   
   [other]
   # number of backends in StarRocks
   be_num = 3
   # `decimal_v3` is supported since StarRocks-1.18.1
   use_decimal_v3 = false
   # file to save the converted DDL SQL
   output_dir = ./result
   
   [table-rule.1]
   # pattern to match databases for setting properties
   database = ^console_19321.*$
   # pattern to match tables for setting properties
   table = ^.*$
   
   ############################################
   ### flink sink configurations
   ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
   ############################################
   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

テーブルの作成および同期の開始

1. starrocks-migrate-tool を実行して CREATE TABLE ステートメントを生成します。出力は /result ディレクトリに保存されます。ls result を実行して、生成されたファイルを確認します。

   flink-create.1.sql    smt.tar.gz              starrocks-create.all.sql
   flink-create.all.sql  starrocks-create.1.sql

2. StarRocks でデータベースおよびテーブルを作成します。

   Mysql -h <フロントエンドの IP アドレス> -P 9030 -u root -p < starrocks-create.1.sql

3. Flink テーブルを作成し、継続的なデータ同期を開始します。

   重要

   Apache Flink 1.13 より前のバージョンを使用している場合、SQL スクリプトを直接実行できません。SQL ステートメントを 1 つずつ実行し、MySQL データベースのバイナリロギングを有効にしてください。

   bin/sql-client.sh -f flink-create.1.sql

4. 実行中の Flink ジョブのステータスを確認します。

   bin/flink list -running

   Flink Web UI または $FLINK_HOME/log ディレクトリ内のログファイルで、ジョブの詳細およびステータスを確認できます。

複数のルールセットの構成

異なるプロパティを持つ複数のデータベースまたはテーブルを同期するには、各ルールセットごとに 1 つの [table-rule.N] セクションを追加します。flink.starrocks.sink パラメーターを各ルールセットごとに個別に構成できます（例：異なるインポート頻度）。

2 つのルールセットを含むサンプル構成：

[table-rule.1]
# pattern to match databases for setting properties
database = ^console_19321.*$
# pattern to match tables for setting properties
table = ^.*$

############################################
### flink sink configurations
### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
############################################
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]
# pattern to match databases for setting properties
database = ^database2.*$
# pattern to match tables for setting properties
table = ^.*$

############################################
### flink sink configurations
### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
############################################
flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
flink.starrocks.load-url= 192.168.**.**:8030
flink.starrocks.username=root
flink.starrocks.password=
# If you cannot select appropriate CSV delimiters, switch to JSON format.
# Note that JSON import performance may deteriorate compared to CSV.
flink.starrocks.sink.properties.strip_outer_array=true
flink.starrocks.sink.properties.format=json

シャード化テーブルの統合

水平シャーディング後、大規模なテーブルのデータは複数のテーブルまたはデータベースに分割されることがあります。単一のルールセットを構成することで、複数のソーステーブルを 1 つの StarRocks テーブルに同期できます。

たとえば、edu_db_1 および edu_db_2 の両方に、同一のスキーマを持つ course_1 および course_2 が存在する場合、以下のルールにより、4 つのテーブルすべてを単一の StarRocks テーブルに統合できます。

[table-rule.3]
# pattern to match databases for setting properties
database = ^edu_db_[0-9]*$
# pattern to match tables for setting properties
table = ^course_[0-9]*$

############################################
### flink sink configurations
### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
############################################
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

starrocks-migrate-tool を実行すると、多対一の同期関係が自動的に作成されます。生成される StarRocks テーブルのデフォルト名は course__auto_shard です。名前を変更する場合は、実行前に生成された SQL ファイルを編集してください。