Help Center/ Data Warehouse Service/ Tool Guide/ DWS-Connector/ dws-connector-flink/ Flink SQL Job Type/ Binlog Tables
Updated on 2026-08-27 GMT+08:00
View PDF
Share

Binlog Tables

The binlog function is supported only by clusters of version 9.1.0 or later.

Syntax

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
 
create table dwsSource (
  attr_name attr_type 
  (',' attr_name attr_type)* 
)
with (
  'connector' = 'dws',
  'url' = '',
  'tableName' = '',
  'username' = '',
  'password' = '',
  'binlog' = 'true'
);

Parameter Description

Table 1 Database configurations

Parameter

Description

Default Value

connector

The Flink framework differentiates connector parameters. This parameter is fixed to dws.

-

url

Database connection address

-

username

Configured connection user

-

password

Database user password

-
Table 2 Connection configuration parameters

Parameter

Name

Type

Description

Default Value

connectionSize

Size of the read thread pool

int

The number of threads used for operations is equal to the number of database connections, which is also equivalent to the size of the write thread.

1

binlog

Whether to read binlog information

boolean

Whether to enable the binlog function.

If the value is false, the following parameters do not take effect.

false

binlogSlotName

Binlog slot name

String

Binlog slot name. Multiple Flink tasks may consume the binlog information of a given table at the same time. Therefore, the slot name of each task must be unique. Otherwise, unknown problems may occur.

Flink table name

binlogSyncPointSize

Size of the binlog synchronization point range

int

Size of the binlog synchronization point range. Each time, the binlog information within the CSN range (current CSN, current CSN + binlogSyncPointSize) is obtained from the DWS data table, and dotting is performed after the data is processed.

If the value is too large, data may be written to disks during data query, causing performance deterioration.

5000

binlogBatchReadSize

Rows of binlog data incrementally read

int

Data volume processed in an incremental synchronization.

5000

fullSyncBinlogBatchReadSize

Rows of binlog data fully read

int

Data volume processed in a full synchronization.

50000

binlogReadTimeout

Timeout for incrementally consuming binlog data, in milliseconds

long

Timeout for incrementally synchronizing binlog information.

600000

fullSyncBinlogReadTimeout

Timeout for fully consuming binlog data, in milliseconds

long

Timeout for fully synchronizing binlog information.

1800000

binlogSleepTime

Sleep duration when no real-time binlog data is consumed, in milliseconds

long

Sleep duration when no binlog data is available for consumption. After the sleep duration expires, the system checks whether new binlog data is available.

500

binlogMaxSleepTime

Maximum sleep duration when no real-time binlog data is consumed, in milliseconds

long

Maximum sleep time when no binlog data can be consumed.

10000

binlogQueueSize

Binlog data queue size

int

Length of the binlog processing event queue. The default value is 0 (unlimited). If resources are insufficient, OOM may occur. You need to adjust the value based on the resource usage.

0

needRedistribution

Compatibility with data redistribution after scale-out

boolean

Whether to be compatible with data redistribution after scale-out. If the value is false, unknown problems may occur when the DWS cluster is in the redistribution state.

false

checkNodeChangeInterval

Interval for checking node changes, in milliseconds

long

This parameter is valid only when needRedistribution is set to true. The system checks node changes at a specified interval.

10000

binlogParallelNum

Number of threads for consuming binlog data

int

Number of threads for consuming binlog data. This parameter is valid only when the task concurrency is less than the number of DNs in the DWS cluster.

3

binlogMaxRetryTimes

Maximum number of retries after a binlog data consumption error

int

Maximum number of retries after a binlog data consumption error.

1

binlogRetryInterval

Retry interval after a binlog data consumption error, in milliseconds

long

Interval between retries after a binlog data consumption error, in milliseconds. Sleep duration during retry, which is calculated as binlogRetryInterval * (1 ~ binlogMaxRetryTimes) + Random(100).

100

binlogStartTime

