Hologres:Replicate binlog with Flink or Blink

Note the following items when consuming Hologres binary logs:

• You can consume binary logs only in Hologres V0.9 and later. You can configure engine whitelists only in Hologres V1.3.21 and later because earlier versions do not support this feature. If you enable the whitelist in an unsupported version, binary log consumption fails. If your instance version is earlier than the required version, join the Hologres DingTalk group to provide feedback. For more information, see Online support.

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

• For more information about binary logging support and how to enable and configure it, see Subscribe to Hologres binary logs.

• Only Realtime Compute for Apache Flink supports consuming Hologres binary logs. In Holohub mode, Flink supports consuming only simple data types from Hologres binary logs. Starting from Flink 6.0.3, you can consume Hologres binary logs in Java Database Connectivity (JDBC) mode. Compared to Holohub mode, JDBC mode supports more data types. For more information, see Data type mapping between Blink/Flink and Hologres. This mode also has additional permission requirements. For more information, see Permissions.

• You cannot consume binary logs from parent partitioned tables.

• Hologres V2.0 and later provide limited support for Holohub mode. Starting from Hologres V2.1, Holohub mode is deprecated and fully replaced by JDBC mode. Before you upgrade your Hologres version, check your Flink tasks that use Holohub mode, upgrade the Flink Ververica Runtime (VVR) job version, and then upgrade the Hologres instance. For more information, see Switch from Holohub mode to JDBC mode.

In this mode, the binary log data consumed by the source is passed to descendant nodes as regular Flink data. All data is of the Insert type. You can choose how to process data of a specific hg_binlog_event_type type based on your business needs. After you enable binary logging for a Hologres table, use the following DDL for the source table (non-CDC mode) in Flink to consume binary logs 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 binary log system fields. Their names and types are fixed and cannot be modified.

• Other fields correspond to user fields and must be in lowercase.

In this mode, an accurate Flink RowKind type, such as INSERT, DELETE, UPDATE_BEFORE, or UPDATE_AFTER, is automatically set for each row of the binary log data based on the hg_binlog_event_type. This lets you mirror the table data, which is similar to the change data capture (CDC) feature in MySQL and PostgreSQL.

After you enable binary logging for a Hologres table, use the following DDL for the source table (CDC mode) in Flink to consume binary logs in real time.

create table test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',// The name of the Hologres database.
  'tablename'='<yourTablename>',// The name of the Hologres table.
  'username'='<yourAccessID>',// The AccessKey ID of your account.
  'password'='<yourAccessSecret>',// The AccessKey secret of your account.
  'endpoint'='<yourEndpoint>',// The VPC endpoint of your Hologres instance.
  'binlog' = 'true',
  'cdcMode' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);