Batch writing to Hologres - Hologres - Alibaba Cloud Documentation Center

Hologres is a one-stop real-time data warehouse engine compatible with the PostgreSQL protocol. It supports writing, updating, and querying large data volumes across both real-time and offline scenarios. Based on performance test results from Spark batch writes to Hologres, this topic helps you choose the optimal write mode for different scenarios.

For real-time data writes and updates, see Stress testing data writes, updates, and point queries.

Comparison of batch write modes

Hologres supports several batch write modes, including the traditional COPY mode, the FIXED COPY streaming import mode built on the COPY protocol, and the INSERT INTO VALUES mode that converts batch imports into streaming writes.

The following table compares the three write modes in detail.

Feature	COPY		FIXED COPY	INSERT INTO VALUES
Description	Batch import to tables without a primary key	Batch import to tables with a primary key	Streaming import mode built on the COPY protocol	Basic streaming import mode
Use cases	Batch import from Spark Batch import from Apache Flink		Import from Flink Import from DataWorks Data Integration Streaming import from Spark	Import from Flink Import from DataWorks Data Integration Streaming import from Spark
Lock granularity	row lock	table lock	row lock	row lock
Data visibility	Visible after the COPY operation completes	Visible after the COPY operation completes	Visible in real time	Visible in real time
Performance	High	High	Medium	Medium
Hologres resource consumption	Low	Low	High	High
Client resource consumption	Low	High	Low	Medium
Supported primary key conflict policies	Not applicable	NONE UPDATE: Supports full-row updates in V3.0.4 or later and partial updates in V3.1.1 or later. IGNORE: Supported in V3.0.4 or later.	NONE (returns an error if a conflict occurs) UPDATE IGNORE	NONE (returns an error if a conflict occurs) UPDATE IGNORE

Batch write mode selection

In batch write scenarios, the three modes offer the following features:

COPY mode: The traditional COPY mode offers the best write performance while consuming the fewest Hologres resources.
FIXED COPY mode: A streaming import mode built on the COPY protocol. It generates only a row lock and provides real-time data visibility.
INSERT INTO VALUES mode: This mode converts batch imports into traditional streaming writes and provides no significant advantage in batch write scenarios.

Note
The INSERT INTO VALUES mode lacks a distinct advantage only in batch write scenarios. However, it is the only mode that supports data retraction (deletion) and is required for use cases such as writing binary log data.

Unless you have specific requirements for real-time data visibility, lock granularity (a table lock prevents concurrent write tasks), or data source load, we recommend using the COPY mode for batch writes.

For batch writes from Spark: We recommend upgrading your Hologres instance to V2.2.25 or later and setting the write parameter write.mode for the connector to auto (the default value). The system then automatically selects the optimal batch write mode.
For batch writes from Flink: First, use the decision tree below to select either the COPY mode or the FIXED COPY mode. Then, configure the following parameters:
- jdbccopywritemode: Set to TRUE. This prevents the use of the INSERT INTO VALUES mode.
- bulkload: Set to TRUE for COPY mode or FALSE for FIXED COPY mode based on your requirements.

Use the following decision tree to select the appropriate batch write mode for your scenario.

Batch write performance test

This test uses Hologres's open-source Hologres-Spark-Connector.

Prerequisites

Environment setup

Before you begin, make sure you have the following environment set up:

Important

The Hologres instance and the EMR Spark cluster must be in the same region and use the same VPC.

Provision a Hologres instance of V2.2.25 or later and create a database.
Create an EMR Spark cluster that runs Spark 3.3.0 or later. For more information, see Create a cluster.
Download the Spark-Connector package.

To enable Spark to read from and write to Hologres, download the hologres-connector-spark-3.x JAR file from the Maven Central Repository.

The following table describes the environment used for the performance test.

Service	Version	Specifications
Hologres	V3.0.30	64 cores, 256 GB (1 CU = 1 core, 4 GB)
EMR Spark	EMR-5.18.1, Spark-3.5.3	8 cores, 32 GB × 8 (1 master node, 7 core nodes) Important You must activate the OSS-HDFS service.
Spark-Connector	1.5.2	N/A

Test data preparation

Prepare the source data.

Log on to the master node of the EMR Spark cluster. For more information, see Connect to an ECS instance.
Download the TPC-H_Tools_v3.0.0.zip TPC-H tool package. Then, copy the package to the master node's ECS instance, extract it, and navigate to the TPC-H_Tools_v3.0.0/TPC-H_Tools_v3.0.0/dbgen directory.

Run the following command in the dbgen directory to generate a customer.tbl file for a 1 TB test dataset.

./dbgen -s 1000 -T c

The following table describes the customer table in the 1 TB TPC-H dataset.

Item	Description
Number of fields	8
Field types	INT, BIGINT, TEXT, DECIMAL
Number of rows	150,000,000
Number of shards	40

Import the test data into Spark.

Run the following command to upload the customer.tbl file to the Spark cluster.
```
hadoop fs -put customer.tbl <spark_resource>
```
In the command, spark_resource refers to the Root Storage Directory of Cluster that you set when you created the EMR Spark cluster.

Row-column hybrid storage

CREATE TABLE test_table_mixed (
    C_CUSTKEY BIGINT PRIMARY KEY,
    C_NAME TEXT,
    C_ADDRESS TEXT,
    C_NATIONKEY INT,
    C_PHONE TEXT,
    C_ACCTBAL DECIMAL(15, 2),
    C_MKTSEGMENT TEXT,
    C_COMMENT TEXT
)
WITH (
    orientation = 'column,row'
);

Column-oriented storage (with primary key)

CREATE TABLE test_table_column (
    C_CUSTKEY BIGINT PRIMARY KEY,
    C_NAME TEXT,
    C_ADDRESS TEXT,
    C_NATIONKEY INT,
    C_PHONE TEXT,
    C_ACCTBAL DECIMAL(15, 2),
    C_MKTSEGMENT TEXT,
    C_COMMENT TEXT
)
WITH (
    orientation = 'column'
);

Column-oriented storage (without primary key)

CREATE TABLE test_table_column_no_pk (
    C_CUSTKEY BIGINT,
    C_NAME TEXT,
    C_ADDRESS TEXT,
    C_NATIONKEY INT,
    C_PHONE TEXT,
    C_ACCTBAL DECIMAL(15, 2),
    C_MKTSEGMENT TEXT,
    C_COMMENT TEXT
)
WITH (
    orientation = 'column'
);

Row-oriented storage (with primary key)

CREATE TABLE test_table_row (
    C_CUSTKEY BIGINT PRIMARY KEY,
    C_NAME TEXT,
    C_ADDRESS TEXT,
    C_NATIONKEY INT,
    C_PHONE TEXT,
    C_ACCTBAL DECIMAL(15, 2),
    C_MKTSEGMENT TEXT,
    C_COMMENT TEXT
)
WITH (
    orientation = 'row'
);

Row-oriented storage (without primary key)

CREATE TABLE test_table_row_no_pk (
    C_CUSTKEY BIGINT,
    C_NAME TEXT,
    C_ADDRESS TEXT,
    C_NATIONKEY INT,
    C_PHONE TEXT,
    C_ACCTBAL DECIMAL(15, 2),
    C_MKTSEGMENT TEXT,
    C_COMMENT TEXT
)
WITH (
    orientation = 'row'
);

Performance test

Test configuration

This topic tests the import performance of different modes.

Log on to the master node of the EMR Spark cluster. For more information, see Connect to an ECS instance. Upload the downloaded Spark-Connector package, and then run the following command to enter the Spark SQL interactive shell.

Note
You can adjust the value of the spark.sql.files.maxPartitionBytes parameter to control the concurrency for reading HDFS files in Spark. In this example, the concurrency is set to 40.
```
# Enter the Spark SQL interactive shell.
spark-sql --jars <path>/hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
--conf spark.executor.instances=40 \
--conf spark.executor.cores=1 \
--conf spark.executor.memory=4g \
--conf spark.sql.files.maxPartitionBytes=644245094
```
In the command, path is the directory that contains the hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar file.

In the Spark SQL interactive shell, run the following SQL statements to create temporary views and write data.

A temporary table is used in this test to facilitate parameter adjustments when evaluating the write performance of different modes.

Note

In practice, you can also use a catalog to load Hologres tables directly.

-- Create a temporary table in CSV format.
CREATE TEMPORARY VIEW csvtable (
  c_custkey BIGINT,
  c_name STRING,
  c_address STRING,
  c_nationkey INT,
  c_phone STRING,
  c_acctbal DECIMAL(15, 2),
  c_mktsegment STRING,
  c_comment STRING)
USING csv OPTIONS (
  path "<spark_resources>/customer.tbl", sep "|"
);
CREATE TEMPORARY VIEW hologresTable (
  c_custkey BIGINT,
  c_name STRING,
  c_address STRING,
  c_nationkey INT,
  c_phone STRING,
  c_acctbal DECIMAL(15, 2),
  c_mktsegment STRING,
  c_comment STRING)
USING hologres OPTIONS (
  jdbcurl "jdbc:postgresql://<hologres_vpc_endpoint>/<database_name>",
  username "<accesskey_id>", 
  password "<accesskey_secret>", 
  table "<table_name>",
  direct_connect "false",
  write.mode "auto",
  write.insert.thread_size "3",
  write.insert.batch_size "2048"
);
INSERT INTO hologresTable SELECT * FROM csvTable;

The following table describes the parameters.

Parameter	Description
spark_resources	The cluster root storage path that you configure when you create the EMR Spark cluster. You can find this path in the Cluster Information section on the Basic Information page of your cluster in the EMR on ECS console.
hologres_vpc_endpoint	The VPC Domain Name of the Hologres instance. You can log on to the Hologres console, click the ID of the target instance, and find the endpoint for the specified VPC in the network information section of the instance details page. For example, a VPC endpoint in the China (Hangzhou) region has the format `<Instance ID>-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80`.
database_name	The name of the database in the Hologres instance.
accesskey_id	The AccessKey ID with read permissions on the specified Hologres database.
accesskey_secret	The AccessKey secret with read permissions on the specified Hologres database.
table_name	The name of the destination Hologres table.
write.mode	The write mode. Valid values: `auto`: The default value. The connector automatically selects the optimal mode. `insert`: Writes data by using INSERT INTO VALUES statements. `stream`: Performs streaming writes by using FIXED COPY. `bulk_load`: Performs a batch import to a table without a primary key by using COPY. `bulk_load_on_conflict`: Performs a batch import to a table with a primary key by using COPY.
write.insert.thread_size	The write concurrency. This parameter applies only to the `insert` mode.
write.insert.batch_size	The write batch size. This parameter applies only to the `insert` mode.
write.on_conflict_action	`INSERT_OR_REPLACE` (default): If a primary key conflict occurs, the data is updated. `INSERT_OR_IGNORE`: If a primary key conflict occurs, the data is ignored.

For more information about the parameters, see Parameters.

Test scenarios

Test scenario	Options
Table storage format	Row-oriented storage Column-oriented storage Row-column hybrid storage
Data update method	Append-only writes to a table without a primary key Initial writes to an empty table with a primary key Full-row updates to a table with a primary key
Write mode	insert (INSERT INTO VALUES) stream (FIXED COPY) bulk_load (COPY import to a table without a primary key) bulk_load_on_conflict (COPY import to a table with a primary key)

Test results

The test results include the following fields:

