StarRocks SQL コネクタ - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

背景情報

StarRocks は、あらゆるシナリオで非常に高速なパフォーマンスと統一された分析エクスペリエンスを実現する次世代の超並列処理 (MPP) データウェアハウスです。StarRocks には次の利点があります：

StarRocks は MySQL プロトコルと互換性があるため、MySQL クライアントや一般的なビジネスインテリジェンス (BI) ツールを使用してデータに接続し、分析することができます。
StarRocks は分散アーキテクチャを使用しています：
- データテーブルを水平にパーティション分割し、複数のレプリカで保存します。
- クラスターは柔軟に拡張でき、最大 10 ペタバイト (PB) のデータを分析できます。
- MPP フレームワークを使用して、並列計算を高速化します。
- 複数のレプリカをサポートし、フォールトトレランスを実現します。

Flink コネクタは、データをキャッシュし、Stream Load を使用してシンクテーブルにバッチで書き込みます。ソーステーブルからは、データをバッチでフェッチして読み取ります。次の表に、StarRocks コネクタの機能を示します。

カテゴリ	説明
対応タイプ	ソーステーブル、ディメンションテーブル、シンクテーブル、およびデータインジェストターゲット
実行モード	ストリーミングモードとバッチモード
データ形式	CSV
コネクタ固有のメトリクス	なし
API タイプ	DataStream、SQL、およびデータインジェスト用の YAML
シンクテーブルの更新/削除サポート	はい

前提条件

E-MapReduce (EMR) 上の StarRocks クラスター、または Elastic Compute Service (ECS) 上のセルフマネージド StarRocks クラスターが必要です。

制約

ディメンションテーブルとの結合は、Ververica Runtime (VVR) 11.1 以降でのみサポートしています。
ネットワークアクセスの制約を回避するには、以下の StarRocks クラスターのポートをセキュリティグループまたはファイアウォールのホワイトリストに追加してください：9030、8030、8040、9060、8060、9020。
対象の StarRocks テーブルに非表示の生成列が含まれている場合は、Flink ジョブで書き込む実際の物理列のみを宣言し、生成列のフィールドは除外してください。 StarRocks の式パーティショニングによって自動的に作成された非表示の生成列 (たとえば __generated_partition_column_0 など) は、外部からの書き込みを受け付けません。デフォルトでは、コネクタは完全なスキーマに基づいて書き込みリクエストを作成するため、ジョブは失敗します。 StarRocks の生成列の詳細については、「Generated columns」をご参照ください。

機能

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

パラメーター

タイプ	パラメーター	説明	データ型	必須	デフォルト値	備考
一般	connector	使用するコネクタを指定します。	文字列	はい	—	値は `starrocks` である必要があります。
	jdbc-url	Java Database Connectivity (JDBC) の URL です。	文字列	はい	—	FE の IP アドレスと JDBC ポートを `jdbc:mysql://ip:port` のフォーマットで指定します。
	database-name	StarRocks データベースの名前です。	文字列	はい	—	—
	table-name	StarRocks テーブルの名前です。	文字列	はい	—	—
	username	StarRocks に接続するためのユーザー名です。	文字列	はい	—	—
	password	StarRocks に接続するためのパスワードです。	文字列	はい	—	—
	starrocks.create.table.properties	テーブルの自動作成プロパティを指定します。	文字列	いいえ	—	エンジンやレプリカ数などの初期テーブルプロパティを指定します。例：'starrocks.create.table.properties' = 'buckets 8' または 'starrocks.create.table.properties' = 'replication_num=1'。
ソース固有	scan-url	データスキャン URL です。	文字列	いいえ	—	FE の IP アドレスと HTTP ポートを指定します。フォーマット： `fe_ip:http_port;fe_ip:http_port`。説明複数の IP アドレスとポートを指定するには、セミコロン (;) で区切ります。
	scan.connect.timeout-ms	`flink-connector-starrocks` が StarRocks に接続する際のタイムアウトです。このタイムアウト内に接続が確立されない場合、コネクタはエラーを返します。	文字列	いいえ	1000	単位：ミリ秒。
	scan.params.keep-alive-min	クエリタスクのキープアライブ期間です。	文字列	いいえ	10	—
	scan.params.query-timeout-s	クエリタスクのタイムアウトです。この期間内に結果が返されない場合、システムはクエリタスクを停止します。	文字列	いいえ	600	単位：秒。
	scan.params.mem-limit-byte	BE ノード上の単一クエリのメモリ制限です。	文字列	いいえ	1073741824 (1 GB)	単位：バイト。
	scan.max-retries	失敗したクエリの最大リトライ回数です。この制限を超えると、コネクタはエラーを返します。	文字列	いいえ	1	—
