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

StarRocks コネクターの使用方法について説明します。

背景情報

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

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

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

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

前提条件

EMR 上の StarRocks クラスター、または ECS 上のセルフマネージド StarRocks クラスターが必要です。

制限事項

Ververica Runtime (VVR) 11.1 以降でのみ、ディメンションテーブルとの結合をサポートします。
ネットワークアクセスの制限を回避するため、次の StarRocks クラスターのポートをセキュリティグループまたはファイアウォールのホワイトリストに追加してください：9030、8030、8040、9060、8060、9020。

SQL

機能

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	範囲：64,000 から 5,000,000 です。
	sink.buffer-flush.interval-ms	バッファフラッシュの間隔です。	文字列	いいえ	300000	範囲：1,000 ms から 3,600,000 ms です。
	sink.max-retries	失敗した書き込みの最大再試行回数です。	文字列	いいえ	3	範囲：0 から 10 です。
	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 です。したがって、Flink は、自動的に拡張された長さが 255 を超えない場合にのみ、CHAR 型を StarRocks の CHAR 型にマッピングします。	CHAR(n)
VARCHAR(m) 説明 VVR 8.0.10 は、MySQL と StarRocks のエンコーディングの違いに対応するため、VARCHAR の長さを自動的に 3 倍 (m=n3, n>85) に拡張します。 VVR 8.0.11 以降は、MySQL と StarRocks のエンコーディングの違いに対応するため、VARCHAR の長さを自動的に 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

コード例

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

データインジェスト

StarRocks パイプラインコネクタを使用して、アップストリームのデータソースからデータレコードとスキーマの変更を外部 StarRocks データベースに書き込みます。 StarRocks コネクタは、コミュニティ版と Alibaba Cloud のフルマネージド EMR Serverless StarRocks の両方をサポートしています。

機能

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

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

StarRocks コネクタは、CreateTableEvent、AddColumnEvent、および DropColumnEvent イベントをダウンストリームのデータベースに自動的に適用します。
VVR 11.1 以降では、互換性のあるカラムタイプの変更をサポートしています。詳細については、「ALTER TABLE | StarRocks」をご参照ください。

使用上の注意

同期された各テーブルにはプライマリキーが必要です。プライマリキーがないテーブルの場合、ダウンストリームにデータを書き込むには、transform ブロックでプライマリキーを指定する必要があります。たとえば、
```
transform:
  - source-table: ...
    primary-keys: id, ...
```
自動作成されたテーブルでは、バケットキーはプライマリキーと同一であり、テーブルにパーティションキーを設定できません。
スキーマ変更を同期する場合、新しいカラムは既存カラムの末尾にのみ追加できます。デフォルトの Lenient スキーマ変更モードでは、他の位置への挿入は自動的に末尾へ移動します。
StarRocks 2.5.7 より前のバージョンを使用する場合は、 table.create.num-buckets パラメーターでバケット数を明示的に指定する必要があります。StarRocks 2.5.7 以降では、適切なバケット数を自動的に決定できます。
StarRocks 3.2 以降を使用する場合は、スキーマ変更を高速化するために、 table.create.properties.fast_schema_evolution オプションを有効にすることを推奨します。
EMR Serverless StarRocks へのデータインジェストに CDC YAML を使用すると、ストリーミングの問題が発生する場合があります。次のいずれかの回避策を使用できます：
- Flink SQL StarRocks コネクタを使用し、 sink.version=V1 パラメーターを設定します。
- FE パラメーター emr_internal_redirect を有効にします。
- SLB の代わりに 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   # バッファーフラッシュ間隔を設定します。

設定

