Realtime Compute for Apache Flink:Parameter configuration (VVR 11+)

Parameter

Description

Type

Required

Default

Remarks

connector

The type of table.

String

Yes

—

The value must be hologres.

dbname

The name of the database.

String

Yes

—

You can connect to a specific virtual warehouse by appending a suffix to the database name. For example, to connect a dimension table to the read_warehouse virtual warehouse, specify 'dbname' = 'db_test@read_warehouse'.

tablename

The name of the table.

String

Yes

—

If the schema is not public, use the schema.tableName format.

username

• The username for a custom account, in the format of BASIC$<user_name>.

• The AccessKey ID of an Alibaba Cloud account or RAM user.

String

Yes

—

• The specified user must have permissions to access the corresponding Hologres database. For more information about permissions and user management in Hologres, see Development permission models and Manage users.

• For information about how to obtain an AccessKey ID and AccessKey secret, see How do I view the AccessKey ID and AccessKey secret?.

password

• The password for the custom account.

• The AccessKey secret of an Alibaba Cloud account or RAM user.

String

Yes

—

endpoint

The endpoint of the Hologres service.

String

Yes

—

For more information, see Endpoints.

connection.pool.size

The size of the JDBC connection pool created for a single Flink table within a task.

Integer

No

5

If job performance is insufficient, consider increasing the connection pool size. The pool size is proportional to the data throughput. This parameter affects only dimension tables and sink tables.

connection.pool.name

Tables in the same TaskManager can share a connection pool by using the same name.

String

No

'default'

The default value is 'default'. If multiple tables are configured to use the same connection pool, the largest value specified for the connection.pool.size parameter among them is used.

For example, if a job contains five Hologres tables (dimension tables A and B, and sink tables C, D, and E), you can configure tables A and B to use pool1, tables C and D to use pool2, and table E, which handles high traffic, to use pool3.

connection.fixed.enabled

Specifies whether to use the lightweight connection mode.

Boolean

No

—

Hologres has a connection limit. Starting from Hologres V2.1, real-time writes support lightweight connections that are not subject to this limit.

connection.max-idle-ms

The idle timeout for a JDBC connection, in milliseconds.

Long

No

60000

If a connection remains idle longer than this time, it is closed. A new connection is automatically created when needed.

connection.ssl.mode

Specifies whether to enable Secure Sockets Layer (SSL) encryption in transit and which mode to use.

String

No

disable

• disable (default): Disables encryption in transit.

• require: Enables SSL and encrypts only the data link.

• verify-ca: Enables SSL, encrypts the data link, and uses a CA certificate to verify the authenticity of the Hologres server.

• verify-full: Enables SSL, encrypts the data link, uses a CA certificate to verify the authenticity of the Hologres server, and verifies that the Common Name (CN) or DNS name in the certificate matches the configured Hologres endpoint.

connection.ssl.root-cert.location

The path to the CA certificate, required for certificate-based encryption modes.

String

No

—

If connection.ssl.mode is set to verify-ca or verify-full, you must specify the path to the CA certificate. You can upload the certificate file by using the Manage artifacts feature on the Realtime Compute console. After the upload, the file is stored in the /flink/usrlib directory. For example, if the CA certificate file is named certificate.crt, set this parameter to '/flink/usrlib/certificate.crt'.

retry-count

The number of times to retry a write or query operation on connection failure.

Integer

No

10

—

retry-sleep-step-ms

The incremental wait time for each retry, in milliseconds.

Long

No

5000

For example, with the default value of 5,000 (5 seconds), the first retry waits for 5 seconds, the second for 10 seconds, and so on.

meta-cache-ttl-ms

The time-to-live (TTL) for locally cached TableSchema information, in milliseconds.

Long

No

600000

—

serverless-computing.enabled

Specifies whether to use serverless resources.

Boolean

No

false

If this parameter is set to true, the connector uses Hologres serverless resources for read and write operations instead of the resources of your Hologres instance. This parameter is supported only for batch reads and batch imports. It is not valid for binary log consumption, dimension table point queries, or real-time writes. For more information, see Overview.

Parameter

Description

Type

Required

Default

Remarks

source.binlog

Specifies whether to consume binlog data.

Boolean

No

true

• true (default): Consumes binlog data.

