Hologres:Python

Pré-requisitos

Python 3.7 ou posterior instalado.

Instale o Psycopg 3

pip  install   --upgrade pip             # Upgrade pip to version 20.3 or later.
pip install "psycopg[binary]"

Conectar-se ao Hologres

Após instalar o Psycopg 3, siga as etapas abaixo para se conectar ao Hologres.

1. Carregue o Psycopg 3.

   Execute o comando a seguir para carregar o Psycopg 3 instalado:

   import psycopg

2. Crie uma conexão com o banco de dados.

   Use a função psycopg.connect() para estabelecer a conexão com o Hologres. O código a seguir apresenta a sintaxe e os parâmetros:

   conn = psycopg.connect(
       host="<Endpoint>",
       port=<Port>, 
       dbname="<databases>", 
       user="<Access ID>", 
       password="<Access Key>",
       keepalives=<keepalives>, 
       keepalives_idle=<keepalives_idle>,
       keepalives_interval=<keepalives_interval>, 
       keepalives_count=<keepalives_count>
   )
   

   Parâmetro	Descrição
   Endpoint	O endpoint e o número da porta da instância do Hologres. Importante Selecione o endpoint e o número da porta corretos conforme o ambiente de rede onde o código será executado. Caso contrário, a conexão falhará.
   Port
   databases	Nome do banco de dados do Hologres.
   Access ID	ID do AccessKey da conta Alibaba Cloud usada para conectar-se à instância do Hologres. Acesse a página AccessKey para obter o ID do AccessKey.
   Access Key	Segredo do AccessKey da conta Alibaba Cloud usada na conexão com a instância do Hologres.
   keepalives	Opcional. Defina o método de conexão. Recomendamos configurar este parâmetro. Valores válidos: 1: usa conexão persistente. 0: usa conexão não persistente.
   keepalives_idle	Intervalo de envio de mensagens keepalive quando a conexão com o banco de dados do Hologres entra em estado ocioso. Unidade: segundos.
   keepalives_interval	Tempo de espera para reenviar uma mensagem keepalive caso nenhuma resposta seja recebida. Unidade: segundos.
   keepalives_count	Número máximo de tentativas de envio de mensagens keepalive.

   Exemplo:

   conn = psycopg.connect(
       host="<Endpoint>",
       port=<Port>, 
       dbname="<databases>", 
       user="<Access ID>", 
       password="<Access Key>",
       keepalives=1, # Maintain the connection.
       keepalives_idle=130, # Send a keepalive message every 130 seconds when the connection is idle.
       keepalives_interval=10, # Wait 10 seconds before resending a keepalive message if no response is returned.
       keepalives_count=15, # Send a keepalive message up to 15 times.
       application_name="<Application Name>"
   )

   Nota

   Configurar o parâmetro Application Name facilita a identificação rápida da aplicação que originou a requisição na lista histórica de consultas lentas.

Usar o Hologres

Após conectar-se ao banco de dados do Hologres, use o Psycopg 3 para desenvolver dados. Siga as etapas abaixo para criar uma tabela, inserir dados, consultá-los e liberar os recursos. Para obter maior desempenho em operações de leitura e escrita com o recurso fixed plan, configure os parâmetros GUC correspondentes. Para mais informações, consulte Acelerar a execução de SQL com fixed plans.

1. Crie um cursor.

   Antes de iniciar o desenvolvimento de dados, execute o comando cur = conn.cursor() para criar um cursor associado à conexão.

2. Desenvolva os dados.

   1. Crie uma tabela.

      Execute o comando abaixo para criar uma tabela chamada holo_test com tipo de dado inteiro. Adapte o nome da tabela e o tipo de dado conforme as necessidades do seu negócio.

      cur.execute("CREATE TABLE holo_test (num integer);")

   2. Insira dados na tabela.

      Use o comando a seguir para inserir valores de 1 a 1000 na tabela holo_test criada anteriormente.

      cur.execute("INSERT INTO holo_test SELECT generate_series(%s, %s)", (1, 1000))

   3. Consulte os dados da tabela.

      cur.execute("SELECT sum(num) FROM holo_test;")
      cur.fetchone()

