Realtime Compute for Apache Flink:StarRocks

特徴

E-MapReduce 上の StarRocks は、CREATE TABLE AS SELECT (CTAS) および CREATE DATABASE AS SELECT (CDAS) 文をサポートします。CTAS は単一テーブルのスキーマとデータを同期し、CDAS はデータベース全体または同じデータベース内の複数のテーブルを同期します。詳細については、「Realtime Compute for Apache Flink の CTAS および CDAS 文を使用して MySQL データベースから StarRocks にデータを同期する」をご参照ください。

構文

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

パラメーター

データ型のマッピング

コード例

CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_source` (
  `runoob_id` BIGINT NOT NULL,
  `runoob_title` STRING NOT NULL,
  `runoob_author` STRING NOT NULL,
  `submission_date` DATE NULL
) WITH (
  'connector' = 'starrocks',
  'jdbc-url' = 'jdbc:mysql://ip:9030',
  'scan-url' = 'ip:18030',
  'database-name' = 'db_name',
  'table-name' = 'table_name',
  'password' = 'xxxxxxx',
  'username' = 'xxxxx'
);
CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_sink` (
  `runoob_id` BIGINT NOT NULL,
  `runoob_title` STRING NOT NULL,
  `runoob_author` STRING NOT NULL,
  `submission_date` DATE NULL
  PRIMARY KEY(`runoob_id`)
  NOT ENFORCED
) WITH (
  'jdbc-url' = 'jdbc:mysql://ip:9030',
  'connector' = 'starrocks',
  'load-url' = 'ip:18030',
  'database-name' = 'db_name',
  'table-name' = 'table_name',
  'password' = 'xxxxxxx',
  'username' = 'xxxx',
  'sink.buffer-flush.interval-ms' = '5000'
);

INSERT INTO runoob_tbl_sink SELECT * FROM runoob_tbl_source;

特徴

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

  アップストリームのデータベースまたはテーブルがダウンストリームの StarRocks インスタンスに存在しない場合、コネクタはそれを自動的に作成します。table.create.properties.* パラメーターを使用して、自動テーブル作成のオプションを構成できます。

• スキーマ変更の同期。

  StarRocks コネクタは、CreateTableEvent、AddColumnEvent、および DropColumnEvent イベントをダウンストリームデータベースに自動的に適用します。

• VVR 11.1 以降は、互換性のある列の型の変更をサポートします。詳細については、「ALTER TABLE | StarRocks」をご参照ください。

構文

source:
  ...

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030
  load-url: 127.0.0.1:8030
  username: root
  password: pass
  sink.buffer-flush.interval-ms: 5000   # データフラッシュ間隔を設定します。

構成

組み込みカタログの再利用

VVR 11.5 以降では、[データ管理] ページで作成された組み込みの StarRocks カタログを Flink CDC データインジェストジョブで直接参照できます。これにより、手動で設定する必要があるプロパティの数が減り、構成が簡素化されます。

sink:
  type: starrocks
  using.built-in-catalog: starrocks_catalog

データインジェストジョブは、次の StarRocks カタログオプションを自動的に再利用できます。

• jdbc-url

• http-url

• username

• password

• table.num-buckets

これらの値をオーバーライドするには、対応する YAML オプションを明示的に設定します。YAML オプションが優先されます。

型マッピング

スキーマ変更

データインジェスト sink として、StarRocks は次のスキーマ変更イベントをサポートします。

• CREATE TABLE イベント

  説明

  ダウンストリームの StarRocks テーブルが既に存在する場合、コネクタは再度作成しようとはしません。ダウンストリームのテーブルスキーマがアップストリームのスキーマと互換性があることを確認してください。

• ADD COLUMN イベント

  説明

  StarRocks では、プライマリキー列がテーブルの最初に表示される必要があります。新しい列は、それらの後に追加する必要があります。

• ALTER COLUMN TYPE イベント

  説明

  サポートされているスキーマ変更パスについては、公式 StarRocks ドキュメントをご参照ください。

• DROP COLUMN イベント

• TRUNCATE TABLE EVENT

• DROP TABLE EVENT

