Hologres:Use fully managed Flink の使用

Realtime Compute for Apache Flink に SQL ドラフトをデプロイする場合、SQL デプロイで Hologres のソーステーブル、ディメンションテーブル、または結果テーブルを使用するかどうかに関係なく、Realtime Compute for Apache Flink でサポートされているデータ型を定義する必要があります。Hologres で内部テーブルを作成する場合にのみ、Hologres でサポートされているデータ型を定義できます。

Realtime Compute for Apache Flink を使用して JSON データを Hologres に書き込む場合は、SQL デプロイのソーステーブルと結果テーブルの列のデータ型を VARCHAR 型に定義し、Hologres 内部テーブルの列のデータ型を JSONB 型に定義する必要があります。

• Hologres 内部テーブル：message のデータ型を JSONB に設定します。

  BEGIN ;
  DROP TABLE IF EXISTS holo_internal_table;
  CREATE TABLE IF NOT EXISTS holo_internal_table  -- Hologres の内部テーブルを作成する場合のコメント
  (
      id BIGINT NOT NULL,
      message JSONB NOT NULL
  );
  CALL set_table_property('holo_internal_table', 'distribution_key', 'id');
  COMMIT ;

• Flink デプロイ：ソーステーブルと結果テーブルの message のデータ型を VARCHAR に設定します。

  CREATE TEMPORARY TABLE randomSource (
      id BIGINT,
      message VARCHAR
    )
  WITH ('connector' = 'datagen');
  
  CREATE TEMPORARY TABLE sink_holo (
      id BIGINT,
      message VARCHAR
    )
  WITH (
      'connector' = 'hologres',
      'endpoint' = '',
      'username' = '',
      'password' = '',
      'dbname' = '',
      'tablename' = 'holo_internal_table'
    );
  
  INSERT INTO sink_holo
  SELECT
    1,
    '{"k":"v"}'
  FROM
    randomSource;