3. Confirme as transações.

   O exemplo anterior envolve operações DDL, DML e DQL. Execute o comando conn.commit() após cada instrução SQL para confirmar a transação e garantir a persistência da operação. Recomendamos definir o parâmetro autocommit como true logo após o código de conexão conn para confirmar automaticamente os comandos SQL. Veja os exemplos a seguir:

   • Exemplo de invocação síncrona

     conn = psycopg.connect(
         host="<Endpoint>",
         port=<Port>, 
         dbname="<databases>", 
         user="<Access ID>", 
         password="<Access Key>",
         keepalives=1, # Maintain the connection.
         keepalives_idle=130, # Send a keepalive message every 130 seconds when the connection is idle.
         keepalives_interval=10, # Wait 10 seconds before resending a keepalive message if no response is returned.
         keepalives_count=15, # Send a keepalive message up to 15 times.
         application_name="<Application Name>"
     )
     conn.autocommit = "True"

   • Exemplo de invocação assíncrona

     async with await psycopg.AsyncConnection.connect(
         host="<Endpoint>",
         port=<Port>, 
         dbname="<databases>", 
         user="<Access ID>", 
         password="<Access Key>",
         application_name="<Application Name>",
         autocommit = "True"
         ) as aconn:
         async with aconn.cursor() as acur:
             await acur.execute(
                 "INSERT INTO test (num, data) VALUES (%s, %s)",
                 (100, "abc'def"))
             await acur.execute("SELECT * FROM test")
             await acur.fetchone()
             # will return (1, 100, "abc'def")
             async for record in acur:
                 print(record)

4. Libere os recursos.

   Para evitar impactos em operações futuras, execute os comandos abaixo ao concluir as tarefas anteriores. Isso fechará o cursor e encerrará a conexão do Psycopg 3 com o banco de dados do Hologres:

   cur.close()
   conn.close()

Melhor prática para importar DataFrames eficientemente para o Hologres

No uso de Python, é comum utilizar o Pandas para converter registros de dados em um DataFrame, processá-lo e depois importá-lo para o Hologres. Nesses cenários, agilizar a importação é essencial.

# pip install Pandas==1.5.1

import psycopg
import pandas as pd

# Connect to Hologres.
conn = psycopg.connect(
    host="hgpostcn-cn-xxxxx-cn-hangzhou.hologres.aliyuncs.com",
    port=80,
    dbname="db",
    user="xxx",
    password="xxx",
    application_name="psycopg3"
)

cur = conn.cursor()

# Delete redundant tables.
cur.execute("""
            DROP TABLE IF EXISTS df_data;
            """)
conn.commit()

# Create a test table for data import.
cur.execute("""
            CREATE TABLE IF NOT EXISTS df_data(
                col1 int,
                col2 int,
                col3 int,
                primary key(col1)
            );
            """)
conn.commit()

# Build a DataFrame.
data = [('1','1','1'),('2','2','2')]
cols = ('col1','col2','col3')
pd_data = pd.DataFrame(data, columns=cols)

# Write data in batches.
# Use StringIO to convert the DataFrame into a CSV-formatted string.
from io import StringIO

# Create a buffer.
buffer = StringIO()
        
# Write the DataFrame to the buffer in CSV format.
pd_data.to_csv(buffer, index=False, header=False)
        
# Reset the buffer position to the beginning.
buffer.seek(0)

with cur.copy("COPY df_data(col1,col2,col3) FROM STDIN WITH (STREAM_MODE TRUE,ON_CONFLICT UPDATE,FORMAT CSV);") as copy:
    while data := buffer.read(1024):
        copy.write(data)
conn.commit()

# Query the data.
cur.execute("SELECT * FROM df_data")
cur.fetchone()
cur.commit()