Hologres:Prácticas recomendadas para la escritura por lotes en Hologres

Configuración del entorno

Antes de empezar, asegúrese de contar con el siguiente entorno configurado:

• Aprovisione una instancia de Hologres V2.2.25 o posterior y create a database.

• Cree un clúster EMR Spark que ejecute Spark 3.3.0 o posterior. Para obtener más información, consulte Create a cluster.

• Descargue el paquete Spark-Connector.

  Para permitir que Spark lea y escriba en Hologres, descargue el archivo JAR hologres-connector-spark-3.x desde el repositorio central de Maven.

La siguiente tabla describe el entorno utilizado para la prueba de rendimiento.

Preparación de datos de prueba

1. Prepare los datos de origen.

   1. Inicie sesión en el nodo maestro del clúster EMR Spark. Para obtener más información, consulte Connect to an ECS instance.

   2. Descargue el paquete de herramientas TPC-H TPC-H_Tools_v3.0.0.zip. A continuación, copie el paquete en la instancia ECS del nodo maestro, extráigalo y vaya al directorio TPC-H_Tools_v3.0.0/TPC-H_Tools_v3.0.0/dbgen.

   3. Ejecute el siguiente comando en el directorio dbgen para generar un archivo customer.tbl correspondiente a un conjunto de datos de prueba de 1 TB.

      ./dbgen -s 1000 -T c

      La siguiente tabla describe la tabla customer en el conjunto de datos TPC-H de 1 TB.

      Elemento	Descripción
      Número de campos	8
      Tipos de campo	INT, BIGINT, TEXT, DECIMAL
      Número de filas	150,000,000
      Número de shards	40

2. Importe los datos de prueba a Spark.

   Ejecute el siguiente comando para cargar el archivo customer.tbl en el clúster Spark.

   hadoop fs -put customer.tbl <spark_resource>

   En el comando, spark_resource hace referencia al Root Storage Directory of Cluster que estableció al crear el clúster EMR Spark.

3. Almacenamiento híbrido de filas y columnas

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

   Almacenamiento orientado a columnas (con clave principal)

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

   Almacenamiento orientado a columnas (sin clave principal)

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

   Almacenamiento orientado a filas (con clave principal)

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

   Almacenamiento orientado a filas (sin clave principal)

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

Configuración de la prueba

Este tema evalúa el rendimiento de importación de los diferentes modos.

1. Inicie sesión en el nodo maestro del clúster EMR Spark. Para obtener más información, consulte Connect to an ECS instance. Cargue el paquete Spark-Connector descargado y ejecute el siguiente comando para ingresar al shell interactivo de Spark SQL.

   Nota

   Puede ajustar el valor del parámetro spark.sql.files.maxPartitionBytes para controlar la concurrencia en la lectura de archivos HDFS desde Spark. En este ejemplo, la concurrencia se establece en 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

   En el comando, path es el directorio que contiene el archivo hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar.

2. En el shell interactivo de Spark SQL, ejecute las siguientes sentencias SQL para crear vistas temporales y escribir datos.

   En esta prueba se utiliza una tabla temporal para facilitar el ajuste de parámetros al evaluar el rendimiento de escritura de los distintos modos.

   Nota

   En la práctica, también puede utilizar catalog para cargar tablas de Hologres directamente.

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

   La siguiente tabla describe los parámetros.

   Parámetro	Descripción
   spark_resources	La cluster root storage path que configura al crear el clúster EMR Spark. Puede encontrar esta ruta en la sección Cluster Information, dentro de la página Basic Information de su clúster en la consola EMR on ECS.
   hologres_vpc_endpoint	El Domain Name de VPC de la instancia de Hologres. Inicie sesión en la consola de Hologres, haga clic en el ID de la instancia objetivo y busque el endpoint de la VPC especificada en la sección de información de red de la página de detalles de la instancia. Por ejemplo, un endpoint de VPC en la región China (Hangzhou) tiene el formato <Instance ID>-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80.
   database_name	El nombre de la base de datos en la instancia de Hologres.
   accesskey_id	El AccessKey ID con permisos de lectura en la base de datos de Hologres especificada.
   accesskey_secret	El AccessKey secret con permisos de lectura en la base de datos de Hologres especificada.
   table_name	El nombre de la Hologres table de destino.
   write.mode	El modo de escritura. Valores válidos: auto: Valor predeterminado. El conector selecciona automáticamente el modo óptimo. insert: Escribe datos mediante sentencias INSERT INTO VALUES. stream: Realiza escrituras mediante streaming utilizando FIXED COPY. bulk_load: Realiza una importación por lotes a una tabla sin clave principal mediante COPY. bulk_load_on_conflict: Realiza una importación por lotes a una tabla con clave principal mediante COPY.
   write.insert.thread_size	La concurrencia de escritura. Este parámetro aplica únicamente al modo insert.
   write.insert.batch_size	El tamaño del lote de escritura. Este parámetro aplica únicamente al modo insert.
   write.on_conflict_action	INSERT_OR_REPLACE (predeterminado): Si ocurre un conflicto de clave principal, los datos se actualizan. INSERT_OR_IGNORE: Si ocurre un conflicto de clave principal, los datos se ignoran.

   Para obtener más información sobre los parámetros, consulte Parameters.

Escenarios de prueba

Resultados de las pruebas

Los resultados de las pruebas incluyen los siguientes campos:

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

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

La siguiente tabla muestra los resultados detallados de las pruebas.