Hologres:Python

Prasyarat

Python 3.7 atau versi yang lebih baru telah diinstal.

Instal Psycopg 3

pip  install   --upgrade pip             # Upgrade pip ke versi 20.3 atau yang lebih baru.
pip install "psycopg[binary]"

Hubungkan ke Hologres

Setelah Psycopg 3 diinstal, Anda dapat menghubungkan ke Hologres.

1. Impor Psycopg 3.

   Impor pustaka Psycopg 3:

   import psycopg

2. Buat koneksi database.

   Untuk menghubungkan ke Hologres, gunakan fungsi psycopg.connect() dengan parameter berikut:

   conn = psycopg.connect(
       host="<endpoint>",
       port=<Port>, 
       dbname="<database_name>", 
       user="<AccessKey ID>", 
       password="<AccessKey Secret>",
       keepalives=<keepalives>, 
       keepalives_idle=<keepalives_idle>,
       keepalives_interval=<keepalives_interval>, 
       keepalives_count=<keepalives_count>
   )
   

   Parameter	Deskripsi
   endpoint	Titik akhir dan Port dari instans Hologres. Buka Konsol Hologres. Di panel navigasi sebelah kiri, klik Instances lalu klik instans target. Di bagian Network Information pada halaman Instance Details, pilih Network Type yang sesuai dengan lingkungan jaringan tempat kode Anda dijalankan, lalu salin Domain Name-nya. Penting Pastikan Anda memilih endpoint dan Port yang benar sesuai lingkungan jaringan Anda untuk mencegah kegagalan koneksi.
   port
   dbname	Nama database yang telah Anda buat di Hologres.
   AccessKey ID	ID AccessKey Akun Alibaba Cloud Anda. Untuk mendapatkan ID AccessKey, buka halaman AccessKey.
   AccessKey Secret	Secret AccessKey Akun Alibaba Cloud Anda.
   keepalives	Disarankan. Menentukan apakah akan menggunakan koneksi persisten. Nilai yang valid: 1: menggunakan koneksi persisten. 0: menggunakan koneksi non-persisten.
   keepalives_idle	Waktu idle dalam detik sebelum probe keepalive dikirim.
   keepalives_interval	Waktu dalam detik untuk menunggu respons terhadap probe keepalive sebelum mengirim ulang.
   keepalives_count	Jumlah maksimum probe keepalive yang dapat dikirim sebelum koneksi dianggap terputus.

   Kode berikut memberikan contoh penggunaannya.

   conn = psycopg.connect(
       host="<endpoint>",
       port=<Port>, 
       dbname="<database_name>", 
       user="<AccessKey ID>", 
       password="<AccessKey Secret>",
       keepalives=1, # Aktifkan koneksi persisten.
       keepalives_idle=130, # Kirim probe keepalive setiap 130 detik pada koneksi idle.
       keepalives_interval=10, # Tunggu 10 detik untuk respons sebelum mengirim ulang.
       keepalives_count=15, # Kirim ulang hingga 15 kali sebelum menutup koneksi.
       application_name="<Application Name>"
   )

   Catatan

   Menyetel parameter application_name membantu Anda mengidentifikasi sumber aplikasi suatu kueri dengan cepat di daftar kueri lambat historis.

Gunakan Hologres

Setelah terhubung ke database Hologres, Anda dapat menggunakan Psycopg 3 untuk melakukan operasi data, seperti membuat tabel, memasukkan data, mengkueri data, dan melepaskan sumber daya. Untuk mencapai performa baca-tulis yang lebih tinggi dengan fitur Fixed Plan, Anda harus mengonfigurasi parameter GUC terkait. Untuk informasi selengkapnya, lihat Percepat eksekusi SQL menggunakan Fixed Plan.

1. Buat cursor.

   Sebelum melakukan operasi data, Anda harus menjalankan perintah cur = conn.cursor() untuk membuat cursor pada koneksi tersebut.

2. Lakukan operasi data.

   1. Buat tabel

      Jalankan perintah berikut untuk membuat tabel bernama holo_test dengan kolom integer.

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

   2. Masukkan data

      Jalankan perintah berikut untuk memasukkan bilangan bulat dari 1 hingga 1000 ke dalam tabel holo_test.

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

   3. Kueri data

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

3. Commit transaksi.

   Secara default, Psycopg memulai transaksi yang harus Anda commit secara eksplisit menggunakan conn.commit(). Untuk kenyamanan, kami menyarankan menyetel parameter autocommit ke True agar setiap perintah langsung di-commit secara otomatis. Contoh berikut menunjukkan cara melakukannya:

   • Contoh panggilan sinkron

     conn = psycopg.connect(
         host="<endpoint>",
         port=<Port>, 
         dbname="<database_name>", 
         user="<AccessKey ID>", 
         password="<AccessKey Secret>",
         keepalives=1, # Aktifkan koneksi persisten.
         keepalives_idle=130, # Kirim probe keepalive setiap 130 detik pada koneksi idle.
         keepalives_interval=10, # Tunggu 10 detik untuk respons sebelum mengirim ulang.
         keepalives_count=15, # Kirim ulang hingga 15 kali sebelum menutup koneksi.
         application_name="<Application Name>"
     )
     conn.autocommit = True

   • Contoh panggilan asinkron

     async with await psycopg.AsyncConnection.connect(
         host="<endpoint>",
         port=<Port>, 
         dbname="<database_name>", 
         user="<AccessKey ID>", 
         password="<AccessKey Secret>",
         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()
             # akan mengembalikan (1, 100, "abc'def")
             async for record in acur:
                 print(record)

4. Lepaskan sumber daya.

   Saat selesai, tutup cursor dan koneksi database untuk melepaskan sumber daya.

   cur.close()
   conn.close()

Praktik terbaik: Tulis DataFrame ke Hologres

Alur kerja umum di Python adalah memproses data dalam Pandas DataFrame sebelum mengimpornya ke Hologres. Bagian ini menunjukkan metode efisien untuk impor tersebut.

# pip install Pandas==1.5.1

import psycopg
import pandas as pd

# Hubungkan ke 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()

# Hapus tabel jika sudah ada.
cur.execute("""
            DROP TABLE IF EXISTS df_data;
            """)
conn.commit()

# Buat tabel uji untuk ingesti data.
cur.execute("""
            CREATE TABLE IF NOT EXISTS df_data(
                col1 int,
                col2 int,
                col3 int,
                primary key(col1)
            );
            """)
conn.commit()

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

# Gunakan StringIO untuk mengonversi DataFrame menjadi string berformat CSV.
from io import StringIO

# Buat buffer di memori.
buffer = StringIO()
        
# Tulis DataFrame ke buffer dalam format CSV.
pd_data.to_csv(buffer, index=False, header=False)
        
# Atur ulang posisi buffer ke awal.
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()

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