Consume Hologres binlog data with Flink

Before consuming Hologres binlog data, note the following:

• You can consume binlog data only from Hologres V0.9 or later. You can configure an engine whitelist only for Hologres V1.3.21 or later. If you enable a whitelist on an instance that runs a version earlier than Hologres V1.3.21, binlog consumption fails. If your instance version is earlier than required, you can Obtain online support for Hologres for assistance.

• Hologres supports the binlog feature at the table level for row-oriented tables and column-oriented tables. Starting from Hologres V1.1, row-column hybrid storage tables are also supported. After you enable the binlog feature, column-oriented tables theoretically have higher overhead than row-oriented tables. Therefore, for scenarios with frequent data updates, we recommend that you enable the binlog feature for row-oriented tables.

• For more information about binlog support and how to enable and configure this feature, see Subscribe to Hologres binlogs.

• Only Realtime Compute for Apache Flink provided by Alibaba Cloud supports consuming Hologres binlog data. In HoloHub mode, consuming Hologres binlog data with Flink supports only simple data types. Starting with Flink VVR 6.0.3, you can consume Hologres binlog data in JDBC mode. Compared with HoloHub mode, JDBC mode supports more data types. For more information, see Data type mappings between Blink/Flink and Hologres. This mode requires additional permissions. For details, see Permissions.

• You cannot consume binlog data from a parent partitioned table.

• Hologres V2.0 offers limited support for HoloHub mode. Starting from V2.1, HoloHub mode is deprecated and fully replaced by JDBC mode. Before you upgrade your Hologres instance, refer to Switch from HoloHub mode to JDBC mode to check your Flink jobs that currently use HoloHub mode, upgrade your Flink VVR jobs accordingly, and then upgrade your Hologres instance.

In this mode, the binlog data consumed by the Source is passed to downstream nodes as regular Flink data. This means all data is treated as data of the Insert type, and you can choose how to process data of a specific hg_binlog_event_type type based on your business requirements. After binlog is enabled for a Hologres table, you can use the following DDL for a source table (in non-CDC mode) in Flink to consume the binlog in real time.

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'
);

• The three binlogxxx parameters are binlog system fields. Their names and types are fixed and cannot be modified.

• Other fields correspond to your user-defined fields and must be in all lowercase.

In this mode, the Source consumes binlog data and automatically sets the correct Flink RowKind type (INSERT, DELETE, UPDATE_BEFORE, or UPDATE_AFTER) for each row based on the hg_binlog_event_type. This enables mirror synchronization of table data, similar to the CDC feature in MySQL and PostgreSQL.

After you enable the binlog feature for a Hologres table, you can use the following DDL in Flink to create a source table to consume binlog data in real time in CDC mode.

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'
);