[游戏开发]Flink订阅及消费Hologres Binlog

订阅Hologres Binlog

要求

• Hologres V0.9及以上版本。

开启Binlog

Hologres默认关闭Binlog，可以通过binlog.level和binlog.ttl开启该功能。

列存表开启Binlog的成本大于行存表。

使用限制

• V0.10以下版本，已存在的表无法通过修改表属性的方式开启Binlog，如需开启必须重建表。
• 不支持消费分区表的Binlog。
• 对于更新频繁的场景，建议使用行存表开启Binlog。

使用示例

begin;
create table test_message_src(
  id int primary key, 
  title text not null, 
  body text);
call set_table_property('test_message_src', 'orientation', 'row');--创建行存表test_message_src
call set_table_property('test_message_src', 'clustering_key', 'id');--在id列建立聚簇索引
call set_table_property('test_message_src', 'binlog.level', 'replica');--设置表属性开启Binlog功能
call set_table_property('test_message_src', 'binlog.ttl', '86400');--binlog.ttl，Binlog的TTL，单位为秒
commit;

参数说明：

• binlog.level：是否开启Binlog，replica(开启) 、none(关闭)。
• binlog.ttl：Binlog的TTL，单位秒。默认30天，即2592000。

按需开启Binlog

HologresV1.1之后版本支持。

• 开启binlog

  -- 设置表属性开启Binlog功能
  begin;
  call set_table_property('test_message_src', 'binlog.level', 'replica');
  commit;
  
  -- 设置表属性，配置Binlog TTL时间，单位秒
  begin;
  call set_table_property('test_message_src', 'binlog.ttl', '2592000');
  commit;
  

• 关闭binlog

  -- 设置表属性关闭Binlog功能
  begin; 
  call set_table_property('bin_demo', 'binlog.level', 'none'); 
  commit; 
  

• 修改binlog的TTL

  begin; 
  call set_table_property('bin_demo', 'binlog.ttl', '8640'); --单位秒 
  commit;
  

Binlog格式说明

Binlog字段由Binlog系统字段和用户Table字段组成。如下所示：

注意事项：

• hg_binlog_event_type有四种取值：
  • DELETE=2，表示当前Binlog为删除记录。
  • INSERT=5，表示当前Binlog为插入新记录。
  • BEFORE_UPDATE=3，表示当前Binlog为一条更新前的记录。
  • AFTER_UPDATE=7，表示当前Binlog为一条更新后的记录。
• UPDATE操作会产生两条Binlog记录，分别为更新前和更新后的记录。订阅Binlog功能会保证这两条记录是连续的且更新前的Binlog记录在前，更新后的Binlog记录在后。
• 用户字段顺序与DDL定义顺序一致。

Flink和Blink实时消费Hologres Binlog

Flink实时消费Binlog

Flink VVP-2.4及以上版本，支持Hologres Connector实时消费Binlog，如下:

使用DDL语句创建源表

源表 DDL（非CDC模式）

Source消费的Binlog数据作为普通的Flink数据传递给下游节点，所有数据都是作为Insert类型的数据，可以根据业务情况选择如何处理特定hg_binlog_event_type类型的数据。

create table test_message_src_binlog_table(
  hg_binlog_lsn BIGINT,
  hg_binlog_event_type BIGINT,
  hg_binlog_timestamp_us BIGINT,
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);

源表 DDL（CDC模式）

Source消费的Binlog数据，将根据hg_binlog_event_type自动为每行数据设置准确的Flink RowKind类型（INTERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER)，这样就能完成表的数据的镜像同步，类似MySQL和Postgres的CDC功能。

create table test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'cdcMode' = 'true'
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);

参数说明

hg_binlog_xxx开头的三个字段表示Binlog的系统字段，命名和类型不支持修改。其余字段需要和用户字段一一对应，且必须为小写。

配置Binlog并发

Binlog订阅的并发等于Hologres中Table的Shard个数，请执行如下语句查看Shard数。Binlog并发建议执行计划配置，将其并发数与Binlog对应的Hologres中Table的Shard数保持一致。

select tg.property_value from hologres.hg_table_properties tb join hologres.hg_table_group_properties tg on tb.property_value = tg.tablegroup_name where tb.property_key = 'table_group' and tg.property_key = 'shard_count' and table_name = 'Hologres表名';

Blink实时消费Binlog

Blink 3.7及以上版本，支持Hologres Connector实时消费Binlog。

使用DDL语句创建源表

create table test_message_src_binlog_table(
  hg_binlog_lsn BIGINT,
  hg_binlog_event_type BIGINT,
  hg_binlog_timestamp_us BIGINT,
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  type = 'hologres',
  'endpoint' = 'ip:port',--Hologres的vpc网络地址
  'username' = 'xxxx',--当前账号的AccessKey ID
  'password' = 'xxxx',--当前账号的AccessKey Secret
  'dbname' = 'xxxx',--Hologres的DB名
  'tablename' = 'xxxx',--Hologres的表名
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '256'
);

参数说明及Binlog并发配置同Flink。