• false: Does not consume binlog data. The job performs a batch read and then stops.

source.binlog.read-mode

Specifies the read mode.

ENUM

No

AUTO

• AUTO (default): Automatically selects the optimal mode based on the instance version.

• HOLOHUB: Uses HoloHub mode to consume binlog data.

• JDBC: Uses JDBC mode to consume binlog data.

source.binlog.change-log-mode

Specifies the ChangeLog types supported by the CDC source table.

ENUM

No

UPSERT

• ALL: Supports all ChangeLog types, including INSERT, DELETE, UPDATE_BEFORE, and UPDATE_AFTER.

• UPSERT (default): Supports only upsert ChangeLogs, including INSERT, DELETE, and UPDATE_AFTER.

• ALL_AS_APPEND_ONLY: Treats all ChangeLog types as INSERT.

source.binlog.startup-mode

Specifies the consumption mode for binlog data.

ENUM

No

INITIAL

• INITIAL (default): Performs a full read of the data, and then starts incremental consumption from the binlog.

• EARLIEST_OFFSET: Starts consumption from the earliest available binlog offset.

• TIMESTAMP: Starts consumption from the binlog at the specified startTime.

• LATEST_OFFSET: Starts consumption from the latest binlog offset.

source.binlog.batch-size

Specifies the number of rows to read per batch from the binlog.

Integer

No

512

Not applicable.

source.binlog.request-timeout-ms

Specifies the timeout for reading binlog data.

Long

No

300000

Unit: milliseconds.

source.binlog.project-columns.enabled

Specifies whether to read only the fields specified in the user table when reading binlog data.

Boolean

No

None

The specified fields are those declared in the CREATE TEMPORARY TABLE statement; undeclared fields are not read. When a table has many fields but you only need to consume a subset, this option can avoid unnecessary data transfer and conversion, which improves read performance and saves bandwidth.

source.binlog.compression.enabled

Specifies whether to enable data compression during transit when reading binlog data.

Boolean

No

None

When consuming the binlog, the server returns a byte stream compressed with the LZ4 algorithm. This improves read performance and reduces bandwidth usage.

source.binlog.partition-binlog-mode

Specifies the binlog consumption mode for a partitioned table.

Enum

No

DISABLE

• DISABLE (default): Use for non-partitioned source tables. The job throws an exception if the specified Hologres table is a partitioned table.

• DYNAMIC: Continuously consumes the latest partitions of a partitioned table. The table must have dynamic partitioning enabled. DYNAMIC mode consumes partitions in chronological order. When consumption reaches the second-latest partition, it starts consuming the latest partition when a new time unit begins.

• STATIC: Consumes a fixed set of partitions from a partitioned table. Multiple partitions can be consumed simultaneously. Partitions cannot be added or removed during consumption. By default, it consumes all partitions of the parent table.

source.binlog.partition-binlog-lateness-timeout-minutes

In DYNAMIC mode, specifies the maximum lateness timeout when consuming a partitioned table.

Integer

No

60

• Unit: minutes. In DYNAMIC mode, when a new time unit begins, the connector starts consuming the latest partition for the current time. However, it does not immediately close the previous partition, but continues to monitor it to capture late-arriving data.

For example, if dynamic partitioning is by DAY, for partition 20240920, and the maximum data lateness is 1 hour, consumption for this partition stops at 2024-09-21 01:00:00, not at 2024-09-21 00:00:00.

• The lateness-timeout value cannot exceed the partition's time unit.

If partitioning is by day, the maximum value is 24 * 60 = 1440 minutes. In DYNAMIC mode, the connector usually consumes only one partition at a time, but it may consume two partitions simultaneously during the lateness period.

source.binlog.partition-values-to-read

In STATIC mode, specifies the partitions to consume. Use commas (,) to separate partition values.

String

No

None

• If you do not configure this parameter, STATIC mode consumes all partitions of the specified parent table. If configured, it consumes only the specified partitions.

• Specify only the partition values, not the full partition names. Separate multiple values with commas (,). This parameter does not support regular expressions.

startTime

Specifies the start time for the consumption offset.

String

No

None

Format: yyyy-MM-dd hh:mm:ss. If this parameter is not set and the job is not resuming from a saved state, consumption starts from the earliest binlog.

source.scan.fetch-size

Specifies the batch size for batch reads.

Integer

No

512

Not applicable.

source.scan.timeout-seconds

Specifies the timeout for batch reads.

Integer

No

60

Unit: seconds.

source.scan.filter-push-down.enabled

Specifies whether to push down filters during a batch read.

Boolean

No

false

• false (default): Does not push down filters.

• true: Pushes down supported filter conditions to Hologres during a batch read.

source.binlog.filter-push-down.enabled

Specifies whether to push down filters during binlog consumption.

Boolean

No

false

• false (default): Does not push down filters.

• true: Pushes down supported filter conditions to Hologres during binlog consumption.

scan.prefer.physical-column.over.metadata-column

Specifies whether to prioritize reading data from a physical column when it shares the same name as a metadata column.

Boolean

No

false

This parameter is supported only in VVR 11.5 and later. Earlier versions always prioritize reading data from the metadata column.

Parameter

Description

Type

Required

Default

Remarks

sink.write-mode

The data write mode.

ENUM

No

INSERT

• INSERT: Default. Uses JDBC to write data with INSERT statements.

• COPY_STREAM: Uses the fixed COPY streaming method, a high-performance option for scenarios requiring high throughput and low latency. This mode does not support deleting data, writing to a parent partitioned table, or using the sink.ignore-null-when-update.enabled parameter.

• COPY_BULK_LOAD: Performs bulk writes by using the COPY protocol. COPY_BULK_LOAD currently applies only to tables without a primary key (an exception is thrown if a primary key is duplicated). Compared to COPY_STREAM, this mode uses fewer Hologres resources for writes.

• COPY_BULK_LOAD_ON_CONFLICT: Uses the COPY protocol for batch writes and supports writing to tables with a primary key and handling primary key conflicts.

sink.on-conflict-action

The policy for handling a primary key conflict.

ENUM

No

INSERT_OR_UPDATE

• INSERT_OR_IGNORE: Keeps the first record and ignores all subsequent records with the same primary key.

• INSERT_OR_REPLACE: Replaces the entire existing row with the new row.

• INSERT_OR_UPDATE: Default. Updates a subset of columns in an existing row.

  For example, consider a table with columns a, b, c, and d, where a is the primary key. If an incoming record provides values only for columns a and b, a primary key conflict causes the system to update only column b and leave columns c and d unchanged.

sink.create-missing-partition

Specifies whether to automatically create a partition based on the partition value if a matching one does not exist when writing to a partitioned table.

Boolean

No

false

• If you use the DATE data type as the partition key and enable dynamic partitioning, the names of auto-created partitions follow the dynamic partitioning naming convention.

• Use this parameter with caution. Ensure that partition values do not contain dirty data. Otherwise, you might create incorrect partitions, which can lead to a failover.

• If sink.write-mode is not INSERT, you cannot write to a parent partitioned table.

sink.delete-strategy

The policy for handling retraction messages.

String

No

CHANGELOG_STANDARD

• IGNORE_DELETE: Ignores UPDATE_BEFORE and DELETE messages. This option is suitable for scenarios that require only inserts or updates but not deletes.

• NON_PK_FIELD_TO_NULL: Ignores UPDATE_BEFORE messages and processes DELETE messages by setting non-primary key fields to NULL. This option is suitable for partial update scenarios where you want to perform a delete without affecting other columns.

• DELETE_ROW_ON_PK: Ignores UPDATE_BEFORE messages and processes DELETE messages by deleting the entire row based on the primary key. This option is suitable for partial update scenarios where you want to delete the entire row.

• CHANGELOG_STANDARD: Follows the standard Flink SQL changelog mechanism. It does not ignore delete operations. This option performs updates by first deleting the old record and then inserting the new one, which ensures data accuracy. This option is suitable for scenarios that do not involve partial updates.

sink.ignore-null-when-update.enabled

When sink.on-conflict-action is 'INSERT_OR_UPDATE', specifies whether to ignore null values in the incoming data during an update.

Boolean

No

false

• false: Default. Writes null values to the Hologres sink table.

• true: Ignores null values in the incoming update data.

sink.ignore-null-when-update-by-expr.enabled

When sink.on-conflict-action is 'INSERT_OR_UPDATE', specifies whether to use an expression-based method to ignore null values in incoming update data.

Boolean

No

false

This method offers better performance than sink.ignore-null-when-update.enabled.

• false: Default.

  • If sink.ignore-null-when-update.enabled is true, null values in updates are ignored.

  • If sink.ignore-null-when-update.enabled is false, null values are written to the Hologres sink table.

• true: Ignores null values in incoming update data, regardless of the sink.ignore-null-when-update.enabled setting.

sink.default-for-not-null-column.enabled

Specifies whether the connector provides a default value when a null is written to a NOT NULL column that lacks a defined default value.

Boolean

No

true

• true: Default. Allows the connector to provide and write a default value based on the following rules:

  • For a String data type, an empty string ("") is written.

  • For a Number data type, 0 is written.

  • For Date, timestamp, or timestamptz data types, 1970-01-01 00:00:00 is written.

• false: Does not provide a default value. Writing a null value to a NOT NULL column throws an exception.

sink.remove-u0000-in-text.enabled

Specifies whether the connector removes the illegal \u0000 character from string types during a write operation.

Boolean

No

true

• false: The connector does not modify the data. If dirty data is encountered, a write operation may throw an error similar to the following: ERROR: invalid byte sequence for encoding "UTF8": 0x00

  In this case, you must either process the dirty data in the source table or define the logic for handling dirty data in your SQL statement.

• true: Default. The connector removes the \u0000 character from string types to prevent write errors.

sink.partial-insert.enabled

Specifies whether to insert or update only the fields defined in the INSERT statement.

Boolean

No

false

• false: Default. All fields defined in the DDL of the sink table are updated. Fields that are not declared in the INSERT statement are updated to null.

• true: Pushes only the fields defined in the INSERT statement to the connector, which allows you to update or insert only the declared fields.

sink.deduplication.enabled

Specifies whether to perform deduplication during batching.

Boolean

No

true

• true: Default. If a batch contains multiple records with the same primary key, the connector deduplicates them and writes only the last record to the Hologres sink table. Consider a table with two fields, where the first is the primary key:

  • If the records INSERT (1,'a') and INSERT (1,'b') arrive in sequence, only the last record, (1,'b'), is written to the Hologres sink table after deduplication.

  • If the Hologres sink table already contains the record (1,'a'), and the records DELETE (1,'a') and INSERT (1,'b') arrive in sequence, only the last record (1,'b') is written to Hologres. This results in a direct update instead of a delete followed by an insert.

• false: Disables deduplication during batching. If a new record has the same primary key as a record in the current batch, the batch is first written to the sink, and then the new record is processed.

sink.aggressive-flush.enabled

Specifies whether to enable aggressive flush mode.

Boolean

No

false

If set to true, the connector forcibly flushes a batch when the connection is idle, even if the batch has not reached its configured size. This reduces data write latency during low-traffic periods.

sink.insert.check-and-put.column

Enables conditional updates and specifies the column to check.

String

No

None

The value of this parameter must be an existing column name in the Hologres table.

sink.insert.check-and-put.operator

The comparison operator for the conditional update operation.

String

No

GREATER

Compares the check column of the incoming record with the existing value in the table. The update proceeds if the condition is met. Supported operators are: GREATER, GREATER_OR_EQUAL, EQUAL, NOT_EQUAL, LESS, LESS_OR_EQUAL, IS_NULL, and IS_NOT_NULL.

sink.insert.check-and-put.null-as

During a conditional update, treats a null value in the existing data as the value specified by this parameter.

String

No

None

In PostgreSQL, any comparison with NULL returns FALSE. Therefore, when the existing value in the table is NULL, you must set this parameter to a specific value to use for comparison. This is equivalent to the COALESCE function in SQL.

sink.insert.batch-size

In INSERT mode, the maximum number of records to include in a batch before writing.

Integer

No

512

The connector triggers a write when any of the sink.insert.batch-size, sink.insert.batch-byte-size, or sink.insert.flush-interval-ms thresholds is reached.

sink.insert.batch-byte-size

In INSERT mode, the maximum size in bytes of a batch before writing.

Long

No

2097152 (2 MB)

sink.insert.flush-interval-ms

In INSERT mode, the maximum interval in milliseconds to wait before a batch is flushed to Hologres.

Long

No

10000

sink.copy.format

The data transmission format used in COPY mode.

String

No

• COPY_STREAM mode defaults to binary.

• COPY_BULK_LOAD or COPY_BULK_LOAD_ON_CONFLICT mode defaults to text.

Supported formats for COPY_STREAM mode:

• binary

• text

• binaryrow (Requires Hologres engine V4.1.0 or later)

Supported format for COPY_BULK_LOAD or COPY_BULK_LOAD_ON_CONFLICT mode: text only

sink.insert.conflict-update-set

The Hologres expression used to update the row when a primary key conflict occurs.

String

No

None

Equivalent to insert into tbl values(xxx) on conflict(pk) do update set <conflict-update-set>. You can specify a Hologres expression or function.

For example, if this parameter is set to col1=old.col1+excluded.col1,col2=excluded.col2, a primary key conflict causes the value of col1 to be updated to the sum of the old and new values, and the value of col2 to be updated to the new value.

• If this parameter is not specified, the connector updates all incoming fields to their new values by default.

• For stateful expressions where the result depends on the old value (for example, col=old.col+excluded.col), ensure that a column can serve as a row version number and set sink.insert.conflict-where to excluded.seq>old.seq. This ensures data correctness after a failover and recovery.

sink.insert.conflict-where

The Hologres filter condition for triggering an update during a primary key conflict.

String

No

None

Equivalent to insert into tbl values(xxx) on conflict(pk) do update set <conflict-update-set> where <conflict-where>. You can specify a Hologres expression or function.

For example, if this parameter is set to excluded.col1>old.col1, an update is triggered only when the new value of col1 is greater than the existing value.

Parameter

Description

Type

Required

Default

Notes

lookup.read.batch-size

The maximum number of records to batch for point queries on a Hologres dimension table.

Integer

No

256

None

lookup.read.timeout-ms

The timeout for point queries on a dimension table.

Long

No

0, which means no timeout.

None

lookup.read.column-table.enabled

Specifies whether to use a column-oriented table as the dimension table.

Boolean

No

false

Column-oriented tables perform poorly for point queries. We recommend using a row-oriented or hybrid row-column table as a dimension table. A warning is logged if this parameter is enabled for a column-oriented table.

lookup.insert-if-not-exists

Specifies whether to insert a record if it does not exist.

Boolean

No

false

If a point query finds no match for the current record in the dimension table, the connector inserts the record.

cache

The cache policy.

String

No

None

Hologres supports only two cache policies: None and LRU.

cacheSize

The cache size, in rows.

Integer

No

10000

After you select the LRU cache policy, you can set the cache size. The unit is entries.

cacheTTLMs

The cache time-to-live (TTL), in milliseconds.

Long

No

See Notes.

The unit is milliseconds. The default value of cacheTTLMs depends on the cache configuration:

• If the cache is configured as LRU, cacheTTLMs specifies the cache timeout. The default setting is no expiration.

• If the cache is configured as None, cacheTTLMs can be omitted, which indicates that the cache does not time out.

cacheEmpty

Specifies whether to cache empty join results.

Boolean

No

true

• true (Default): Caches empty join results.

• false: Does not cache empty join results.

  However, if the condition before AND in a join statement is met but the condition after AND is not, the empty result is still cached. The following code provides an example.

  LEFT JOIN latest_emergency FOR SYSTEM_TIME AS OF PROCTIME() AS t2
   ON t1.alarm_id = t2.alarm_id -- If a dynamic alarm is detected, match it by using the dynamic alarm ID. Otherwise, ignore the dynamic alarm ID field.
   AND CASE
   WHEN alarm_type = 2 THEN t1.dynamic_id = t2.dynamic_alarm_id
   ELSE true
   END

async

Specifies whether to return results asynchronously.

Boolean

No

false

• true: Returns results asynchronously.

• false (Default): Returns results synchronously.

lookup.async

Specifies whether to enable asynchronous queries for dimension tables.

Boolean

No

true

• true (Default): Enables asynchronous queries.

• false: Disables asynchronous queries.

lookup.filter-push-down.enabled

Specifies whether to push down dimension table filter conditions to the Hologres server.

Boolean

No

false

This feature supports only comparison operations between a column and a constant. The connector pushes down only equality (=) and comparison (<, <=, >, >=) operators.