シンク固有	load-url	データインポート URL です。	文字列	はい	—	FE の IP アドレスと HTTP ポートを `fe_ip:http_port;fe_ip:http_port` のフォーマットで指定します。説明複数の IP アドレスとポートを指定するには、セミコロン (;) で区切ります。
	sink.semantic	書き込みの配信セマンティクスです。	文字列	いいえ	at-least-once	有効な値： `at-least-once` (デフォルト) ：データが少なくとも 1 回配信されることを保証します。 `exactly-once` ：データが正確に 1 回配信されることを保証します。
	sink.buffer-flush.max-bytes	フラッシュする前にバッファに保持するデータの最大サイズです。	文字列	いいえ	94371840 (90 MB)	有効範囲： 64 MB ～ 10 GB。
	sink.buffer-flush.max-rows	フラッシュする前にバッファに保持する行の最大数です。	文字列	いいえ	500000	有効範囲： 1,000 ～ 5,000,000。
	sink.buffer-flush.interval-ms	バッファのフラッシュ間隔です。	文字列	いいえ	300000	有効範囲： 1,000 ms ～ 3,600,000 ms。
	sink.max-retries	失敗した書き込みの最大リトライ回数です。	文字列	いいえ	3	有効範囲： 0 ～ 1000。
	sink.connect.timeout-ms	StarRocks への接続タイムアウトです。	文字列	いいえ	1000	有効範囲： 100 ～ 60,000。単位：ミリ秒。
	sink.properties.*	シンク用の追加の Stream Load プロパティです。	文字列	いいえ	—	これらのパラメーターは Stream Load の動作を制御します。たとえば、`sink.properties.format` は、CSV などのインポートされるデータのフォーマットを指定します。その他のパラメーターについては、「Stream Load」をご参照ください。
ディメンション固有	lookup.cache.enabled	ディメンションテーブルのキャッシュを有効にするかどうかを指定します。	ブール値	いいえ	true	有効な値： `true` ：キャッシュを有効にします。テーブルデータが初めて読み取られた後、システムはそれをメモリにキャッシュします。後続のリクエストでは、有効期間内のキャッシュされたデータを使用することで I/O オーバーヘッドを削減します。 `false` ：キャッシュを無効にします。各クエリは直接データソースにアクセスします。重要この機能には、Realtime Compute for Apache Flink エンジンバージョン VVR 11.1 以降が必要です。以下のシナリオでは、この機能を無効にすることを推奨します：ディメンションテーブルのデータが頻繁に更新され、リアルタイムデータが必要な場合。テーブルに大量のデータが含まれており、メモリオーバーフローのリスクがある場合。

データ型マッピング

StarRocks データ型	Flink データ型
NULL	NULL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
BIGINT UNSIGNED 説明 Realtime Compute for Apache Flink エンジンの VVR 8.0.10 以降が必要です。	DECIMAL(20,0)
LARGEINT	DECIMAL(20,0)
FLOAT	FLOAT
DOUBLE	DOUBLE
DATE	DATE
DATETIME	TIMESTAMP
DECIMAL	DECIMAL
DECIMALV2	DECIMAL
DECIMAL32	DECIMAL
DECIMAL64	DECIMAL
DECIMAL128	DECIMAL
CHAR(m) 説明 VVR 8.0.10 では、MySQL と StarRocks 間のエンコーディングの違いに対応するため、CHAR の長さを自動的に 3 倍に拡張します (m=n3、n<=85)。 VVR 8.0.11 以降では、MySQL と StarRocks 間のエンコーディングの違いに対応するため、CHAR の長さを自動的に 4 倍に拡張します (m=n4、n<=63)。 StarRocks の CHAR 型の最大長は 255 です。したがって、自動的に拡張された長さが 255 を超えない場合にのみ、Flink は CHAR 型を StarRocks の CHAR 型にマッピングします。	CHAR(n)
VARCHAR(m) 説明 VVR 8.0.10 では、MySQL と StarRocks 間のエンコーディングの違いに対応するため、CHAR の長さを自動的に 3 倍に拡張します (m=n3、n>85)。 VVR 8.0.11 以降では、MySQL と StarRocks 間のエンコーディングの違いに対応するため、CHAR の長さを自動的に 4 倍に拡張します (m=n4、n>63)。 StarRocks の CHAR 型の最大長は 255 です。したがって、Flink の CHAR 型で自動的に拡張された長さが 255 を超える場合、Flink はその型を StarRocks の VARCHAR 型にマッピングします。	CHAR(n)
VARCHAR	STRING
VARBINARY 説明 Realtime Compute for Apache Flink エンジンの VVR 8.0.10 以降が必要です。	VARBINARY

例

以下の例のソーステーブル、シンクテーブル、およびディメンションテーブルは、組み合わせて使用できます。ソーステーブルの例で宣言されている runoob_tbl_source テーブルは、シンクテーブルとディメンションテーブルの例のデータソースとしても使用されます。これらの例を使用する前に、StarRocks で対応するフィールド型を持つテーブルを準備し、エンドポイント、データベース名とテーブル名、およびシークレット変数を置き換えてください。

ソーステーブル

