Hologres:Best practices for batch writing data to Hologres

Prepare basic environments

You need to prepare the following environment:

• Purchase a Hologres instance of V2.2.25 or later, and create a database.

• Create an EMR Spark cluster. The Spark version must be greater than or equal to 3.3.0). For more information, see Create a cluster.

• Download the Spark-Connector package.

  You can download the connector JAR package hologres-connector-spark-3.x, which is required for Spark to read data from and write data to Hologres, from the Maven Central Repository.

The following environment information is used in this topic.

Prepare test data

1. Prepare raw data.

   1. Log on to the master node of the EMR Spark cluster. For more information, see Log on to the master node of a cluster by connecting to the specific ECS instance.

   2. Click TPC-H_Tools_v3.0.0.zip to download the TPC-H tool, copy it to the ECS instance of the master node, extract it, and then enter the TPC-H_Tools_v3.0.0/TPC-H_Tools_v3.0.0/dbgen directory.

   3. Execute the following code to generate the 1 TB test dataset file customer.tbl in the dbgen directory. The original tbl file size is 23 GB.

      ./dbgen -s 1000 -T c

      The following table provides the detailed information of the customer table in the TPC-H 1 TB dataset.

      Table information	Description
      Number of fields	8
      Field types	INT, BIGINT, TEXT, DECIMAL
      Number of data rows	150,000,000
      Number of table shards	40

2. Import test data to Spark.

   Execute the following command to upload the customer.tbl file to the Spark cluster:

   hadoop fs -put customer.tbl <spark_resource>

   spark_resource indicates the path that is configured for the Root Storage Directory of Cluster parameter when you create the EMR Spark cluster.

3. Create tables with different storage formats in Hologres. Sample SQL statements:

   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'
   );

Test configuration

This topic mainly tests the import performance under different modes.

1. Log on to the master node of the EMR Spark cluster (for the logon method, see Log on to the master node of a cluster by connecting to the specific ECS instance), upload the downloaded Spark-Connector package, and then run the following commands to enter the spark-sql command-line interface (CLI).

   Note

   By adjusting the value of the spark.sql.files.maxPartitionBytes parameter, you can control the concurrency of Spark reading HDFS files. The concurrency in this example is 40.

   # Enter the spark-sql CLI.
   
   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

   path indicates the root path where the hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar package is located.

2. Execute the following SQL statements in the spark-sql CLI to write data by creating temporary tables.

   Since parameters need to be adjusted multiple times during the testing process to test the write performance of different modes, temporary tables are used in this example.

   Note

   In actual business scenarios, you can directly use catalogs to load Hologres tables, which is more convenient.

   -- 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 path that is configured for the Root Storage Directory of Cluster parameter when you create the EMR Spark cluster. You can log on to the EMR on ECS console, click the target cluster ID, and obtain the root storage path of the cluster in the Cluster Information section of the Basic Information tab.
   hologres_vpc_endpoint	The endpoint of the VPC network type of the Hologres instance. You can log on to the Hologres console, click the target instance ID, and obtain the VPC endpoint in the Network Information section of the Instance Details page. For example, the VPC endpoint in the China (Hangzhou) region is in the format of <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 corresponding Hologres database.
   accesskey_secret	The AccessKey secret with read permissions on the corresponding Hologres database.
   table_name	The name of the destination Hologres table.
   write.mode	The write mode. Valid values: auto: Default value, indicating that the connector automatically selects the optimal mode . insert: Use the INSERT INTO VALUES mode. stream: Use the FIXED COPY mode for streaming writing. bulk_load: Use the COPY mode for batch import to tables without primary keys. bulk_load_on_conflict: Use the COPY mode for batch import to tables with primary keys.
   write.insert.thread_size	The write concurrency, only effective when using the INSERT INTO VALUES mode.
   write.insert.batch_size	The batch write size, only effective when using the INSERT INTO VALUES mode.
   write.on_conflict_action	INSERT_OR_REPLACE (default): Update on primary key conflicts. INSERT_OR_IGNORE: Ignore on primary key conflicts.

   For more parameter information, see Parameters.

Test scenarios

Test results

The test results include the following fields.

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 '%<test_table_column>%' AND command_tag = 'COPY';

The following table provides the detailed test results.