Hologres:Hologres結果テーブル

制限事項

• Blink排他モードのバージョンによって、サポートされる開発構文が異なります。 Realtime Compute for Apache Flinkを使用する前に、Blink排他モードのバージョンを確認し、そのバージョン用に提供されているサンプルコードを参照する必要があります。

• Realtime ComputeサービスとHologresサービスが同じリージョンにあることを確認してください。 そうでない場合、サービスは相互に接続できません。

• V3.6より前のバージョンのBlink排他モードでは、組み込みのHolo-blinkコネクタは提供されていません。 Hologres結果テーブルを作成するには、関連するJARファイルを参照する必要があります。 HologresコンソールでHologresインスタンスを手動でアップグレードするか、Hologres DingTalkグループに参加してインスタンスのアップグレードを申請できます。 Hologresインスタンスを手動でアップグレードする方法の詳細については、 インスタンスのアップグレード をご参照ください。DingTalkグループへの参加方法の詳細については、Hologresのオンラインサポートを受ける をご参照ください。

  説明

  Blink排他モードをV3.6以降にアップグレードすることをお勧めします。

• Blink V3.7排他モードでは、ジョブのコードに createparttable='true' と入力した後でのみ、Hologresパーティションテーブルを自動的に作成できます。 パーティションテーブルを使用する場合は、次の点に注意してください。

  • Hologresテーブルは、リストパーティションのみをサポートしています。

  • パーティションテーブルを作成する場合は、パーティションキー列を指定する必要があります。 パーティションキー列のデータは、TEXT または INT4 タイプである必要があります。 パーティションキー値には、ハイフン（-）を含めることはできません。 たとえば、パーティションキー値 2020-09-12 は無効です。

  • パーティションテーブルにプライマリキーを設定する場合は、プライマリキーにパーティションキー列を含める必要があります。

  • 子テーブルを作成する場合は、パーティションキー値として固定値を指定する必要があります。

  • 子テーブルにデータを書き込む場合は、ソースデータで指定された列の値が、子テーブルに指定したパーティションキー値の範囲内にあることを確認してください。 そうでない場合、エラーメッセージが返されます。

  • Hologresテーブルは、default パーティションをサポートしていません。

• データの書き込み先のHologresテーブルにプライマリキーを設定した場合、リアルタイムのデータ書き込みのデフォルト構文はプライマリキーに基づいて更新されません。 書き込もうとする新しいデータが既存のデータと同じプライマリキー値を共有している場合、新しいデータは破棄されます。

• Hologresはデータを非同期で書き込みます。 そのため、コードに blink.checkpoint.fail_on_checkpoint_error=true と入力する必要があります。 これにより、ジョブの例外が発生した場合にのみフェイルオーバーがトリガーされます。 Blink V3.7.6以降の排他モードを使用する場合は、この設定をコードに追加する必要はありません。

DDL構文

create table Hologres_sink(
  name varchar,
  age BIGINT,
  birthday BIGINT
) with (
  type='hologres',
  dbname='<yourDbname>', -- 接続先のHologresデータベースの名前。
  tablename='<yourTablename>', -- データを書き込むHologresテーブルの名前。
  username='<yourUsername>', -- Alibaba CloudアカウントのAccessKey ID。
  password='<yourPassword>', -- Alibaba CloudアカウントのAccessKeyシークレット。
  endpoint='<yourEndpoint>'); -- Hologresインスタンスの仮想プライベートクラウド（VPC）エンドポイント。

WITH句のパラメーター

標準のHologres結果テーブルにリアルタイムでデータを書き込む

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

   データを書き込むHologresテーブルを作成します。 ステートメント例:

    create table blink_test (a int, b text, c text, d float8, e bigint);

2. リアルタイムコンピューティングジョブを作成します。

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

   2. リアルタイムコンピューティングジョブを作成します。

      • Hologresデータソースは、Blink V3.6以降の排他モードでサポートされています。 リアルタイムコンピューティングジョブを作成するときに、Hologresデータソースを呼び出すことができます。 ステートメント例:

        create table randomSource (a int, b VARCHAR, c VARCHAR, d DOUBLE, e BIGINT) with (type = 'random');
        
        create table test (
          a int,
          b VARCHAR,
          c VARCHAR,
          PRIMARY KEY (a)
        ) with (
          type = 'hologres',
          `endpoint` = '$ip:$port', -- HologresインスタンスのVPCエンドポイントとポート番号。
          `username` = 'Alibaba CloudアカウントのAccessKey ID',
          `password` = 'Alibaba CloudアカウントのAccessKeyシークレット',
          `dbname` = '接続先のHologresデータベースの名前',
          `tablename` = 'blink_test'-- データを書き込むHologresテーブルの名前。
        );
        
        insert
          into test
        select
          a,b,c
        from
          randomSource;

