Hologres カタログを設定すると、Realtime Compute 開発コンソールで Hologres メタデータを直接読み取ることができます。これにより、Hologres テーブルを手動で登録する必要がなくなり、開発効率が向上し、データの正確性が保証されます。このトピックでは、Hologres カタログの設定、表示、使用、削除方法について説明します。
前提条件
専用の Hologres インスタンスが作成され、そのインスタンスにデータベースが作成されていること。詳細については、「データベースの作成」をご参照ください。
制限事項
カタログは変更できません。変更を加えるには、既存のカタログを削除して新しいカタログを作成する必要があります。
Hologres インスタンスは専用型インスタンスである必要があります。Realtime Compute for Apache Flink は Hologres の内部テーブルからの読み取りと書き込みしかできないため、共有クラスターインスタンスはサポートされていません。
Hologres カタログの作成
カタログが作成されると、その構成は変更できません。変更を加えるには、既存のカタログを削除して新しいカタログを作成する必要があります。
UI の使用
Hologres コネクタでサポートされているパラメーターを設定するには、SQL コマンドを使用して Hologres カタログを作成できます。
データ管理ページに移動します。
Realtime Compute コンソールにログインします。対象のワークスペースの [アクション] 列で、[コンソール] をクリックします。
[データ管理] をクリックします。
[カタログの作成] をクリックし、[Hologres] を選択して、[次へ] をクリックします。
パラメーターを設定します。

