Hologres:Streaming writes to Hologres with Apache Flink 1.11+

Prerequisites

Before you begin, ensure that you have:

• A Hologres instance with a connected development tool. See Connect to HoloWeb

• An Apache Flink cluster (this topic uses Flink 1.15 in standalone mode). To set up a cluster, download the binary from the Apache Flink website and follow Local installation of Flink

• Network connectivity between the Flink cluster and your Hologres instance:

  • Same region: use the Virtual Private Cloud (VPC) endpoint of your Hologres instance

  • Different regions: use the public endpoint

Add the Maven dependency

Add the Hologres connector dependency to your pom.xml. Use the version that matches your Flink installation:

Use Flink 1.15 or later to access more connector features.

The following example uses Flink 1.15:

<dependency>
    <groupId>com.alibaba.hologres</groupId>
    <artifactId>hologres-connector-flink-1.15</artifactId>
    <version>1.4.0</version>
    <classifier>jar-with-dependencies</classifier>
</dependency>

Choose a write mode

The Hologres connector supports three write modes. Choose based on your requirements and table schema:

Bulk load notes:

• Bulk load reduces Hologres instance load by approximately 66.7% compared to fixed copy.

• Writing to a table with a primary key causes table-level locks. To reduce lock granularity to the shard level, set target-shards.enabled=true. This allows concurrent bulk load jobs.

• If you use bulk load to write to a table with a primary key, the table must be empty.

• Requires Hologres V1.3.1+. Bulk load specifically requires Hologres V1.4.0+.

Flush behavior: A write batch is flushed to Hologres when any of the following conditions is met:

• The number of records reaches jdbcWriteBatchSize (default: 256)

• The data size in a single thread reaches jdbcWriteBatchByteSize (default: 2 MB)

• The data size across all threads reaches jdbcWriteBatchTotalByteSize (default: 20 MB)

• The time since the last flush reaches jdbcWriteFlushInterval (default: 10 seconds)

Write data using Flink SQL

Use a Flink SQL sink table to write data to Hologres. The connector type is hologres.

CREATE TABLE sink (
    user_id     BIGINT,
    user_name   STRING,
    price       DECIMAL(38, 2),
    sale_timestamp TIMESTAMP
) WITH (
    'connector'  = 'hologres',
    'dbname'     = '<your-database>',
    'tablename'  = '<your-table>',
    'username'   = '<your-access-key-id>',
    'password'   = '<your-access-key-secret>',
    'endpoint'   = '<your-endpoint>'   -- Format: IP:Port
);

INSERT INTO sink SELECT * FROM source;

Replace the placeholders with your actual values:

Write data using the DataStream API

The following example applications demonstrate common patterns. All examples write to Hologres using the connector options described in Connector options reference.

Connector options reference

Data type mappings

For data type mappings between Apache Flink and Hologres, see the "Data type mappings between Realtime Compute for Apache Flink or Blink and Hologres" section in Data types.

What's next

• Create a Hologres result table — learn about streaming semantics and the mutatetype option

• Data types — full data type mapping reference