すべてのプロダクト
Search

このプロダクト

    Realtime Compute for Apache Flink:Hologres カタログの管理

    ドキュメントセンター

    Realtime Compute for Apache Flink:Hologres カタログの管理

    最終更新日:Sep 19, 2026

    Hologres カタログを設定すると、テーブルを手動で登録しなくても、Realtime Compute for Apache Flink コンソールから Hologres のメタデータを直接読み取れます。この機能により、ジョブ開発の効率を向上させ、データ精度を確保できます。このトピックでは、Hologres カタログの設定、表示、使用、削除の方法について説明します。

    前提条件

    データベースを含む専用 Hologres インスタンスが必要です。詳細については、「データベースの作成」をご参照ください。

    制限事項

    • カタログは変更できません。変更を適用するには、既存のカタログを削除してから新しいカタログを作成する必要があります。

    • Hologres 専用インスタンスを使用する必要があります。Realtime Compute for Apache Flink は Hologres 内部テーブルにのみアクセスできるため、共有クラスターインスタンスをサポートしていません。

    Hologres カタログの作成

    重要

    カタログの設定は、作成後に変更できません。変更を加えるには、既存のカタログをドロップしてから新しいカタログを作成する必要があります。

    UI

    Hologres コネクタがサポートするパラメータを設定する必要がある場合は、SQL を使用して Hologres カタログを作成します。

    1. データ管理ページに移動します。

      1. Realtime Compute for Apache Flink コンソールにログインし、対象のワークスペースの[操作] 列で [コンソール] をクリックします。

      2. [カタログ] をクリックします。

    2. [カタログの作成] をクリックし、[Hologres] を選択して、[次へ] をクリックします。

    3. パラメータを設定します。

      パラメータ

      説明

      必須

      備考

      catalogname

      Hologres カタログの名前。

      はい

      小文字 (a-z) と数字 (0-9) のみ使用できます。大文字、ハイフン (-)、アンダースコア (_)、およびその他の特殊文字は使用できません。

      endpoint

      Hologres インスタンスのネットワークエンドポイント。

      はい

      Hologres インスタンスと Flink ワークスペースが同じ VPC 内にある場合は、[Hologres コンソール]に移動します。対象インスタンスの詳細ページで、[ネットワーク情報] セクションから [指定された VPC] のネットワーク情報を取得します。

      その他のネットワークタイプについては、「Hologres インスタンスのエンドポイントの取得」をご参照ください。

      username

      • カスタムアカウントのユーザー名。形式は BASIC$<user_name> です。

      • Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。

      はい

      • 設定するユーザーは、対応する Hologres データベースへのアクセス権限を持っている必要があります。詳細については、「Hologres 権限モデル」および「ユーザー管理」をご参照ください。

      • AccessKey ID と AccessKey Secret の取得方法については、「AccessKey ペアの取得」をご参照ください。

        重要

        • AK やパスワードの漏洩を防ぐため、プロジェクト変数を使用して値を入力することを推奨します。詳細については、「プロジェクト変数」をご参照ください。

        • カスタムアカウントで作成されたカタログには、そのアカウントが権限を持つデータベースのみが表示されます。AccessKey で作成されたカタログには、インスタンス内のすべてのデータベースが表示されます。

      password

      • カスタムアカウントのパスワード。

      • Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret。

      はい

      dbname

      Hologres データベースの名前。

      はい

      対象の Hologres インスタンス内にデータベースが既に存在している必要があります。存在しない場合、カタログの作成は失敗します。

    4. [OK] をクリックします。

      作成が完了すると、[カタログ] で新しいカタログを表示できます。

    SQL

    1. データクエリ ページのエディタで、Hologres カタログを作成するステートメントを入力します。

      構文：

      CREATE CATALOG <catalogname> WITH (
  'type' = 'hologres',
  'endpoint' = '<endpoint>', 
  'username' = '<AccessKey ID>',
  'password' = '<AccessKey Secret>',
  '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'
);

      Binlog データを利用するには、Binlog を有効にする必要があります。

      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 テーブルプロパティを渡すこともできます。これにより、新しいテーブルに対してデフォルトで binlog が有効になります。
  'table_property.binlog.ttl' = '259200'
);

      以下の表にパラメータを示します。

      パラメータ

      説明

      必須

      備考

      catalogname

      Hologres カタログの名前。

      はい

      小文字 (a-z) と数字 (0-9) のみ使用できます。大文字、ハイフン (-)、アンダースコア (_)、およびその他の特殊文字は使用できません。

      type

      カタログのタイプ。

      はい

      値は hologres である必要があります。

      endpoint

      Hologres インスタンスのエンドポイント。

      はい

      Hologres インスタンスと Flink ワークスペースが同じ VPC 内にある場合は、[Hologres 管理コンソール]に移動し、対象インスタンスの詳細ページにある[ネットワーク情報]セクションから[指定された VPC] のネットワーク情報を取得します。

      その他のネットワークタイプについては、「Hologres インスタンスのエンドポイントの取得」をご参照ください。

      username

      Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ID。

      はい

      詳細については、「AccessKey ペアの取得」をご参照ください。

      説明

      • AK 情報の漏洩を防ぐため、password の値には変数を使用することを推奨します。詳細については、「プロジェクト変数」をご参照ください。

      • 指定された AccessKey に対応するユーザーは、Hologres データベースへのアクセス権限を持っている必要があります。データベース権限の詳細については、「Hologres 権限モデル」をご参照ください。

      password

      Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey Secret。

      はい

      dbname

      Hologres データベースの名前。

      はい

      対象の Hologres インスタンス内にデータベースが既に存在している必要があります。存在しない場合、カタログの作成は失敗します。

      ignore-non-persisted-options

      カタログを通じて非永続化オプションを使用してテーブルを作成する際の動作を指定します。

      いいえ

      有効な値：

      • true (デフォルト)： テーブルが作成され、すべての非永続化オプションは無視されます。

      • false： エラーが発生し、テーブルの作成は失敗します。

      説明

      永続化テーブルオプションとは、DDL ステートメントで定義された値が Hologres カタログに保存され、後で取得できるオプションです。現在、endpoint、username、password、dbname のオプションのみが永続化されます。

      catalog.table.metadata-columns

      このカタログを使用してソーステーブルを指定する際に、必要に応じて Hologres Binlog ソーステーブルのメタデータ列をテーブルスキーマに追加できます。複数のメタデータ列はセミコロン (;) で区切ります。例： hg_binlog_event_type;hg_binlog_timestamp_us。

      いいえ

      デフォルトでは、メタデータ列は追加されません。このパラメータを設定すると、指定されたメタデータ列が返されるテーブルスキーマに追加されます。これらの列は Hologres Binlog ソーステーブルにのみ適用されます。したがって、このカタログのテーブルはソーステーブルとしてのみ使用でき、シンクテーブルやディメンションテーブルとしては使用できません。6 種類のメタデータ列がサポートされています。詳細については、「Hologres Binlog フィールド」をご参照ください。

      説明

      このパラメータは Ververica Runtime (VVR) 8.0.11 以降でのみサポートされています。

      Hologres コネクタがサポートするその他のパラメータ

      カタログを作成する際に、WITH オプション を含むその他のパラメータを指定できます。これらのパラメータは、このカタログを通じてアクセスされるすべてのテーブルのデフォルト設定になります。

      いいえ

      この機能を使用するには、ignore-non-persisted-options パラメータを true に設定します。

    2. カタログ作成ステートメントを入力したら、右上隅の[実行]をクリックします。

    Hologres カタログ

    Hologres カタログを設定した後、次の手順に従ってそのメタデータを表示します。

    1. カタログページに移動します。

      1. Realtime Compute for Apache Flink コンソールにログインします。

      2. 対象のワークスペースの [Actions] 列で、[Console] をクリックします。

      3. ナビゲーションペインで、[Catalogs] をクリックします。

    2. [Catalog List] ページで、[Name] と [Type] でカタログを検索します。

      [View] をクリックして、カタログ内のデータベースとテーブルを表示します。スキーマが public の場合、テーブル名はスキーマプレフィックスなしで表示されます。

    Hologres カタログの使用

    注：

    • スキーマが public の場合、${schema_name}.${table_name} を入力する際に、スキーマ名のプレフィックスを省略して、テーブル名を直接 ${table_name} として入力できます。

    • Hologres カタログ内のテーブルは UPDATE データを消費できます。これらのテーブルでは、ignoredelete プロパティはデフォルトで false に、mutatetype プロパティはデフォルトで insertorupdate になります。これらのプロパティの詳細については、「ワイドテーブルへのデータのマージとデータの部分的な更新」をご参照ください。

    Hologres テーブルの作成

    この例では、holocatalog カタログの holodb データベースに holotable という名前のテーブルを作成します。

    説明

    • 登録済みの Hologres データサービスでテーブルを作成する場合、WITH 句で connector パラメーターが必須となり、その値は hologres である必要があります。endpoint などの他のパラメーターは省略できます。

    • Hologres テーブルでは、サポートされている WITH パラメーターを直接追加または変更できません。これらのパラメーターを追加または変更するには、INSERT ステートメントで SQL ヒント を使用します。

    UI

    1. [Catalogs] ページに移動します。

      1. Realtime Compute for Apache Flink コンソール にログインします。

      2. 対象のワークスペースの [Actions] 列で、[Console] をクリックします。

      3. [Catalogs] をクリックします。

    2. 対象のデータベースに移動します。

      1. 対象のカタログの [Actions] 列で、[View] をクリックします。

      2. 対象のデータベースの [Actions] 列で、[View] をクリックします。

    3. Hologres テーブルを作成します。

      1. [Create Table] をクリックします。

      2. [Built-in] タブで、Hologres コネクタを選択し、[Next] をクリックします。

      3. CREATE TABLE ステートメントを入力し、パラメーターを設定します。次のコードは一例です。

        構文

        例

        		 
        CREATE TABLE `${catalog_name}`.`${db_name}`.`${table_name}` (
 ...
) WITH (
 'connector' = 'hologres'
);
        		 
        CREATE TABLE `holocatalog`.`holodb`.`product` (
 id INT,
 name STRING
) WITH (
 'connector' = 'hologres'
);

      4. [OK] をクリックします。

    SQL

    1. データクエリページのエディターで、CREATE TABLE ステートメントを入力します。

      次のいずれかの方法で Hologres テーブルを作成できます：

      USE CATALOG

      USE ステートメントを使用して Hologres カタログを参照します。

      構文

      例

      		 
      USE CATALOG ${catalog_name};
