不論是將Hologres作為源表、維表還是結果表，在Flink的SQL作業中，都要將數據定義為Flink的數據類型。只有在Hologres中創建內部表時，才需要定義為Hologres的數據類型。

如果期望使用Flink將JSON數據寫入Hologres，則在Flink 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;