A time point from which incremental data (that exceeds the data synchronization point) is synchronized and Binlog is consumed.

String

From the time point, the incremental data (that exceeds the data synchronization point) is synchronized and Binlog is consumed. The format is yyyy-MM-dd hh:mm:ss and the enable_binlog_timestamp parameter must be enabled for the corresponding DWS table.

For example, if binlog_sync_point corresponding to May 1, 2025 00:00:00 is 50000, only incremental data (binlog_sync_point is greater than 50000) is synchronized.

Note: If data is inserted into a job when it is stopped, the data synchronization point corresponding to the time after the insertion will become empty. In this case, executing synchronization may invalidate this parameter.

This is supported only by clusters of version 9.1.0.200 or later.

-

binlogIgnoreUpdateBefore

Whether to filter out before_update records in binlogs and whether to return only primary key information for delete records

bool

Whether to filter out before_update records in binlogs and whether to return only primary key information for delete records. This is supported only by clusters of version 9.1.0.200 or later.

false

executeDirectOn

Whether to enable direct DN connection.

bool

Whether SQL statements are directly executed by DNs. If the value is false, the generated SQL statements are executed by CNs. This consumes scheduling resources and affects performance.

true

binlogSkipFullSync

Whether to skip the full synchronization phase.

bool

Whether to skip full synchronization. If the value is true, the current data location is used as the start point for incremental synchronization when no available point is available.

false

binlogCondition

Binlog table pushdown condition

String

Currently, only simple pushdown conditions for Flink tables are supported.

For example, the current table contains an int field named count. If you do not want a Flink table to collect statistics on data whose count is greater than 100, you can run the following statement:

'binlogCondition'='count <= 100'

-

Examples

  • On DWS:

    When creating a binlog table, set enable_hstore_binlog_table to true. You can query the value by show enable_hstore_binlog_table.

    -- Source table (generating binlogs)

    1
    		 
    CREATE TABLE test_binlog_source(a int, b int, c int, primary key(a)) with(orientation=column, enable_hstore_opt=on, enable_binlog=true);

    -- Target table

    1
    		 
    CREATE TABLE test_binlog_sink(a int, b int, c int, primary key(a)) with(orientation=column, enable_hstore_opt=on);
  • On Flink:

    Run the following commands to perform complete data synchronization:

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
    		 
    -- Create a mapping table for the source table.
CREATE TABLE test_binlog_source ( 
   a int,
   b int,
   c int,
   primary key(a) NOT ENFORCED
) with (
   'connector' = 'dws',
   'url' = 'jdbc:gaussdb://ip:port/gaussdb',
   'binlog' = 'true',
   'tableName' = 'test_binlog_source',   
   'binlogSlotName' = 'slot',   
   'username'='xxx',   
   'password'='xxx');
   
-- Create a mapping table for the target table.
CREATE TABLE test_binlog_sink (  
   a int,
   b int,
   c int,
   primary key(a) NOT ENFORCED
) with (
   'connector' = 'dws',
   'url' = 'jdbc:gaussdb://ip:port/gaussdb',
   'tableName' = 'test_binlog_sink',   
   'ignoreUpdateBefore'='false',   
   'connectionSize' = '1',
   'username'='xxx',
   'password'='xxx');

INSERT INTO test_binlog_sink select * from test_binlog_source;

FAQs

  • Q: What should I do if "could not find slot: xxx for rel: xxx, you need to call pgxc_get_binlog_sync_point()" is displayed when a Binlog job is running?

    A: Check the table-level parameters of the Binlog table on DWS and the checkpoint interval of Flink. Pay special attention to the table-level parameter binlog_ttl. If no incremental synchronization is performed after the time specified by this parameter, the synchronization point will be deleted. If binlog_ttl is not explicitly set, the default value 86400 (unit: seconds) is used. If binlog_ttl is set to a small value, the preceding exception may occur.

    You are advised to set binlog_ttl to a value greater than the checkpoint interval on Flink.

Parent topic: Flink SQL Job Type