単一テーブルの同期

単一の MySQL テーブルを StarRocks に同期します。送信先のデータベースとテーブルが存在しない場合、コネクタは自動的にプライマリキーテーブルを作成します。

pipeline:
          name: MySQL to StarRocks Pipeline
      source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.test_source_table
        server-id: 5401-5499
        # (オプション) ジョブを再起動せずに、増分フェーズで新しく追加されたテーブルからデータを同期します。
        scan.binlog.newly-added-table.enabled: true
        # (オプション) テーブルと列のコメントを送信先に同期します。
        include-comments.enabled: true
        # (オプション) 読み取りパフォーマンスを向上させるために、キャプチャされたテーブルのバイナリログのみを逆シリアル化します。
        scan.only.deserialize.captured.tables.changelog.enabled: true
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
       
        # (オプション) データ量が少ないジョブの場合、永続化の遅延を防ぐためにフラッシュ間隔を短縮します。デフォルト：300000 (5 分)。
        sink.buffer-flush.interval-ms: 5000
        # (オプション) アップストリームの文字セットが utf8mb4 の場合、テキストの切り捨てを防ぐためにこのパラメーターを 4 に設定します。デフォルト：3。
        unicode-char.max-bytes: 4
        # (オプション) 自動作成されるテーブルのバケット数。このパラメーターは StarRocks バージョン 2.5.7 より前では必須です。それ以降のバージョンでは自動的に値を推論できます。
        table.create.num-buckets: 8
        # (オプション) 自動作成されるテーブルのレプリカ数。クラスターに基づいてこの値を構成します。
        table.create.properties.replication_num: 3
        # (オプション) StarRocks 3.2 以降では、このオプションを有効にしてスキーマ変更を高速化します。
        table.create.properties.fast_schema_evolution: true
        # 注：変換を使用してプライマリキーを変更する場合は、sink.ignore.update-before: false も設定する必要があります。
        # そうしないと、古いプライマリキーに関連付けられた行が送信先に残ります。
      
      pipeline:
        name: MySQL to StarRocks Pipeline

データベース全体の同期

MySQL データベース内のすべてのテーブルを StarRocks に同時に同期します。コネクタは送信先のデータベースとプライマリキーテーブルを自動的に作成するため、各テーブルを事前に作成する必要はありません。

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        # 正規表現を使用して、データベース内のすべてのテーブルを照合します。複数のデータベースを照合するには、パターンをカンマで区切ります。
        tables: test_db.\.*
        server-id: 5401-5499
        # (オプション) ジョブを再起動せずに、増分フェーズで新しく追加されたテーブルからデータを同期します。
        scan.binlog.newly-added-table.enabled: true
        # (オプション) テーブルと列のコメントを送信先に同期します。
        include-comments.enabled: true
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (オプション) データ量が少ないジョブの場合、永続化の遅延を防ぐためにフラッシュ間隔を短縮します。デフォルト：300000 (5 分)。
        sink.buffer-flush.interval-ms: 5000
        # (オプション) アップストリームの文字セットが utf8mb4 の場合、テキストの切り捨てを防ぐためにこのパラメーターを 4 に設定します。デフォルト：3。
        unicode-char.max-bytes: 4
        # (オプション) 自動作成されるテーブルのバケット数。このパラメーターは StarRocks バージョン 2.5.7 より前では必須です。それ以降のバージョンでは自動的に値を推論できます。
        table.create.num-buckets: 8
        # (オプション) 自動作成されるテーブルのレプリカ数。クラスターに基づいてこの値を構成します。
        table.create.properties.replication_num: 3
        # (オプション) StarRocks 3.2 以降では、このオプションを有効にしてスキーマ変更を高速化します。
        table.create.properties.fast_schema_evolution: true
      
      pipeline:
        name: MySQL to StarRocks Pipeline

データベース全体の同期中に特定のテーブルを除外