Field	Description
Total job duration	The total run time of the Spark job. This value is the time taken to execute the INSERT operation in the Spark SQL interactive shell on an EMR Spark cluster node. This duration appears in the `Time taken` line of the shell output: spark-sql (default)> insert into hologresTable select * from csvTable; 25/04/11 15:05:51 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:05:52 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:05:52 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:02 WARN [main] PGStream: create socket to 25/04/11 15:06:02 WARN [main] JDBCUtil$: could not connect directly 25/04/11 15:06:03 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:04 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:04 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:05 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:05 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:06 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:06 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:07 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:07 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:09 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:09 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:10 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:10 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:11 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:11 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:12 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:12 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] 25/04/11 15:06:13 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker] Time taken: 228.901 seconds
Average data write duration	The average duration of the write job itself, excluding time spent on Spark cluster scheduling, data reading, or data reshuffling. On the HoloWeb page of the Hologres console, run the following SQL statement to obtain the shard concurrency (the value from `COUNT()`) and the average data write duration (`avg_duration_ms`) in milliseconds: `SELECT COUNT(), AVG(duration) AS avg_duration_ms FROM hologres.hg_query_log WHERE query_start >= '<start_time>' AND query_start <= '<end_time>' AND query LIKE '%<hologres_table_name>%' AND command_tag = 'COPY';` The following table describes the parameters. start_time: The start time of the data write operation, such as `2024-04-11 15:00:00`. end_time: The end time of the data write operation, such as `2024-04-11 15:10:00`. hologres_table_name: The name of the destination Hologres table, such as `your_table_name`.
Hologres load	The CPU utilization of the Hologres instance. You can obtain the CPU utilization on the Monitoring Information page of the instance in the Hologres console.
Spark load	The load on the EMR nodes. In the EMR on ECS console, navigate to the Monitoring and Diagnostics > Metric Monitoring tab for your cluster and configure the following settings: Dashboard: Select HOST. nodeGroupId: Select the ID of the core node group of the cluster. hostname: Select the core nodes. Select Time: Select the time period that covers the Spark data write operation. Then, query the CPU utilization metric to view the Spark load.

The following table shows the detailed test results.

Storage format	Primary key	Write mode	Total job duration	Average write duration	Hologres load	Spark load
Column-oriented storage	No primary key	insert	241.61s	232.70s	92%	15%
		stream	228.11s	222.34s	100%	36%
		bulk_load	88.72s	57.16s	97%	47%
	With primary key ignore	insert	190.96s	172.60s	90%	14%
		stream	149.60s	142.16s	100%	14%
		bulk_load_on_conflict	115.96s	42.92s	60%	75%
	With primary key replace	insert	600.40s	574.31s	91%	5%
		stream	550.29s	540.32s	100%	5%
		bulk_load_on_conflict	188.05s	109.77s	93%	78%
Row-oriented storage	No primary key	insert	132.38s	123.79s	94%	22%
		stream	114.41s	103.81s	100%	17%
		bulk_load	68.20s	41.22s	98%	32%
	With primary key ignore	insert	190.48s	170.49s	89%	15%
		stream	185.46s	172.48s	85%	14%
		bulk_load_on_conflict	117.81s	47.69s	58%	75%
	With primary key replace	insert	177.97s	170.78s	93%	15%
		stream	142.44s	130.16s	100%	20%
		bulk_load_on_conflict	137.69s	65.18s	92%	78%
Row-column hybrid storage	With primary key ignore	insert	172.19s	158.74s	86%	16%
		stream	150.63s	149.76s	100%	12%
		bulk_load_on_conflict	128.83s	42.09s	59%	79%
	With primary key replace	insert	690.37s	662.00s	92%	5%
		stream	625.84s	623.08s	100%	4%
		bulk_load_on_conflict	202.07s	121.58s	93%	80%

Note

The write modes correspond to the following operations:

insert: INSERT INTO VALUES
stream: FIXED COPY
bulk_load: COPY import to a table without a primary key
bulk_load_on_conflict: COPY import to a table with a primary key