Hologres:Realtime Compute for Apache 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
  (
      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;