データベース全体を同期する場合、正規表現を使用して、一時テーブルや機密テーブルなど、送信先に同期したくないテーブルをスキップします。

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.\.*
        # この正規表現に一致するテーブルは同期されません。
        tables.exclude: test_db.tmp_.\*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (オプション) ロードインターフェイスのバージョン。V2 には StarRocks 2.4 以降が必要です。EMR Serverless StarRocks でストリーミングの問題が発生した場合は、V1 を使用してください。
        sink.version: V2
        # (オプション) データ量が少ないジョブの場合、永続化の遅延を防ぐためにフラッシュ間隔を短縮します。デフォルト：300000 (5 分)。
        sink.buffer-flush.interval-ms: 5000
        # (オプション) 自動作成されるテーブルのバケット数。このパラメーターは StarRocks バージョン 2.5.7 より前では必須です。
        table.create.num-buckets: 8
        # (オプション) StarRocks 3.2 以降では、このオプションを有効にしてスキーマ変更を高速化します。
        table.create.properties.fast_schema_evolution: true
      
      pipeline:
        name: MySQL to StarRocks Pipeline

指定したデータベースとテーブルへの同期

ODS レイヤーデータベースに書き込む場合など、送信先の StarRocks データベースまたはテーブル名をアップストリーム名と異なるものにする必要がある場合は、ルートを使用して名前を変更します。

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.\.*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (オプション) データ量が少ないジョブの場合、永続化の遅延を防ぐためにフラッシュ間隔を短縮します。デフォルト：300000 (5 分)。
        sink.buffer-flush.interval-ms: 5000
        # (オプション) 自動作成されるテーブルのバケット数。このパラメーターは StarRocks バージョン 2.5.7 より前では必須です。
        table.create.num-buckets: 8
      
      route:
        # MySQL の test_db データベース内のすべてのテーブルを、テーブル名を変更せずに StarRocks の test_db2 データベースに同期します。
        # <> は、一致したソーステーブル名に置き換えられるプレースホルダーです。
        - source-table: test_db.\.*
          sink-table: test_db2.<>
          replace-symbol: <>
      
      pipeline:
        name: MySQL to StarRocks Pipeline

シャーディングされたテーブルのマージ

同一のスキーマを持つ複数のシャーディングされたテーブルを単一の StarRocks テーブルにマージして、クエリと分析を統合します。

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        # user_0 や user_1 など、すべてのシャーディングされたテーブルを照合します。
        tables: test_db.user\.*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (オプション) データ量が少ないジョブの場合、永続化の遅延を防ぐためにフラッシュ間隔を短縮します。デフォルト：300000 (5 分)。
        sink.buffer-flush.interval-ms: 5000
        # (オプション) マージされたテーブルの場合、合計データ量に基づいてバケット数を明示的に指定します。
        table.create.num-buckets: 8
      
      route:
        # すべてのシャーディングされたテーブルを StarRocks の test_db.user テーブルにマージします。
        - source-table: test_db.user\.*
          sink-table: test_db.user
      
      pipeline:
        name: MySQL to StarRocks Pipeline

EVOLVE モードの有効化

デフォルトでは、LENIENT モードは、列の削除、テーブルの削除、テーブルの切り捨てなどのスキーマ変更を送信先に同期しません。厳密なスキーマ同期が必要な場合は、EVOLVE モードを有効にします。このモードには重大な制限があるため、使用する前に以下の注意事項を確認してください。

制限事項と注意事項

• 列名の変更はサポートされていません。アップストリームで列名の変更イベントが発生すると、ジョブは失敗します。

• 列の削除、テーブルの削除、またはテーブルの切り捨ては、送信先に適用されます。アップストリームでの偶発的な操作は、送信先テーブルに直接影響します。デフォルトの LENIENT モードは、テーブルの削除または切り捨てイベントを同期しないため、より安全です。

• ステートなしでジョブを再起動し、sink テーブルを削除しない場合、アップストリームと sink のスキーマの不一致によりジョブが失敗する可能性があります。ダウンストリームのテーブルスキーマを手動で調整する必要があります。

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.test_source_table
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
      
      pipeline:
        name: MySQL to StarRocks Pipeline
        # EVOLVE モードを有効にして、スキーマの変更を厳密に同期します。列名の変更など、サポートされていない変更が発生するとジョブは失敗します。
        schema.change.behavior: evolve