Hologres结果表

DDL定义

create table hologres_sink(
  name varchar,
  age BIGINT,
  birthday BIGINT
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',  --Hologres的数据库名称。
  'tablename'='<yourTablename>',  --Hologres用于接收数据的表名称。
  'username'='<yourUsername>',  --当前阿里云账号的AccessKey ID。
  'password'='<yourPassword>',  --当前阿里云账号的AccessKey Secret。
  'endpoint'='<yourEndpoint>',  --当前Hologres实例VPC网络的Endpoint。
);

流式语义

流处理，也称为流数据或流事件处理，即对一系列无界数据或事件连续处理。执行流数据或流事件处理的系统通常允许您指定一种可靠性模式或处理语义，保证整个系统处理数据的准确性，因为网络或设备故障等可能会导致数据丢失。

如何使用宽表Merge/局部更新功能

• 使用方法

  对于常见的多个流的数据写入至一张Hologres宽表的场景，具体使用方法如下：

  假设Hologres有一张宽表WIDE_TABLE，有A、B、C、D、E几列，其中A字段是主键，Flink一个流包含数据A、B、C，另一个流包含数据A、D、E。

  1. 使用Flink SQL声明两张Hologres结果表，其中一张表只声明字段A、B、C，另一张表只声明字段A、D、E，这两张表都映射至WIDE_TABLE。
  2. 两张结果表的mutatetype属性都设置成insertorupdate。
  3. 两张结果表的ignoredelete属性都设置成true，防止回撤消息产生Delete请求。
  4. 将两个流的数据分别Insert至两张结果表中。
• 使用限制
  该场景的具体使用限制如下：

  1. 宽表必须有主键。
  2. 每个流的数据都必须包含完整主键字段。
  3. 列存表的宽表Merge场景在高RPS的情况下，CPU使用率会偏高，建议关闭表中字段的Dictionary encoding。
• 使用示例
  Hologres建表语句如下，包含6个字段，其中字段a为主键。

  BEGIN;
  CREATE TABLE wide_table(
    a BIGINT primary key,
    b TEXT,
    c BIGINT,
    d TEXT,
    e BIGINT,
    f FLOAT8
  );
  call set_table_property('wide_table', 'dictionary_encoding_columns', 'b:off');
  COMMIT;

  Flink结果表相关DDL如下。

  -- 根据主键a更新或插入b,c字段
  CREATE TEMPORARY TABLE hologres_sink_1(
    a varchar primary key,
    b BIGINT,
    c BIGINT
  ) with (
    'connector'='hologres',
    'dbname'='<yourDbname>',
    'tablename'='wide_table', //Hologres接收数据的宽表表名
    'username'='<yourUsername>',
    'password'='<yourPassword>',
    'endpoint'='<yourEndpoint>',
    'ignoredelete'='true', // 忽略回撤
    'mutatetype'='insertorupdate' //使用insertorupdate参数，更新部分列
  );
  
  -- 根据主键a更新或插入d,e字段
  CREATE TEMPORARY TABLE hologres_sink_2(
    a varchar primary key,
    d BIGINT,
    e varchar
  ) with (
    'connector'='hologres',
    'dbname'='<yourDbname>',
    'tablename'='wide_table', //Hologres接收数据的宽表表名
    'username'='<yourUsername>',
    'password'='<yourPassword>',
    'endpoint'='<yourEndpoint>',
    'ignoredelete'='true', // 忽略回撤
    'mutatetype'='insertorupdate' //使用insertorupdate参数，更新部分列
  );
  
  -- 写入多个Sink语句时，需要以BEGIN STATEMENT SET;开头，以END;结尾。
  BEGIN STATEMENT SET;
  INSERT INTO hologres_sink_1 SELECT a,b,c FROM source_1;
  INSERT INTO hologres_sink_2 SELECT a,d,e FROM source_2;
  END;

Hologres Catalog

Flink全托管支持Hologres Catalog，在Flink全托管控制台直接读取Hologres元数据，不用再手动注册Hologres表，可以提高作业开发的效率且能保证表结构的正确性，详情请参见管理Hologres Catalog。

基于Hologres Catalog，目前全托管的Flink已经支持模型演进（schema evolution）以及整库同步能力，详情请参见CREATE TABLE AS（CTAS）语句和CREATE DATABASE AS（CDAS）语句。

Hologres DataStream Connector

如果您通过DataStream的方式读写Hologres数据，则需要使用Hologres DataStream Connector连接Flink全托管，详情请参见Hologres DataStream Connector。

类型映射

Connector Release Note

版本说明请参见Hologres Connector Release Note。