パラメーター	説明	型	必須	デフォルト	備考
`type`	シンクコネクタのタイプを指定します。	String	はい	—	`starrocks` に設定します。
`name`	シンクの表示名。	String	いいえ	—	—
`jdbc-url`	データベース接続用の JDBC URL。	String	はい	—	複数のアドレスをカンマ (`,`) で区切って指定できます。例: `jdbc:mysql://fe_host1:fe_query_port1,fe_host2:fe_query_port2,fe_host3:fe_query_port3`。
`load-url`	Stream Load 用の FE ノードの HTTP URL。	String	はい	—	複数のアドレスをセミコロン (`;`) で区切って指定できます。例： `fe_host1:fe_http_port1;fe_host2:fe_http_port2`。
`username`	StarRocks 接続用のユーザー名。	String	はい	—	このユーザーは、ターゲットテーブルに対して少なくとも SELECT および INSERT 権限を持っている必要があります。StarRocks の GRANT コマンドで必要な権限を付与できます。
`password`	StarRocks 接続用のパスワード。	String	はい	—	—
`sink.semantic`	データ書き込みの配信セマンティクス。	String	いいえ	at-least-once	有効な値： `at-least-once` (デフォルト)：データが少なくとも 1 回配信されることを保証します。 `exactly-once`：データが正確に 1 回配信されることを保証します。
`sink.label-prefix`	Stream Load ジョブのラベルプレフィックス。	String	いいえ	—	—
`sink.connect.timeout-ms`	HTTP 接続を確立するためのタイムアウト。	整数	いいえ	30000	単位：ミリ秒。値は 100 から 60000 の間である必要があります。
`sink.wait-for-continue.timeout-ms`	サーバーからの 100 Continue レスポンスを待機するためのタイムアウト。	整数	いいえ	30000	単位：ミリ秒。値は 3000 から 600000 の間である必要があります。
`sink.buffer-flush.max-bytes`	フラッシュがトリガーされる前の、インメモリキャッシュの最大サイズ (バイト単位)。	Long	いいえ	157286400	単位：バイト。値は 64 MB から 10 GB の間である必要があります。説明このキャッシュサイズはすべてのテーブルで共有されます。バッファーがいっぱいになると、コネクタはフラッシュするいくつかのテーブルを選択します。値を大きくするとスループットが向上する可能性がありますが、取り込みレイテンシーが増加する可能性があります。
`sink.buffer-flush.max-rows`	フラッシュがトリガーされる前の、インメモリキャッシュ内の最大行数。	Long	いいえ	500000	値は 64,000 から 5,000,000 の間である必要があります。
`sink.buffer-flush.interval-ms`	各テーブルのバッファーのフラッシュ間隔。	Long	いいえ	300000	単位：ミリ秒。説明少量のデータを同期するジョブの場合、データが永続化されるまでの長い遅延を避けるために、この値を小さくしてください。
`sink.max-retries`	最大再試行回数。	Long	いいえ	3	値は 0 から 1000 の間である必要があります。
`sink.scan-frequency.ms`	コネクタがバッファーをフラッシュするかどうかを確認する頻度。	Long	いいえ	50	単位：ミリ秒。
`sink.io.thread-count`	Stream Load に使用されるスレッド数。	整数	いいえ	2	—
`sink.at-least-once.use-transaction-stream-load`	Stream Load トランザクションインターフェイスをデータ取り込みに使用するかどうかを指定します。	ブール値	いいえ	true	このオプションは、データベースがサポートしている場合にのみ有効です。
`sink.properties.*`	シンクの追加プロパティ。	String	いいえ	—	サポートされているプロパティについては、「STREAM LOAD」をご参照ください。
`table.create.num-buckets`	自動作成されるテーブルのバケット数。	整数	いいえ	—	StarRocks 2.5.7 以降では、このパラメーターはオプションであり、バケット数は自動的に推測されます。 StarRocks 2.5.6 以前では、このパラメーターは必須です。
`table.create.properties.*`	テーブルの自動作成に関する追加プロパティ。	String	いいえ	—	たとえば、`'table.create.properties.fast_schema_evolution' = 'true'` を指定することで、高速なスキーマ変更を有効にできます。詳細については、「StarRocks ドキュメント」をご参照ください。
`table.schema-change.timeout`	スキーマ変更操作のタイムアウト。	整数	いいえ	1800	値は秒単位の整数である必要があります。説明スキーマ変更操作がこの制限を超えると、ジョブは失敗します。
`unicode-char.max-bytes`	各 Unicode 文字に割り当てるバイト数。	整数	いいえ	3	CDC では、VARCHAR 型の長さは文字数で測定されますが、StarRocks では、VARCHAR 型の長さはバイト数で測定されます。ほとんどの場合、Unicode 文字は UTF-8 エンコーディング後も 3 バイトを超えません。ただし、一部の珍しい文字や絵文字記号は 4 バイト以上を占有する場合があります。

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

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

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

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

jdbc-url
http-url
username
password
table.num-buckets

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

型マッピング

説明

StarRocks は、すべての CDC YAML 型をサポートしているわけではありません。サポートされていない型を sink に書き込むと、ジョブは失敗します。変換処理で CAST 組み込み関数を使用してサポートされていないデータを変換するか、射影文を使用して結果テーブルから削除できます。詳細については、「Develop a Flink CDC data ingestion job」をご参照ください。

CDC 型	StarRocks 型	備考
TINYINT	TINYINT	—
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
BOOLEAN	BOOLEAN
DATE	DATE
TIMESTAMP	DATETIME
TIMESTAMP_LTZ	DATETIME
DECIMAL(p, s)	DECIMAL(p, s)	StarRocks はプライマリキーとして DECIMAL をサポートしていないため、コネクターはアップストリームの DECIMAL プライマリキーカラムを、同期される StarRocks スキーマ内で自動的に VARCHAR に変換します。
CHAR(n) (n <= 85)	CHAR(n × 3)	CDC は長さを文字数で測定しますが、StarRocks はバイト数を使用します。コネクターは、マルチバイトの UTF-8 文字を考慮して長さを 3 倍にします。説明 StarRocks の CHAR 型の最大長は 255 です。そのため、長さが 85 以下の CDC CHAR 型のみが StarRocks の CHAR 型にマッピングされます。説明 `unicode-char.max-bytes` パラメーターを設定して、各 Unicode 文字に割り当てるバイト数を増やすことができます。
CHAR(n) (n > 85)	VARCHAR(n × 3)	CDC は長さを文字数で測定しますが、StarRocks はバイト数を使用します。コネクターは、マルチバイトの UTF-8 文字を考慮して長さを 3 倍にします。説明 CDC は長さを文字数で測定しますが、StarRocks はバイト数を使用します。コネクターは長さを 3 倍にします。結果が StarRocks の CHAR 型の上限である 255 バイトを超えるため、VARCHAR にマッピングされます。説明 `unicode-char.max-bytes` パラメーターを設定して、各 Unicode 文字に割り当てるバイト数を増やすことができます。
VARCHAR(n)	VARCHAR(n × 3)	CDC は長さを文字数で測定しますが、StarRocks はバイト数を使用します。コネクターは、マルチバイトの UTF-8 文字を考慮して長さを 3 倍にします。説明 `unicode-char.max-bytes` パラメーターを設定して、各 Unicode 文字に割り当てるバイト数を増やすことができます。
BINARY(n)	BINARY(n+2)	データ整合性を確保するため、2 バイトのパディングが追加されます。
VARBINARY(n)	VARBINARY(n+1)	データ整合性を確保するため、1 バイトのパディングが追加されます。

スキーマ変更

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

テーブル作成イベント

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

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

列型変更イベント

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

列削除イベント
テーブル切り捨てイベント
テーブル削除イベント