パラメーター
説明
必須
備考
catalogname
Hologres カタログの名前。
はい
名前には、小文字 (a-z) と数字 (0-9) のみを含めることができます。大文字、ハイフン (-)、アンダースコア (_)、およびその他の特殊文字はサポートされていません。
endpoint
Hologres インスタンスのネットワークポート。
はい
Hologres インスタンスと Flink ワークスペースが同じ VPC にある場合: に移動します。対象インスタンスの製品ページで、[ネットワーク情報] セクションから [VPC] エンドポイントを取得します。
他のネットワークタイプについては、「Hologres インスタンスのエンドポイントを取得するにはどうすればよいですか?」をご参照ください。
username
カスタムアカウントのユーザー名。フォーマットは
BASIC$<user_name>です。Alibaba Cloud アカウントまたは Resource Access Management (RAM) ユーザーの AccessKey ID。
はい
ユーザーは、対応する Hologres データベースにアクセスするための権限を持っている必要があります。Hologres データベースの権限とユーザー管理の詳細については、「Hologres 権限モデル」および「ユーザー管理」をご参照ください。
AccessKey ID と AccessKey Secret の取得方法の詳細については、「AccessKey ID と AccessKey Secret を表示するにはどうすればよいですか?」をご参照ください。
重要AccessKey またはパスワードの漏洩を防ぐために、プロジェクト変数を使用して値を指定します。詳細については、「プロジェクト変数」をご参照ください。
カスタムアカウントを使用して作成されたカタログには、そのアカウントが権限を持つデータベースのみが表示されます。AccessKey ペアを使用して作成されたカタログは、インスタンス内のすべてのデータベースを表示できます。
password
カスタムアカウントのパスワード。
Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret。
はい
dbname
Hologres データベースの名前。
はい
対象の Hologres インスタンスにデータベースを作成する必要があります。そうしないと、カタログの作成時にエラーが発生します。
[OK] をクリックします。
カタログを作成した後、[メタデータ] で表示できます。
SQL コマンドの使用
[データクエリ] ページのエディターで、Hologres カタログを作成するコマンドを入力します。
構文:
CREATE CATALOG <catalogname> WITH ( 'type' = 'hologres', 'endpoint' = '<endpoint>', 'username' = '<AccessKey>', 'password' = '<AccessSecret>', 'dbname' = '<dbname>' );例:
簡単な例
リアルタイム消費の例
CREATE CATALOG holocatalog WITH ( 'type' = 'hologres', 'endpoint' = 'hgpostcn-cn-******-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80', 'username' = 'LTAI********************', 'password' = '${secret_values.ak_holo}', 'dbname' = 'holo_test' );バイナリログデータを消費するには、バイナリログを有効にする必要があります。
CREATE CATALOG holocatalog WITH ( 'type' = 'hologres', 'endpoint' = 'hgpostcn-cn-******-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80', 'username' = 'LTAI********************', 'password' = '${secret_values.ak_holo}', 'dbname' = 'holo_test', 'binlog' = 'true', -- カタログを作成するときに、ソーステーブル、ディメンションテーブル、および結果テーブルでサポートされている WITH パラメーターを設定できます。このカタログのテーブルを使用すると、これらのデフォルトパラメーターが自動的に追加されます。 'cdcmode' = 'true', 'connectionpoolname' = 'the_conn_pool', 'table_property.binlog.level' = 'replica', -- カタログの作成時に、永続化された Hologres テーブルのプロパティを渡すこともできます。その後、テーブルを作成すると、バイナリログがデフォルトで有効になります。 'table_property.binlog.ttl' = '259200' );次の表にパラメーターを説明します。
パラメーター
説明
必須
備考
catalogname
Hologres カタログの名前。
はい
名前には、小文字 (a-z) と数字 (0-9) のみを含めることができます。大文字、ハイフン (-)、アンダースコア (_)、およびその他の特殊文字はサポートされていません。
type
タイプ。
はい
値は `hologres` に固定されています。
endpoint
Hologres インスタンスのエンドポイント。
はい
Hologres インスタンスと Flink ワークスペースが同じ VPC にある場合: に移動します。対象インスタンスの製品ページで、[VPC] エンドポイントを [ネットワーク情報] セクションから取得します。
他のネットワークタイプについては、「Hologres インスタンスのエンドポイントを取得するにはどうすればよいですか?」をご参照ください。
username
Alibaba Cloud アカウントの AccessKey ID。
はい
詳細については、「AccessKey ID と AccessKey Secret を表示するにはどうすればよいですか?」をご参照ください。
説明AccessKey 情報の漏洩を防ぐために、変数を使用してパスワードの値を指定します。詳細については、「プロジェクト変数」をご参照ください。
AccessKey に対応するユーザーは、対応する Hologres データベースにアクセスするための権限を持っている必要があります。Hologres データベースの権限の詳細については、「Hologres 権限モデル」をご参照ください。
password
Alibaba Cloud アカウントの AccessKey Secret。
はい
dbname
Hologres データベースの名前。
はい
対象の Hologres インスタンスにデータベースを作成する必要があります。そうしないと、カタログの作成時にエラーが発生します。
ignore-non-persisted-options
Hologres カタログで非永続化オプションを持つテーブルを作成するときに、非永続化オプションを無視するかどうかを指定します。
いいえ
有効な値:
true (デフォルト): テーブルが作成され、すべての非永続化オプションは無視されます。
false: エラーが報告され、テーブルの作成に失敗します。
説明Hologres カタログのテーブルオプションは、カタログからテーブル情報を再度読み取るときに DDL ステートメントで定義したのと同じ情報を取得できる場合に永続化されます。現在、永続化できるのは endpoint、username、password、および dbname オプションのみです。
catalog.table.metadata-columns
このカタログを使用してソーステーブルを指定する場合、必要に応じて Hologres バイナリログソーステーブルからテーブルスキーマにメタデータ列を追加できます。複数のメタデータ列はセミコロン (;) で区切ります。例:
hg_binlog_event_type;hg_binlog_timestamp_us。いいえ
デフォルトでは、メタデータ列は追加されません。このパラメーターを設定すると、指定されたメタデータ列が返されるテーブルスキーマに追加されます。これらの列は Hologres バイナリログソーステーブルにのみ適用されます。したがって、このカタログによって返されるテーブルは、ソーステーブルとしてのみ使用でき、結果テーブルやディメンションテーブルとしては使用できません。6 種類のメタ列がサポートされています。詳細については、「Hologres バイナリログフィールド」をご参照ください。
説明このパラメーターは、Ververica Runtime (VVR) 8.0.11 以降でのみサポートされています。
Hologres コネクタでサポートされているその他のパラメーター
カタログを作成するときに、WITH パラメーター を含むいくつかのパラメーターを指定できます。このカタログのテーブルを使用すると、これらのパラメーターはデフォルトで設定されます。
いいえ
この機能を使用するには、
ignore-non-persisted-optionsパラメーターを true に設定する必要があります。カタログを作成するコードを入力した後、右上隅の [実行] をクリックします。
Hologres カタログの表示
Hologres カタログを設定した後、次のようにメタデータを表示できます:
データ管理ページに移動します。
Realtime Compute コンソールにログインします。
対象のワークスペースの [アクション] 列にある [コンソール] をクリックします。
[データ管理] をクリックします。
[カタログリスト] ページで、[カタログ名] と [タイプ] を表示します。
[表示] をクリックして、対象のカタログ内のデータベースとテーブルを表示します。スキーマが `public` の場合、スキーマのプレフィックスはテーブル名から省略されます。
Hologres カタログの使用
注:
スキーマが `public` の場合、スキーマ名のプレフィックスを省略し、テーブル名を ${schema_name.table_name} の代わりに ${table_name} として指定できます。
Hologres カタログから読み取られたテーブルは、更新されたデータの消費をサポートします。これらのテーブルでは、`ignoredelete` プロパティはデフォルトで `false` であり、`mutatetype` プロパティはデフォルトで `insertorupdate` です。`ignoredelete` および `mutatetype` プロパティの詳細については、「ワイドテーブルのマージと部分更新」をご参照ください。
Hologres テーブルの作成
この例では、holocatalog Hologres カタログにある holodb データベースに holotable という名前のテーブルを作成します。
登録済みの Hologres カタログにテーブルを作成する場合、WITH 句に `connector` パラメーターが必要であり、その値は `hologres` でなければなりません。`endpoint` などの他のパラメーターは省略できます。
Hologres テーブル定義でサポートされている WITH パラメーターを直接追加または変更することはできません。代わりに、SQL ヒントを使用して INSERT ステートメントで追加または変更できます。
次のステップを実行します:
UI の使用
データ管理ページに移動します。
Realtime Compute コンソールにログインします。
対象のワークスペースの [アクション] 列で、[コンソール] をクリックします。
[データ管理] をクリックします。
対象のデータベースに移動します。
対象のカタログの [アクション] 列で、[表示] をクリックします。
対象のデータベースの [アクション] 列で、[表示] をクリックします。
Hologres テーブルを作成します。
[テーブルの作成] をクリックします。
[組み込みコネクタとの接続] タブで、Hologres コネクタを選択し、[次へ] をクリックします。
CREATE TABLE ステートメントを入力し、パラメーターを設定します。次の表に例を示します。
構文
例
CREATE TABLE `${catalog_name}`.`${db_name}`.`${table_name}` ( ... ) WITH ( 'connector' = 'hologres' );CREATE TABLE `holocatalog`.`holo_test`.`product` ( id INT, name STRING ) WITH ( 'connector' = 'hologres' );[OK] をクリックします。
SQL コマンドの使用
[データクエリ] ページのエディターで、次の CREATE TABLE ステートメントを入力します。
Hologres テーブルは、次のいずれかの方法で作成できます:
USE CATALOG HoloNameコマンドの使用USE ステートメントを使用して、Hologres カタログを直接参照します。
構文
例
USE CATALOG ${catalog_name}; CREATE TABLE `${db_name}`.`${schema_name.table_name}`( ... ) WITH ( 'connector' = 'hologres' );USE CATALOG holocatalog; CREATE TABLE `holodb`.`holotable` ( id INT, name STRING ) WITH ( 'connector' = 'hologres' );DDL ステートメントの使用
DDL ステートメントで Hologres カタログを直接参照します。
構文
例
CREATE TABLE `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`( ... ) WITH ( 'connector' = 'hologres' );CREATE TABLE `holocatalog`.`holodb`.`holotable` ( id INT, name STRING ) WITH ( 'connector' = 'hologres' );DDL ステートメントで物理テーブルのプロパティを設定できます。
CREATE TABLE `holocatalog`.`holodb`.`holotable` ( id INT, name STRING ) WITH ( 'connector' = 'hologres', 'table_property.orientation' = 'column', 'table_property.distribution_key' = 'a', 'table_property.clustering_key' = 'b:desc', 'table_property.bitmap_columns' = 'a,b', 'table_property.segment_key' = 'c', 'table_property.time_to_live_in_seconds' = '86400', 'table_property.binlog.level' = 'replica', 'table_property.binlog.ttl' = '86400' );CREATE TABLE ステートメントを入力し、右上隅の [実行] をクリックします。
Hologres テーブルを作成するときに、WITH 句でテーブルのプロパティを設定できます。一部のプロパティは変更できません。テーブルのプロパティを正しく設定すると、システムがデータを効率的に整理およびクエリするのに役立ちます。
物理テーブルのプロパティ: サポートされているテーブルプロパティは Hologres のものと同じです。カタログは、区別のために
table_property.プレフィックスを追加します。パラメーターの詳細については、「テーブル作成の概要」および「Hologres バイナリログのサブスクライブ」をご参照ください。パラメーター
説明
例
変更可能
table_property.orientation
テーブルのストレージフォーマット。
'table_property.orientation' = 'row,column'
これは変更できません。
table_property.table_group
テーブルグループ。
'table_property.table_group' = 'table_group_xxx'
table_property.distribution_key
分散キー。
'table_property.distribution_key' = 'a,b'
table_property.clustering_key
クラスタリングキー。
'table_property.clustering_key' = 'a,b:desc'
able_property.event_time_column (旧 table_property.segment_key)
セグメントキー。
'table_property.event_time_column' = 'c,d' (または 'table_property.segment_key' = 'c,d')
table_property.bitmap_columns
ビットマップインデックス。
'table_property.bitmap_columns' = 'a:on,b:off'
変更可能
table_property.dictionary_encoding_columns
ビットエンコーディング。
'table_property.dictionary_encoding_columns' = 'a:on,b:off,c:auto'
table_property.time_to_live_in_seconds
テーブルデータの存続時間 (TTL)。
'table_property.time_to_live_in_seconds' = '864000'
table_property.binlog.level
バイナリログを有効にするかどうかを指定します。
'table_property.binlog.level' = 'replica'
table_property.binlog.ttl
バイナリログの TTL。
'table_property.binlog.ttl' = '86400'
寛容モードパラメーター: Hologres カタログを使用してテーブルを作成するときに、enableTypeNormalization パラメーターを使用して型の寛容モードを有効にできます。
ディメンション
説明
シナリオ
Create Table As Select (CTAS) などのシナリオでは、既存のフィールドのデータ型の精度を調整する必要がある場合があります (例: VARCHAR(10) から VARCHAR(20) へ)。または、データ型を変更する必要がある場合があります (例: SMALLINT から INT へ)。
注
CTAS ジョブを初めて開始するときに、寛容モードを有効にする必要があります。最初の起動時に寛容モードが有効になっていない場合は、子孫テーブルを削除し、ジョブのステートレス再起動を実行して設定を有効にする必要があります。
寛容モードが有効になった後、アップストリームでデータ型が変更された場合、変更された型の正規化された型が元の正規化された型と同じであれば、型の変更は成功したと見なされ、CTAS ジョブは正常に実行されます。そうでない場合、型に互換性がなく、CTAS ジョブは例外をスローします。
有効な値
false (デフォルト): 型マッピングに基づいて Hologres 物理テーブルが作成されます。
true: 型の寛容モードが有効になります。Hologres 物理テーブルが作成されると、型正規化後により高い精度のデータ型が使用されます。次のリストは、現在の型正規化ルールを説明しています:
TINYINT、SMALLINT、INT、および BIGINT は BIGINT に正規化されます。
CHAR、VARCHAR、および STRING は STRING に正規化されます。
FLOAT と DOUBLE は DOUBLE に正規化されます。
他のデータ型は、元の型マッピングルールに基づいて作成されます。詳細については、「型マッピング」をご参照ください。
Hologres テーブルの変更
次の表に、テーブルの変更でサポートされている操作と例を示します。
操作 | 構文と例 |
テーブルプロパティの変更 | 一部のテーブルプロパティのみ変更できます。詳細については、「Hologres テーブルの作成」をご参照ください。
|
テーブル名の変更 |
|
列の追加 |
|
列名の変更 |
|
列コメントの変更 |
|
Hologres テーブルからの読み取りと書き込み
Hologres テーブルからデータを読み取り、結果テーブルにデータを書き込みます。
デフォルトでは、Flink は Hologres ソーステーブルからバッチモードでデータを読み取り、リアルタイムで新しいデータを消費しません。Hologres データをリアルタイムで消費するには、次のいずれかの方法を使用します:
カタログ作成時に設定: SQL コマンドを使用してカタログを作成し、バイナリログを有効にして、バイナリログデータを消費します。詳細については、「リアルタイム消費の例」をご参照ください。その後、Hologres データを読み取ります。次の表にサンプルコードを示します。
構文
例
INSERT INTO ${other_sink_table} SELECT ... FROM `${catalog_name}`.`${db_name}`.`${schema_name.table_name}`;INSERT INTO sink_table SELECT id, name FROM `holocatalog`.`holodb`.`holotable`;読み取りモードをストリーミングに変更: テーブルヒントを使用して読み取りモードをストリーミングに変更します。SQL ステートメントに
/*+ OPTIONS('binlog'='true') */ヒントを追加します。次の表にサンプルコードを示します。INSERT INTO sinktable SELECT id, name FROM `holocatalog`.`holodb`.`holotable` /*+ OPTIONS ('binlog' = 'true') */;
ソーステーブルから Hologres テーブルにデータを書き込みます。
構文
例
INSERT INTO `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` SELECT ... FROM ${other_source_table}INSERT INTO `holocatalog`.`holodb`.`holotable` SELECT id, name FROM source_table;
CTAS の宛先カタログとして使用
構文
CREATE TABLE IF NOT EXISTS `${catalog_name}`.`${db_name}`.`${schema_name.table_name}` WITH ( 'connector' = 'hologres' ) AS TABLE ${other_source_table};例
CREATE TABLE IF NOT EXISTS `holocatalog`.`holodb`.`holotable` WITH ( 'connector' = 'hologres' ) AS TABLE source_table;
CTAS を使用すると、WITH 句で物理テーブルのプロパティを設定できます。宛先テーブルが作成されると、対応するプロパティがテーブルに設定されます。サポートされているテーブルプロパティパラメーターの詳細については、「Hologres テーブルの作成」をご参照ください。
ソースからのデータ同期中に、Hologres カタログは、データが Hologres に書き込めるように、宛先スキーマを強制的に再書き込みする場合があります。これは、次の場合に発生します:
アップストリームスキーマが主キーとして DECIMAL 型の列を使用している。
Hologres は主キーとして DECIMAL 型をサポートしていないため、列の型はデフォルトで BIGINT に再書き込みされます。この動作が要件を満たさない場合は、CTAS 構文を使用して型を STRING に変換し、主キーを再確立できます。
アップストリームスキーマに、精度が 6 を超える TIME、TIMESTAMP、または TIMESTAMP_LTZ 型が含まれている。
Hologres がサポートする時間型の精度は 6 であるため、Flink はデータが Hologres に書き込めるように、それを超える精度を暗黙的に破棄します。
CDAS の宛先カタログとして使用
構文
CREATE DATABASE IF NOT EXISTS `${catalog_name}`.`${db_name}` WITH ( 'sink.parallelism' = '5' -- 各結果テーブルの同時実行数を設定します。 ) AS DATABASE ${other_source_database};例
CREATE DATABASE IF NOT EXISTS `holocatalog`.`holodb` WITH ( 'sink.parallelism' = '5' -- 各結果テーブルの同時実行数を設定します。 ) AS DATABASE source_database;
WITH パラメーターの設定:
結果テーブルのパラメーターを宣言: ジョブが開始されると、これらのパラメーターは同期が必要なダウンストリームテーブルに適用されます。サポートされているパラメーターの詳細については、「Hologres 結果テーブル」をご参照ください。
schemaname の指定: 宛先の Hologres データベースの指定されたスキーマにデータを同期します。次の表にパラメーターを説明します。
パラメーター
説明
必須
備考
schemaname
スキーマ名。
いいえ
デフォルト値は public です。
物理テーブルプロパティの設定: 各宛先テーブルには、異なるテーブルプロパティが必要な場合があります。WITH 句は、各テーブルのプロパティを個別に設定することをサポートしていません。
テーブルプロパティを設定するには、CDAS ジョブを開始する前に宛先テーブルを手動で作成する必要があります。物理テーブルプロパティの詳細については、「Hologres カタログの作成」をご参照ください。
Hologres カタログの削除
Hologres カタログを削除しても、実行中のジョブには影響しません。ただし、まだ公開されていないジョブや、一時停止して再開する必要があるジョブには影響します。この操作は注意して実行してください。
UI の使用
データ管理ページに移動します。
Realtime Compute コンソールにログインします。
対象のワークスペースの [アクション] 列で、[コンソール] をクリックします。
[データ管理] をクリックします。
[カタログリスト] ページで、対象のカタログを見つけ、[アクション] 列の [削除] をクリックします。
表示されるダイアログボックスで、[削除] をクリックします。
削除が完了したら、左側の [メタデータ] セクションから対象のカタログが削除されていることを確認します。
SQL コマンドの使用
[データクエリ] ページのエディターで、次のコマンドを入力します。
DROP CATALOG ${catalog_name}コマンドの ${catalog_name} は、Realtime Compute 開発コンソールに表示される、削除したい Hologres カタログの名前です。
カタログを削除するコマンドを右クリックし、[実行] を選択します。
左側の [メタデータ] セクションで、対象のカタログが削除されていることを確認します。
よくある質問
Hologres からのリアルタイムデータ消費: Hologres カタログを使用して Hologres ソーステーブルからデータを読み取ると、ジョブのステータスが「完了」と表示されます。リアルタイムでデータを消費するためにジョブを実行し続けるにはどうすればよいですか?
ネットワーク問題のトラブルシューティング: Hologres インスタンスのエンドポイントを取得するにはどうすればよいですか?
アップストリームおよびダウンストリームのストレージ制限: エラー: CREATE TABLE ... AS TABLE ... ステートメントには、ターゲットカタログ ... が org.apache.flink.table.catalog.CatalogTableProvider インターフェイスを実装する必要があります。
リファレンス
Hologres WITH パラメーター: WITH パラメーター
Hologres カタログのシナリオ: