Batch import data into Hologres using Flink - Hologres - Alibaba Cloud Documentation Center

Background

Hologres integrates with Apache Flink to provide real-time data streaming capabilities. For use cases that are not time-sensitive—such as loading historical data, processing offline data, or aggregating logs—batch import is the recommended approach. Batch import writes large volumes of data to Hologres at once, which is more efficient and conserves resources. Choose between real-time and batch import based on your business needs and available resources. For more information about real-time import, see Realtime Compute for Apache Flink.

Prerequisites

You have purchased a Hologres instance. For more information, see Purchase a Hologres instance.
You have deployed an Apache Flink cluster, version 1.15 or later. For more information, see the following topics:
- Apache Flink: Deploy Flink.
- Realtime Compute for Apache Flink: Activate Realtime Compute for Apache Flink.

Batch import using Realtime Compute for Apache Flink

Create a Hologres result table to store data imported from Flink. For instructions, see Connect to HoloWeb and run queries. This topic uses the test_sink_customer table as an example.

-- Create a Hologres result table.
CREATE TABLE test_sink_customer
(
  c_custkey     BIGINT,
  c_name        TEXT,
  c_address     TEXT,
  c_nationkey   INT,
  c_phone       TEXT,
  c_acctbal     NUMERIC(15,2),
  c_mktsegment  TEXT,
  c_comment     TEXT,
  "date"        DATE
) WITH (
  distribution_key="c_custkey,date", 
  orientation="column"
);

Note

The field names and data types in the Flink source table must match those in the Hologres result table.

Log in to the Realtime Compute for Apache Flink console. On the Deployments page, click Create Deployment. Configure the deployment parameters and click Deploy. For more information about the parameters, see Deploy a JAR job.

The following table describes the key parameters.

Parameter	Description
Deployment Type	Select JAR.
Deployment Mode	You can select stream mode or batch mode. This topic uses batch mode as an example.
Engine Version	For more information about engine versions, see Engine versions and Lifecycle policies. This topic uses version `vvr-8.0.7-flink-1.17` as an example.
JAR URI	Upload the open source Flink connector: hologres-connector-flink-repartition.jar. Note You can use the open source Flink connector to batch import data to Hologres. For the source code of the Flink connector, see the official Hologres GitHub repository.
Entry Point Class	The entry point class of the program. The main class for the Flink connector is `com.alibaba.ververica.connectors.hologres.example.FlinkToHoloRePartitionExample`.
Entry Point Main Arguments	Provide the path to the `repartition.sql` file. In the Realtime Compute for Apache Flink runtime, additional dependency files are stored in `/flink/usrlib/`. Therefore, the complete argument is `--sqlFilePath="/flink/usrlib/repartition.sql"`.
Additional Dependencies	Upload the `repartition.sql` file. This file is a Flink SQL script used to define the data source, declare the result table, and configure the connection to Hologres. The following code is an example `repartition.sql` file. -- DDL for the source table. This example uses the Flink DataGen connector to generate test data. CREATE TEMPORARY TABLE source_table ( c_custkey BIGINT ,c_name STRING ,c_address STRING ,c_nationkey INTEGER ,c_phone STRING ,c_acctbal NUMERIC(15, 2) ,c_mktsegment STRING ,c_comment STRING ) WITH ( 'connector' = 'datagen' ,'rows-per-second' = '10000' ,'number-of-rows' = '1000000' ); -- DQL for the source table. The query result must match the schema of the result table defined in the sink DDL, including the number and types of fields. SELECT , cast('2024-04-21' as DATE) FROM source_table; -- DDL for the sink table. This declares the result table and configures the connection to Hologres. CREATE TABLE sink_table ( c_custkey BIGINT ,c_name STRING ,c_address STRING ,c_nationkey INTEGER ,c_phone STRING ,c_acctbal NUMERIC(15, 2) ,c_mktsegment STRING ,c_comment STRING ,`date` DATE ) WITH ( 'connector' = 'hologres' ,'dbname' = 'doc_*' ,'tablename' = 'test_sink_customer' ,'username' = 'yourAccessKeyId' ,'password' = 'yourAccessKeySecret' ,'endpoint' = 'hgpostcn-cn-7pp2e1k7-cn-hangzhou.hologres.aliyuncs.com:80' ,'jdbccopywritemode' = 'true' ,'bulkload' = 'true' ,'target-shards.enabled'='true' ); Note** For more information about the Hologres connection parameters in the `repartition.sql` file, see Hologres Flink connector parameters.

Click the deployment name and go to the Deployment Details page. In the Resource Configurations section, modify the Parallelism.

Note
We recommend that you set the parallelism to the ShardCount of the Hologres result table.
Query the Hologres result table.

After the Flink job is submitted, you can query the written data in Hologres. Sample statement:
```
SELECT * FROM test_sink_customer;
```

Batch import using Apache Flink

Create a Hologres result table to store data imported from Flink. For instructions, see Connect to HoloWeb and run queries. This topic uses the test_sink_customer table as an example.

-- Create a Hologres result table.
CREATE TABLE test_sink_customer
(
  c_custkey     BIGINT,
  c_name        TEXT,
  c_address     TEXT,
  c_nationkey   INT,
  c_phone       TEXT,
  c_acctbal     NUMERIC(15,2),
  c_mktsegment  TEXT,
  c_comment     TEXT,
  "date"        DATE
) WITH (
  distribution_key="c_custkey,date", 

  orientation="column"
);

Note

You can set the shard count based on your data volume. For more information about shards, see Manage table groups and shard count.

