Hologres:Batch writing to Hologres

Configuração do ambiente

Antes de começar, certifique-se de ter o seguinte ambiente configurado:

• Provisione uma instância do Hologres V2.2.25 ou posterior e crie um banco de dados.

• Crie um cluster EMR Spark que execute o Spark 3.3.0 ou posterior. Para mais informações, consulte Criar um cluster.

• Baixe o pacote do Spark-Connector.

  Para permitir que o Spark leia e escreva no Hologres, baixe o arquivo JAR hologres-connector-spark-3.x no Repositório Central Maven.

A tabela a seguir descreve o ambiente utilizado para o teste de desempenho.

Preparação dos dados de teste

1. Prepare os dados de origem.

   1. Faça login no nó mestre do cluster EMR Spark. Para mais informações, consulte Conectar-se a uma instância ECS.

   2. Baixe o pacote de ferramentas TPC-H TPC-H_Tools_v3.0.0.zip. Copie o pacote para a instância ECS do nó mestre, extraia-o e navegue até o diretório TPC-H_Tools_v3.0.0/TPC-H_Tools_v3.0.0/dbgen.

   3. Execute o seguinte comando no diretório dbgen para gerar um arquivo customer.tbl para um conjunto de dados de teste de 1 TB.

      ./dbgen -s 1000 -T c

      A tabela a seguir descreve a tabela customer no conjunto de dados TPC-H de 1 TB.

      Item	Descrição
      Número de campos	8
      Tipos de campo	INT, BIGINT, TEXT, DECIMAL
      Número de linhas	150.000.000
      Número de shards	40

2. Importe os dados de teste para o Spark.

   Execute o seguinte comando para carregar o arquivo customer.tbl no cluster Spark.

   hadoop fs -put customer.tbl <spark_resource>

   No comando, spark_resource refere-se ao Root Storage Directory of Cluster definido durante a criação do cluster EMR Spark.

3. Armazenamento híbrido linha-coluna

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

   Armazenamento orientado a colunas (com chave primária)

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

   Armazenamento orientado a colunas (sem chave primária)

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

   Armazenamento orientado a linhas (com chave primária)

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

   Armazenamento orientado a linhas (sem chave primária)

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

Configuração do teste

Este tópico testa o desempenho de importação de diferentes modos.

1. Faça login no nó mestre do cluster EMR Spark. Para mais informações, consulte Conectar-se a uma instância ECS. Carregue o pacote baixado do Spark-Connector e execute o seguinte comando para entrar no shell interativo do Spark SQL.

   Nota

   É possível ajustar o valor do parâmetro spark.sql.files.maxPartitionBytes para controlar a concorrência de leitura de arquivos HDFS no Spark. Neste exemplo, a concorrência está definida como 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

   No comando, path é o diretório que contém o arquivo hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar.

2. No shell interativo do Spark SQL, execute as seguintes instruções SQL para criar visualizações temporárias e escrever dados.

   Uma tabela temporária é usada neste teste para facilitar o ajuste de parâmetros ao avaliar o desempenho de escrita de diferentes modos.

   Nota

   Na prática, também é possível usar um catálogo para carregar tabelas do Hologres diretamente.

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

   A tabela a seguir descreve os parâmetros.

   Parâmetro	Descrição
   spark_resources	O cluster root storage path configurado ao criar o cluster EMR Spark. Esse caminho encontra-se na seção Cluster Information, na página Basic Information do seu cluster, no console EMR on ECS.
   hologres_vpc_endpoint	O Domain Name da VPC da instância do Hologres. Faça login no console do Hologres, clique em ID da instância desejada e localize o endpoint da VPC especificada na seção de informações de rede da página de detalhes da instância. Por exemplo, um endpoint de VPC na região China (Hangzhou) tem o formato <Instance ID>-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80.
   database_name	Nome do banco de dados na instância do Hologres.
   accesskey_id	AccessKey ID com permissões de leitura no banco de dados especificado do Hologres.
   accesskey_secret	AccessKey secret com permissões de leitura no banco de dados especificado do Hologres.
   table_name	Nome da tabela do Hologres de destino.
   write.mode	Modo de escrita. Valores válidos: auto: Valor padrão. O conector seleciona automaticamente o modo ideal. insert: Escreve dados usando instruções INSERT INTO VALUES. stream: Realiza escritas por streaming usando FIXED COPY. bulk_load: Realiza uma importação em lote para uma tabela sem chave primária usando COPY. bulk_load_on_conflict: Realiza uma importação em lote para uma tabela com chave primária usando COPY.
   write.insert.thread_size	Concorrência de escrita. Aplica-se apenas ao modo insert.
   write.insert.batch_size	Tamanho do lote de escrita. Aplica-se apenas ao modo insert.
   write.on_conflict_action	INSERT_OR_REPLACE (padrão): Se ocorrer um conflito de chave primária, os dados serão atualizados. INSERT_OR_IGNORE: Se ocorrer um conflito de chave primária, os dados serão ignorados.

   Para mais informações sobre os parâmetros, consulte Parâmetros.

Cenários de teste

Resultados do teste

Os resultados do teste incluem os seguintes 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';

A tabela a seguir mostra os resultados detalhados do teste.