3. ジョブを発行します。

   1. ジョブコードを編集した後、ジョブエディターの上部にある [構文チェック] をクリックします。 [成功] メッセージが表示された場合、コードの構文は正しいです。

   2. [保存] をクリックします。

   3. [発行] をクリックします。 表示されるダイアログボックスで、ジョブを本番環境に発行するためのパラメーターを設定します。

4. ジョブを開始します。

   ジョブが本番環境に発行されたら、ジョブを手動で開始します。

   [開発プラットフォーム] ページの上部ナビゲーションバーで、右上隅にある [管理] をクリックします。 [管理] ページで、開始するジョブを選択し、右上隅にある [開始] をクリックします。

5. Hologresのデータをリアルタイムでクエリします。

   手順 1 で作成したテーブルに書き込まれたデータをクエリできます。 ステートメント例:

   select * from blink_test;

ワイドテーブルにデータをマージする

このセクションでは、複数のデータストリームから同じHologresワイドテーブルにデータを書き込む方法について説明します。

Hologresのパーティション結果テーブルにリアルタイムでデータを書き込む

HoloHub APIを呼び出して、親テーブルにデータを書き込むことができます。 その後、システムは異なるパーティションのデータを対応する子テーブルにルーティングします。 パーティションテーブルにデータを書き込むこともできます。 HoloHub APIの詳細については、HoloHub APIの概要 をご参照ください。

1. Hologresにパーティションテーブルを作成します。

   データを書き込むHologresにパーティションテーブルを作成し、パーティションテーブルに対応する子テーブルを作成します。 ステートメント例:

   -- test_messageという名前の親テーブルと、親テーブルに対応する子テーブルを作成します。
   
   drop table if exists test_message;
   
   begin;
   create table test_message (
    "bizdate" text NOT NULL,
    "tag" text NOT NULL,
    "id" int4 NOT NULL,
    "title" text NOT NULL,
    "body" text,
   PRIMARY KEY (bizdate,tag,id)
   )
   PARTITION BY LIST (bizdate);
   commit;

   説明

   • ステートメントを実行するときは、${bizdate} パラメーターに値を割り当てます。

   • Blink V3.7排他モードでのみ、子テーブルを自動的に作成できます。 Blink排他モードのバージョンがV3.7より前の場合は、データをインポートする前に、Hologresで手動で子テーブルを作成する必要があります。 そうでない場合、データのインポートは失敗します。

2. Blink排他モードを使用する場合にリアルタイムコンピューティングジョブを作成します。

   この例では、次のステートメントを実行して、Blink排他モードを使用する場合にリアルタイムコンピューティングジョブを作成します。

   説明

   次のステートメント例は、Blink V3.7以降の排他モードに適用されます。 Blink排他モードのバージョンがV3.7より前の場合は、Blink排他モードをV3.7以降にアップグレードするか、子テーブルを自動的に作成するために使用されるコードから `createparttable` = 'true' を削除します。

   create table test_message_src(
     tag VARCHAR,
     id INTEGER,
     title VARCHAR,
     body VARCHAR
   ) with (
     type = 'random',
     `interval` = '10',
     `count` = '100'
   );
   
   create table test_message_sink (
     bizdate VARCHAR,
     tag VARCHAR,
     id INTEGER,
     title VARCHAR,
     body VARCHAR
   ) with (
     type = 'hologres',
     `endpoint` = '$ip:$port', -- HologresインスタンスのVPCエンドポイント。
     `username` ='<AccessID>', -- Alibaba CloudアカウントのAccessKey ID。
     `password` = '<AccessKey>', -- Alibaba CloudアカウントのAccessKeyシークレット。
     `dbname` = '<DBname>', -- 接続先のHologresデータベースの名前。
     `tablename` = '<Tablename>', -- データを書き込むHologresテーブルの名前。
     `partitionrouter` = 'true', -- Hologresのパーティションテーブルにデータを書き込みます。
     `createparttable` = 'true', -- Hologresに子テーブルを自動的に作成します。
   );
   
   insert into test_message_sink select "20200327",* from test_message_src;
   insert into test_message_sink select "20200328",* from test_message_src;

3. ジョブを発行して開始します。

   詳細については、「ジョブを発行する」および「ジョブを開始する」の手順について、「標準のHologres結果テーブルにリアルタイムでデータを書き込む」セクションをご参照ください。

4. Hologresのデータをリアルタイムでクエリします。

   手順 1 で作成した親テーブルと子テーブルに書き込まれたデータをクエリできます。 ステートメント例:

   select * from test_message;
   select * from test_message where bizdate = '20200327';

データ型のマッピング

Blink排他モードとHologresのデータ型間のマッピングの詳細については、データ型 をご参照ください。