Create the repartition.sql file and upload it to any directory in your Flink cluster environment. This topic uses /flink-1.15.4/src/repartition.sql as an example path. The following code is an example repartition.sql file.

Note

This file is a Flink SQL script used to define the data source, declare the result table, and configure the connection to Hologres.

-- DDL for the source table. This example uses the Flink DataGen connector to generate test data.
CREATE TEMPORARY TABLE source_table
(
  c_custkey     BIGINT
  ,c_name       STRING
  ,c_address    STRING
  ,c_nationkey  INTEGER
  ,c_phone      STRING
  ,c_acctbal    NUMERIC(15, 2)
  ,c_mktsegment STRING
  ,c_comment    STRING
)
WITH (
  'connector' = 'datagen'
  ,'rows-per-second' = '10000'
  ,'number-of-rows' = '1000000'
);

-- DQL for the source table. The query result must match the schema of the result table defined in the sink DDL, including the number and types of fields.
SELECT *, cast('2024-04-21' as DATE) FROM source_table;

-- DDL for the sink table. This declares the result table and configures the connection to Hologres.
CREATE TABLE sink_table
(
  c_custkey     BIGINT
  ,c_name       STRING
  ,c_address    STRING
  ,c_nationkey  INTEGER
  ,c_phone      STRING
  ,c_acctbal    NUMERIC(15, 2)
  ,c_mktsegment STRING
  ,c_comment    STRING
  ,`date`       DATE
)
WITH (
  'connector' = 'hologres'
  ,'dbname' = 'doc_****'
  ,'tablename' = 'test_sink_customer'
  ,'username' = 'yourAccessKeyId'
  ,'password' = 'yourAccessKeySecret'
  ,'endpoint' = 'hgpostcn-cn-7pp2e1k7****-cn-hangzhou.hologres.aliyuncs.com:80'
  ,'jdbccopywritemode' = 'true'
  ,'bulkload' = 'true'
  ,'target-shards.enabled'='true'
);

The following table describes the key parameters.

Parameter	Required	Description
connector	Yes	The type of the connector. The value must be `hologres`.
dbname	Yes	The name of the Hologres database.
tablename	Yes	The name of the Hologres table that receives the data.
username	Yes	The AccessKey ID of your Alibaba Cloud account. You can obtain the AccessKey ID from the AccessKey Pair page.
password	Yes	The AccessKey secret that corresponds to your AccessKey ID.
endpoint	Yes	The VPC endpoint of the Hologres instance. Go to the instance details page in the Hologres console and obtain the endpoint from the Configurations section. Note The endpoint must include the port number in the format of `ip:port`. Use the VPC endpoint for connections within the same region. Use the public endpoint for cross-region connections.
jdbccopywritemode	No	The data write method. Valid values: `false` (default): Uses the `INSERT` method. `true`: Uses the `COPY` method. The `COPY` method includes streaming `COPY` (Fixed Copy) and batch `COPY`. By default, streaming `COPY` is used. Note Compared with the `INSERT` method, streaming `COPY` uses a streaming model to achieve higher throughput, lower data latency, and reduced client memory consumption, as data is not batched. However, it does not support data retraction.
bulkload	No	Whether to use the batch `COPY` method. Valid values: `true`: Uses batch `COPY`. This setting takes effect only when `jdbccopywritemode` is also set to `true`. Otherwise, streaming `COPY` is used. Note Compared with streaming `COPY`, batch `COPY` is more efficient and uses Hologres resources more effectively, resulting in superior write performance. Choose the appropriate write method based on your business requirements. When you use batch `COPY` to write data to a table with a primary key, table locks may occur. You can set the `target-shards.enabled` parameter to `true` to reduce the lock granularity from the table level to the shard level. This allows multiple batch import tasks to run concurrently and reduces table lock contention. Compared with streaming `COPY`, this approach significantly reduces the load on the Hologres instance when you write to a table with a primary key. Tests show a load reduction of approximately 66.7%. When you use batch `COPY`, if the destination table has a primary key, the table must be empty before the write operation. Otherwise, the write process will be slowed by data deduplication based on the primary key. `false` (default): Does not use batch `COPY`.
target-shards.enabled	No	Whether to enable target shard batch writing. Valid values: `true`: Enables target shard batch writing. When the source data is repartitioned by shard, this reduces the lock granularity to the shard level. `false` (default): Disables this feature.

Note

For more information about the Hologres connection parameters in the repartition.sql file, see Hologres Flink connector parameters.

In your Flink cluster environment, upload the open source Flink connector hologres-connector-flink-repartition.jar to any directory. This topic uses the root directory as an example.

Note
You can use the open source Flink connector to batch import data to Hologres. For the source code of the Flink connector, see the official Hologres GitHub repository.
Submit the Flink job. Sample command:
```
./bin/flink run -Dexecution.runtime-mode=BATCH -p 3 -c com.alibaba.ververica.connectors.hologres.example.FlinkToHoloRePartitionExample hologres-connector-flink-repartition.jar --sqlFilePath="/flink-1.15.4/src/repartition.sql"
```
Parameters in the preceding command:
- Dexecution.runtime-mode: The execution mode of the Flink job. For more information, see Execution Mode.
- p: The parallelism of the job. We recommend setting this value to the ShardCount of the result table, or a divisor of the ShardCount.
- c: The fully qualified name of the main class in the hologres-connector-flink-repartition.jar file.
- sqlFilePath: The path to the repartition.sql file.
Query the Hologres result table.

After the Flink job is submitted, you can query the written data in Hologres. Sample statement:
```
SELECT * FROM test_sink_customer;
```