Flink全托管

不论是将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;