CREATE TABLE `${db_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}`.`${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'
);

    2. ステートメントを入力した後、右上隅にある [Run] をクリックしてテーブルを作成します。

    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'

      table_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

      テーブルデータの Time to Live (TTL) を指定します。

      'table_property.time_to_live_in_seconds' = '86400'

      table_property.binlog.level

      バイナリログを有効にするかどうかを指定します。

      'table_property.binlog.level' = 'replica'

      table_property.binlog.ttl

      バイナリログの TTL を指定します。

      'table_property.binlog.ttl' = '86400'

    • 寛容モードパラメーター：Hologres カタログでテーブルを作成する際に寛容モードを有効にするには、enableTypeNormalization パラメーターを設定します。Flink CDC データインジェストジョブでは、sink.type-normalize-strategy を設定して寛容モードを有効にします。

      項目

      説明

      ユースケース

      データ同期のシナリオでは、既存のフィールドのデータ型の精度を調整 (例：VARCHAR(10) から VARCHAR(20) へ) したり、データ型を変更 (例：SMALLINT から INT へ) したりする必要がある場合があります。

      注意

      • 初回実行時に寛容モードを有効にする必要があります。初回実行時に寛容モードを有効にしなかった場合、変更を有効にするには、ダウンストリームテーブルを削除し、ジョブのステートレス再起動を実行する必要があります。

      • 寛容モードを有効にした後、アップストリームのデータ型が変更されても、新しい型と元の型が同じ正規化された型を共有している限り、ジョブは実行を継続します。それ以外の場合、型は互換性がないと見なされ、ジョブは例外をスローします。

      値

      • false (デフォルト) ：標準の型マッピングに基づいて Hologres 物理テーブルを作成します。

      • true：寛容モードが有効になります。Hologres 物理テーブルが作成されるとき、型の正規化の後、より高い精度のデータ型が使用されます。現在の型の正規化ルールは次のとおりです：

        • TINYINT、SMALLINT、INT、および BIGINT は BIGINT に正規化されます。

        • CHAR、VARCHAR、および STRING は STRING に正規化されます。

        • FLOAT と DOUBLE は DOUBLE に正規化されます。

        • 他のデータ型は、元の型マッピングルールに基づいて作成されます。詳細については、「型マッピング」をご参照ください。

    Hologres テーブルの変更

    Hologres カタログは、次のテーブル変更操作をサポートしています：

    アクション

    構文と例

    テーブルプロパティの変更

    一部のテーブルプロパティのみ変更できます。詳細については、「Hologres テーブルの作成」をご参照ください。

    • 構文：

      ALTER TABLE `${catalog_name}`.`${db_name}`.`${table_name}` SET (
  'table_property.binlog.level' = 'replica',
  'table_property.binlog.ttl' = '64700'
);

    • 例：

      ALTER TABLE `holocatalog`.`holodb`.`holotable` SET (
  'table_property.binlog.level' = 'replica',
  'table_property.binlog.ttl' = '64700'
);

    テーブル名の変更

    • 構文：

      ALTER TABLE `${catalog_name}`.`${db_name}`.`${table_name}` 
RENAME TO `${catalog_name}`.`${db_name}`.`${new_table_name}`;

    • 例：

      ALTER TABLE `holocatalog`.`holodb`.`holotable` 
RENAME TO `holocatalog`.`holodb`.`new_holotable`;

    列の追加

    • 構文：

      ALTER TABLE `${catalog_name}`.`${db_name}`.`${table_name}` 
ADD <column_name> <column_datatype> COMMENT '列コメント';

    • 例：

      ALTER TABLE `holocatalog`.`holodb`.`holotable` 
ADD address STRING COMMENT '住所情報';

    列名の変更

    • 構文：

      ALTER TABLE `${catalog_name}`.`${db_name}`.`${table_name}` 
RENAME <old_column_name> TO <new_column_name>;

    • 例：

      ALTER TABLE `holocatalog`.`holodb`.`holotable` 
RENAME address TO new_address;

    列コメントの変更

    • 構文：

      ALTER TABLE `${catalog_name}`.`${db_name}`.`${table_name}` 
MODIFY <column_name> <original_column_type> COMMENT '新しい列コメント';

    • 例：

      ALTER TABLE `holocatalog`.`holodb`.`holotable` 
MODIFY new_address STRING COMMENT '新しい住所情報';

    Hologres テーブルの読み取りと書き込み

    • Hologres テーブルからデータを読み取り、シンクテーブルに書き込みます。

      デフォルトでは、Flink は Hologres ソーステーブルをバッチモードで読み取り、新しいデータをリアルタイムで処理しません。Hologres からリアルタイムでデータをストリーミングするには、次のいずれかの方法を使用します：

      • 作成時に Hologres カタログを設定する：SQL を使用してカタログを作成する際に、バイナリログを有効にしてバイナリログデータを消費できます。詳細については、「リアルタイム消費の例」をご参照ください。その後、Hologres データを読み取ります。次のコードは一例です。

        構文

        例

        		 
        INSERT INTO ${other_sink_table}
SELECT ...
FROM `${catalog_name}`.`${db_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}`.`${table_name}`
SELECT ... 
FROM ${other_source_table}
      		 
      INSERT INTO `holocatalog`.`holodb`.`holotable`
SELECT id, name 
FROM source_table;

    Flink CDC データインジェストの宛先

    • 例

      source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog
  tables: tpcds.web_sales
  server-id: <server id>
  
sink:
  type: hologres
  using.built-in-catalog: hologres_catalog

    Flink CDC データインジェストジョブでは、シンクに物理テーブルプロパティを設定できます。これらのプロパティは、宛先テーブルが作成されるときに設定されます。サポートされているテーブルプロパティの詳細については、「Hologres テーブルの作成」をご参照ください。

    ソースからのデータ同期中に、Hologres カタログは、データが Hologres に書き込めるように、宛先テーブルのスキーマを書き換えることがあります。これは、次の場合に発生します：

    • ソーススキーマが DECIMAL 型の列をプライマリキーとして使用している。

      Hologres は DECIMAL をプライマリキーとしてサポートしていないため、デフォルトで列の型を BIGINT に書き換えます。この変更が要件を満たさない場合は、transform モジュールを使用して参照される列を STRING 型にキャストし、プライマリキーを再作成できます。

    • ソーススキーマに、精度が 6 を超える TIME、TIMESTAMP、または TIMESTAMP_LTZ 型の列が含まれている。

      Hologres は、時間型の精度を最大 6 までサポートします。データが Hologres に書き込めるように、Flink は Hologres がサポートする最大精度を超える桁を暗黙的に破棄します。

    Hologres カタログの削除

    重要

    Hologres カタログを削除しても、実行中のジョブには影響しませんが、未公開のジョブや一時停止と再開が必要なジョブには影響します。操作は慎重に行ってください。

    UI

    1. [Catalogs] ページに移動します。

      1. Realtime Compute for Apache Flink コンソールにログインします。

      2. 対象ワークスペースの [Actions] 列で、[Console] をクリックします。

      3. 左側のナビゲーションペインで、[Catalogs] をクリックします。

    2. [カタログ一覧ページ]で対象のカタログを見つけ、[Actions] 列の [Delete] をクリックします。

    3. 確認ダイアログで [Delete] をクリックします。

    4. 左側の [Catalogs] セクションに、対象のカタログが表示されなくなったことを確認します。

    Flink SQL

    1. データクエリページのテキストエディターで、次のコマンドを入力します。

      DROP CATALOG ${catalog_name}

      ここで、${catalog_name} は、Realtime Compute for Apache Flink コンソールに表示される、削除対象の Hologres カタログの名前です。

    2. コマンドを選択して右クリックし、[実行] を選択します。

    3. 左側の [Catalogs] セクションに、対象のカタログが表示されなくなったことを確認します。

    よくある質問

    関連ドキュメント