Hologres:Hologres 結果テーブル

制限事項

• Real-time Compute for Flink の排他モードのバージョンによって、開発者向けのセマンティクスが異なります。作業を開始する前に、ご利用のバージョンを確認し、該当バージョンの例を参照してください。

• 接続エラーを防ぐため、Real-time Compute for Flink サービスと Hologres インスタンスが同一リージョンにあることを確認してください。

• Real-time Compute for Flink の排他モードでバージョン 3.6 より前のものには、組み込みの Hologres コネクタが含まれていません。Hologres にリアルタイムでデータを書き込むには、JAR ファイルを参照する必要があります。トラブルシューティングについては、「一般的なスペックアップ準備失敗エラー」をご参照ください。フィードバックについては、Hologres DingTalk グループにご参加いただけます。グループへの参加方法の詳細については、「オンラインサポートをさらに利用する方法」をご参照ください。

  説明

  ジョブを実行するには、バージョン 3.6 以降へのスペックアップを推奨します。

• Real-time Compute for Flink の排他モード バージョン 3.7 では、Hologres パーティションテーブルの自動作成がサポートされています。ただし、ジョブ内で createparttable='true' を構成する必要があります。パーティションテーブルを使用する場合、以下の点にご注意ください。

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

  • パーティションテーブルを作成する際は、パーティションキー列を明示的に指定する必要があります。現在、パーティションキー列としてサポートされているのは text 型および int4 型のみです。パーティション値にはハイフン (-) を含めることはできません（例：2020-09-12）。

  • パーティションテーブルにプライマリキーが設定されている場合、パーティションキー列はプライマリキーの一部である必要があります。

  • 子パーティションテーブルを作成する際、子テーブルのパーティションキー列には静的フィールド値を指定する必要があります。

  • 子パーティションテーブルに書き込まれるデータのパーティションキー列の値は、子テーブル作成時に定義された値と完全に一致している必要があります。一致しない場合はエラーが報告されます。

  • DEFAULT パーティション機能はサポートされていません。

• 宛先の Hologres テーブルにプライマリキーが設定されている場合、リアルタイム書き込みのデフォルトセマンティクスでは、プライマリキーに基づくデータ更新は行われません。重複するプライマリキーを持つデータは破棄されます。

• Hologres は非同期でデータを書き込みます。ジョブに blink.checkpoint.fail_on_checkpoint_error=true 構成を追加する必要があります。これにより、ジョブが失敗した際にフェールオーバーがトリガーされます。このパラメーターは、Real-time Compute for Flink の排他モード バージョン 3.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 Secret。
  endpoint='<yourEndpoint>'); --Hologres インスタンスの VPC ネットワークのエンドポイント。

WITH パラメーター

通常の Hologres 結果テーブルへのリアルタイムデータ書き込み

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

   データを受信するためのテーブルを Hologres に作成します。以下はテーブル作成 SQL 文の例です。

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

2. Real-time Compute for Flink ジョブを作成します。

   1. Real-time Compute for Flink コンソール にログインします。

   2. Real-time Compute for Flink ジョブを作成します。

      • Real-time Compute for Flink の排他モード V3.6 以降では、Hologres データソースがサポートされています。直接呼び出すことができます。以下は SQL 文の例です。

        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 Secret',
          `dbname` = 'Hologres データベースの名前',
          `tablename` = 'blink_test'--Hologres でデータを受信するテーブルの名前。
        );
        
        insert
          into test
        select
          a,b,c
        from
          randomSource;

3. ジョブを公開します。

   1. ジョブを作成後、エディターで 構文チェック をクリックします。成功しました と表示された場合、構文は正しいです。

   2. 保存 をクリックしてジョブを保存します。

   3. 公開 をクリックして、ジョブを本番環境に提出します。必要に応じて、ジョブ公開設定を構成してください。

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

   ジョブを本番環境に提出した後、手動で開始する必要があります。

   Real-time Compute for Flink 開発者プラットフォーム ページのメニューバー右上隅で、O&M をクリックします。O&M ページで開始したいジョブを選択し、右上隅の 起動 をクリックします。

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

   Hologres でデータを受信するテーブルをクエリし、書き込まれたデータをリアルタイムで取得します。以下はクエリ SQL 文の例です。

   select * from blink_test;

ワイドテーブルのマージ／部分更新機能の使用方法

複数のストリームから単一の Hologres ワイドテーブルにデータを書き込む（一般的なシナリオ）には、以下の手順に従います。

パーティション Hologres 結果テーブルへのリアルタイムデータ書き込み

Hologres は、リアルタイムデータ API を呼び出すことで、親パーティションテーブルに直接データを書き込むことをサポートしています。データは自動的に対応する子パーティションテーブルにルーティングされます。リアルタイムデータ API の詳細については、「リアルタイムデータ API」をご参照ください。

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

   データを受信するためのパーティションテーブルを Hologres に作成し、対応する子パーティションテーブルを作成します。以下はテーブル作成 SQL 文の例です。

   --親パーティションテーブル 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} パラメーターを実際の値に置き換えてください。

   • Real-time Compute for Flink の排他モード バージョン 3.7 が、自動パーティション作成をサポートする最初のバージョンです。バージョン 3.7 より前のものを使用している場合、事前に Hologres で子パーティションテーブルを作成しておく必要があります。そうしないと、データインポートが失敗します。

2. Real-time Compute for Flink の排他モードでジョブを作成します。

   以下は、Real-time Compute for Flink の排他モードでジョブを作成するための文の例です。

   説明

   以下の例は、Real-time Compute for Flink の排他モード V3.7 以降に適用されます。バージョン 3.7 より前のものを使用している場合は、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 Secret。
     `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 のデータをリアルタイムでクエリします。

   Hologres でデータを受信するテーブルをクエリし、書き込まれたデータをリアルタイムで取得します。以下はクエリ SQL 文の例です。

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

データ型のマッピング

Real-time Compute for Flink の排他モードと Hologres 間のデータ型マッピングの詳細については、「データ型まとめ」をご参照ください。