Hologres:Replicate binlog with Flink or Blink

Prerequisites

Before you begin, make sure you have:

• A Hologres instance running V0.9 or later

• Binary logging enabled on the target table. For details, see Subscribe to Hologres binary logs

• Ververica Platform (VVP) 2.4 or later for real-time binlog consumption via the Hologres connector

Limitations

• Binary logging is supported at the table level for row-oriented, column-oriented, and (from Hologres V1.1) row-column hybrid storage tables.

• Column-oriented tables have higher write overhead than row-oriented tables when binary logging is enabled. For tables with frequent updates, use row-oriented storage.

• Binary logs cannot be consumed from parent partitioned tables.

• Engine whitelists require Hologres V1.3.21 or later. If you enable a whitelist on an earlier version, binlog consumption fails.

• CDC mode source tables do not support watermark definitions. For window aggregation, use non-window aggregation instead.

• Holohub mode is deprecated as of Hologres V2.1 and fully replaced by JDBC mode. See Switch from Holohub mode to JDBC mode.

• If your instance is earlier than the required version, contact the Hologres team via online support.

Permissions

For JDBC mode, the hg_binlog extension must also exist on the instance. This extension is created by default in Hologres V2.0 and later. For setup details, see Consume Hologres binary logs using JDBC.

Binlog system fields

All binlog source tables expose three system metadata fields. Their names and types are fixed and cannot be changed.

All user-defined fields in a binlog source table must use lowercase column names.

Consume binlogs in non-CDC mode

In non-CDC mode, all binlog records arrive as Flink INSERT rows. The hg_binlog_event_type field carries the original change type, and your downstream logic decides how to handle each event.

After enabling binary logging on a Hologres table, define the source table with the following 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 (
  'connector'            = 'hologres',
  'dbname'               = '<your-database>',
  'tablename'            = '<your-table>',
  'username'             = '<your-access-key-id>',
  'password'             = '<your-access-key-secret>',
  'endpoint'             = '<your-vpc-endpoint>',
  'binlog'               = 'true',
  'binlogMaxRetryTimes'  = '10',
  'binlogRetryIntervalMs'= '500',
  'binlogBatchReadSize'  = '100'
);

Replace the placeholders with your actual values:

Consume binlogs in CDC mode

In CDC mode, Flink automatically maps each binlog record to an accurate RowKind (INSERT, DELETE, UPDATE_BEFORE, or UPDATE_AFTER) based on hg_binlog_event_type. This mirrors table changes similarly to CDC from MySQL or PostgreSQL, and is useful for replicating Hologres table data to downstream sinks.

The DDL is the same as non-CDC mode, with two differences:

• Omit the three binlog system fields from the schema — Flink manages them internally.

• Add 'cdcMode' = 'true' to the WITH clause.

CREATE TABLE test_message_src_binlog_table (
  id    INTEGER,
  title VARCHAR,
  body  VARCHAR
) WITH (
  'connector'            = 'hologres',
  'dbname'               = '<your-database>',
  'tablename'            = '<your-table>',
  'username'             = '<your-access-key-id>',
  'password'             = '<your-access-key-secret>',
  'endpoint'             = '<your-vpc-endpoint>',
  'binlog'               = 'true',
  'cdcMode'              = 'true',
  'binlogMaxRetryTimes'  = '10',
  'binlogRetryIntervalMs'= '500',
  'binlogBatchReadSize'  = '100'
);

CDC mode source tables do not support watermark definitions. For time-based window aggregation, see MySQL/Hologres CDC source tables do not support window functions.

Consume both full and incremental data

Starting from Ververica Runtime (VVR) engine 1.13-vvr-4.0.13 and Hologres V0.10, CDC source tables support a combined full + incremental mode. The job first reads all existing data from the table, then automatically switches to consuming incremental binlog records without any gap. For configuration details, see Hologres connector.

Use JDBC mode

Starting from Flink 6.0.3, binlog consumption is also available in JDBC mode. JDBC mode supports more data types and custom Hologres accounts, making it more flexible than Holohub mode. For configuration details, see Hologres connector.

Switch from Holohub mode to JDBC mode

Hologres has been phasing out Holohub mode since V2.0, and fully replaced it with JDBC mode in V2.1. Before upgrading your Hologres instance, upgrade your Flink VVR jobs first.

Find jobs that need upgrading

If you have many Flink VVR jobs, use the following tool to identify which jobs and tables need to be upgraded before migrating.

This tool supports SQL jobs (DDL-based table definitions) and Catalog jobs (hint-based parameters). It does not support JAR jobs or Catalog tables without hint parameters.

Step 1: Download the tool

Download find-incompatible-flink-jobs-1.0-SNAPSHOT-jar-with-dependencies.jar.

Step 2: Run the scan

Requires JDK 8 or later.

java -cp find-incompatible-flink-jobs-1.0-SNAPSHOT-jar-with-dependencies.jar \
  com.alibaba.hologres.FindIncompatibleFlinkJobs \
  <region> <url> <AccessKeyID> <AccessKeySecret> <binlog|rpc>

Example:

java -cp find-incompatible-flink-jobs-1.0-SNAPSHOT-jar-with-dependencies.jar \
  com.alibaba.hologres.FindIncompatibleFlinkJobs \
  Beijing \
  https://vvp.console.aliyun.com/web/xxxxxx/zh/#/workspaces/xxxx/namespaces/xxxx/operations/stream/xxxx \
  my-access-key-id \
  my-access-key-secret \
  binlog

Parameters:

Parameter	Description
region	The region of the Realtime Compute for Apache Flink project. See the region value table below.
url	The URL of any job in the target Flink project.
AccessKeyID	The AccessKey ID of an account with access to the Flink project.
AccessKeySecret	The AccessKey secret of that account.
binlog | rpc	binlog: scans all binlog source tables in the project. rpc: scans all dimension tables and sink tables that use RPC mode.

Region values:

Region	Value
China (Beijing)	Beijing
China (Shanghai)	Shanghai
China (Hangzhou)	Hangzhou
China (Shenzhen)	Shenzhen
China (Zhangjiakou)	Zhangjiakou
China (Hong Kong)	Hong Kong
Singapore	Singapore
Germany (Frankfurt)	Germany
Indonesia (Jakarta)	Indonesia
Malaysia (Kuala Lumpur)	Malaysia
US (Silicon Valley)	US
Shanghai Finance Cloud	Shanghai Finance Cloud

Step 3: Review the results

The tool outputs a list of jobs and tables that require upgrading.

What's next

• Subscribe to Hologres binary logs

• Consume Hologres binary logs using JDBC

• Data type mapping between Flink and Hologres

• Hologres connector for Realtime Compute for Apache Flink