Hologres:Hologres シンクテーブル

制限事項

• 専用モードの Blink は、バージョンによって開発構文が異なります。開始する前に、お使いの専用モードの Blink のバージョンを特定し、対応する例を使用してください。

• 接続エラーを回避するために、Realtime Compute for Apache Flink のデプロイメントと Hologres のインスタンスが同じリージョンにあることを確認してください。

• バージョン 3.6 より前の専用モードの Blink には、Hologres コネクタが組み込まれていません。Hologres にリアルタイムでデータを書き込むには、JAR ファイルを参照する必要があります。サポートが必要な場合は、「アップグレード準備中の一般的なエラー」をご参照いただくか、Hologres DingTalk グループに参加してお問い合わせください。詳細については、「オンラインサポートの利用方法」をご参照ください。

  説明

  ジョブを実行するには、専用モードの Blink をバージョン 3.6 以降にアップグレードすることを推奨します。

• Blink 専用モード 3.7 は、Hologres パーティションテーブルの自動作成をサポートしていますが、ジョブで createparttable='true' を設定する必要があります。パーティションテーブルを使用する場合の考慮事項は次のとおりです。

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

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

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

  • 子パーティションテーブルを作成する際、そのパーティションキー列の値は固定値である必要があります。

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

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

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

• Hologres はデータを非同期に書き込みます。ジョブで例外が発生した場合にフェイルオーバーがトリガーされるように、ジョブに blink.checkpoint.fail_on_checkpoint_error=true 設定を追加する必要があります。このパラメーターは、Blink 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. Realtime Compute for Apache Flink ジョブを作成します。

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

   2. ジョブを作成します。

      • 専用モードの Blink バージョン 3.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. ジョブを開始します。

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

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

5. Hologres 内のデータをリアルタイムでクエリする

   Hologres の宛先テーブルをクエリして、書き込まれたデータをリアルタイムで表示します。以下はクエリの例です。

   select * from blink_test;

ワイドテーブルのマージと更新

このセクションでは、複数のストリームからのデータを単一の Hologres ワイドテーブルに書き込むという、一般的なユースケースについて説明します。

パーティション化された Hologres シンクテーブルへの書き込み

Hologres では、HoloHub API を呼び出して、親パーティションテーブルに直接データを書き込むことができます。データは、その後、正しい子パーティションテーブルに自動的にルーティングされます。詳細については、「HoloHub 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} パラメーターを実際の値に置き換えてください。

   • パーティションの自動作成をサポートしているのは、バージョン 3.7 以降の専用モードの Blink のみです。以前のバージョンを使用している場合は、事前に Hologres で子パーティションテーブルを作成する必要があります。そうしないと、データのインポートに失敗します。

2. Blink で排他モードでジョブを作成する

   以下は、専用モードの Blink でジョブを作成するためのステートメントの例です。

   説明

   以下の例は、Blink exclusive mode 3.7 以降に適用されます。Blink exclusive mode 3.7 より前のバージョンを使用している場合は、バージョン 3.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 の宛先テーブルをクエリして、書き込まれたデータをリアルタイムで表示します。以下はクエリの例です。

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

データ型のマッピング

専用モードの Blink と Hologres 間のデータ型のマッピングについては、「データ型の概要」をご参照ください。