CREATE TEMPORARY TABLE runoob_tbl_source (
    runoob_id BIGINT NOT NULL,
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    proc_time AS PROCTIME() -- 処理時間列。ディメンションテーブルの例のテンポラル結合で使用されます
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', -- FE の MySQL プロトコルエンドポイント。複数のエンドポイントはカンマ (,) で区切ります。
    'scan-url' = '<fe_host>:8030', -- FE の HTTP エンドポイント。複数のエンドポイントはカンマ (,) で区切ります。
    'database-name' = '<database_name>',
    'table-name' = '<source_table_name>',
    'username' = '${secret_values.starrocks_username}', -- 認証情報の漏洩を防ぐためにシークレット変数を使用します
    'password' = '${secret_values.starrocks_password}',
    'scan.params.query-timeout-s' = '600' -- 単一クエリのタイムアウト (秒)
);

説明

ソーステーブルはバッチモードで読み取られます。ジョブが開始されると、テーブルの現在のデータをスキャンし、その後、出力を停止します。増分変更を継続的に生成することはありません。

シンクテーブル

説明

StarRocks では、プライマリキー列を NULLABLE にすることができます。ただし、Flink は NULLABLE な列を含むプライマリキーをサポートしていません。Flink のデータ整合性モデルでは、プライマリキーは一意で非 NULL である必要があります。そうでない場合、Flink は Invalid primary key. Column 'xxx' is nullable というエラーをスローします。詳細については、「"Invalid primary key. Column 'xxx' is nullable." error」をご参照ください。

VVR 11+

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, -- プライマリキー列は NOT NULL として宣言する必要があります
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', -- FE の MySQL プロトコルエンドポイント。複数のエンドポイントはカンマ (,) で区切ります。
    'load-url' = '<fe_host>:8030', -- FE の HTTP エンドポイント。複数のエンドポイントはセミコロン (;) で区切ります。
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', -- 認証情報の漏洩を防ぐためにシークレット変数を使用します
    'password' = '${secret_values.starrocks_password}',
    'sink.version' = 'V2', -- トランザクショナルストリームロード。StarRocks 2.4 以降が必要です。
    'sink.semantic' = 'at-least-once', -- オプション。exactly-once もサポートされており、チェックポイント処理が必要です。
    'sink.buffer-flush.interval-ms' = '5000' -- バッチフラッシュ間隔 (ミリ秒)
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

VVR 8+

VVR 8.x では、sink.version のデフォルト値は V1 であり、V1、V2、および AUTO をサポートしています。sink.ignore.update-before のように、VVR 11.x でのみ登録されているパラメーターはサポートされていません：

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, -- プライマリキー列は NOT NULL として宣言する必要があります
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', -- FE の MySQL プロトコルエンドポイント。複数のエンドポイントはカンマ (,) で区切ります。
    'load-url' = '<fe_host>:8030', -- FE の HTTP エンドポイント。複数のエンドポイントはセミコロン (;) で区切ります。
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', -- 認証情報の漏洩を防ぐためにシークレット変数を使用します
    'password' = '${secret_values.starrocks_password}',
    'sink.semantic' = 'at-least-once', -- オプション。exactly-once もサポートされており、チェックポイント処理が必要です。
    'sink.buffer-flush.interval-ms' = '5000', -- バッチフラッシュ間隔 (ミリ秒)
    'sink.max-retries' = '3' -- Stream Load 失敗後のリトライ回数
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

ディメンションテーブル

ディメンションテーブル結合は、VVR 11.1 以降でのみサポートされています。ディメンションテーブルでは結合キーとして主キーを宣言する必要があり、テンポラル結合では処理時間を使用します。以下の例は、ソーステーブルの例にある runoob_tbl_source テーブルをベースにしています。このテーブルには、proc_time 処理時間列が含まれています。

CREATE TEMPORARY TABLE sr_dim (
    runoob_id BIGINT NOT NULL,
    runoob_author STRING,
    PRIMARY KEY (runoob_id) NOT ENFORCED -- ディメンションテーブルでは、結合キーとしてプライマリキーを宣言する必要があります
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', -- FE の MySQL プロトコルエンドポイント。複数のエンドポイントはカンマ (,) で区切ります。
    'scan-url' = '<fe_host>:8030', -- FE の HTTP エンドポイント。複数のエンドポイントはカンマ (,) で区切ります。
    'database-name' = '<database_name>',
    'table-name' = '<dim_table_name>',
    'username' = '${secret_values.starrocks_username}', -- 認証情報の漏洩を防ぐためにシークレット変数を使用します
    'password' = '${secret_values.starrocks_password}',
    'lookup.cache.ttl-ms' = '5000', -- ディメンションテーブルキャッシュの TTL (ミリ秒)
    'lookup.cache.enabled' = 'true' -- デフォルト： true。キャッシュを使用せずにテーブルを直接クエリする場合は false に設定します。
);

SELECT o.runoob_id, o.runoob_title, d.runoob_author
FROM runoob_tbl_source AS o
JOIN sr_dim FOR SYSTEM_TIME AS OF o.proc_time AS d -- ソーステーブルの処理時間列を使用するテンポラル結合
ON o.runoob_id = d.runoob_id;