Hologres:Python

前提条件

Python 3.7 以降がインストールされていること。

Psycopg 3 のインストール

pip install --upgrade pip             # pip をバージョン 20.3 以降にアップグレードします。
pip install "psycopg[binary]"

Hologres への接続

Psycopg 3 をインストールした後、Hologres に接続できます。

1. Psycopg 3 のインポート

   Psycopg 3 ライブラリをインポートします。

   import psycopg

2. データベース接続の作成

   Hologres に接続するには、次のパラメーターを指定して psycopg.connect() 関数を使用します。

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

   パラメーター	説明
   endpoint	Hologres インスタンスのエンドポイントとポート。 Hologres コンソールに移動します。 左側のナビゲーションペインで、インスタンス一覧 をクリックし、対象のインスタンスをクリックします。 インスタンスの詳細 ページの ネットワーク情報 セクションで、コードが実行されるネットワーク環境に一致する ネットワークタイプ を選択し、その ドメイン名 をコピーします。 重要 接続の失敗を防ぐために、ネットワーク環境に適したエンドポイントとポートを選択してください。
   port
   dbname	Hologres で作成したデータベースの名前。
   AccessKey ID	Alibaba Cloud アカウントの AccessKey ID。 AccessKey ID を取得するには、AccessKey ページに移動します。
   AccessKey Secret	Alibaba Cloud アカウントの AccessKey Secret。
   keepalives	推奨。TCP キープアライブを有効にするかどうかを指定します。有効な値： 1：キープアライブを有効にします。 0：キープアライブを無効にします。
   keepalives_idle	キープアライブプローブを送信するまでのアイドル時間 (秒単位)。
   keepalives_interval	キープアライブプローブへの応答を待機してから再送信するまでの時間 (秒単位)。
   keepalives_count	接続が切断されたと判断されるまでに送信できるキープアライブプローブの最大数。

   次のコードは例を示しています。

   conn = psycopg.connect(
       host="<endpoint>",
       port=<Port>, 
       dbname="<database_name>", 
       user="<AccessKey ID>", 
       password="<AccessKey Secret>",
       keepalives=1, # キープアライブを有効にします。
       keepalives_idle=130, # アイドル接続で 130 秒ごとにキープアライブプローブを送信します。
       keepalives_interval=10, # 再送信する前に応答を 10 秒間待機します。
       keepalives_count=15, # 接続を閉じる前に最大 15 回再送信します。
       application_name="<Application Name>"
   )

   説明

   application_name パラメーターを設定すると、低速クエリ履歴リストでクエリのソースアプリケーションをすばやく特定できます。

Hologres の使用

Hologres データベースに接続した後、Psycopg 3 を使用して、テーブルの作成、データの挿入、データのクエリ、リソースの解放などのデータ操作を実行できます。固定プラン機能を使用して読み取り/書き込みパフォーマンスを向上させるには、関連する GUC パラメーターを設定する必要があります。詳細については、「固定プランを使用した SQL 実行の高速化」をご参照ください。

1. カーソルの作成

   データ操作を実行する前に、cur = conn.cursor() コマンドを実行して接続用のカーソルを作成する必要があります。

2. データ操作の実行

   1. テーブルの作成

      次のコマンドを実行して、整数列を持つ holo_test という名前のテーブルを作成します。

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

   2. データの挿入

      次のコマンドを実行して、1 から 1000 までの整数を holo_test テーブルに挿入します。

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

   3. データのクエリ

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

3. トランザクションのコミット

   デフォルトでは、Psycopg はトランザクションを開始するため、conn.commit() を使用して明示的にコミットする必要があります。利便性のため、オートコミット パラメーターを True に設定して、各コマンドを自動的にコミットすることを推奨します。次の例は、その方法を示しています。

   • 同期呼び出しの例

     conn = psycopg.connect(
         host="<endpoint>",
         port=<Port>, 
         dbname="<database_name>", 
         user="<AccessKey ID>", 
         password="<AccessKey Secret>",
         keepalives=1, # キープアライブを有効にします。
         keepalives_idle=130, # アイドル接続で 130 秒ごとにキープアライブプローブを送信します。
         keepalives_interval=10, # 再送信する前に応答を 10 秒間待機します。
         keepalives_count=15, # 接続を閉じる前に最大 15 回再送信します。
         application_name="<Application Name>"
     )
     conn.autocommit = True

   • 非同期呼び出しの例

     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()
             # (1, 100, "abc'def") が返されます。最初の列は自動インクリメントの主キーです。
             async for record in acur:
                 print(record)

4. リソースの解放

   完了したら、カーソルとデータベース接続を閉じてリソースを解放します。

   cur.close()
   conn.close()

ベストプラクティス: DataFrame を Hologres に書き込む

Python での一般的なワークフローは、Pandas DataFrame でデータを処理してから Hologres にインポートすることです。このセクションでは、このインポートの効率的な方法を示します。

# Pandas==1.5.1 をインストールします

import psycopg
import pandas as pd

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

# テーブルが既に存在する場合は削除します。
cur.execute("""
            DROP TABLE IF EXISTS df_data;
            """)
conn.commit()

# データ取り込み用のテストテーブルを作成します。
cur.execute("""
            CREATE TABLE IF NOT EXISTS df_data(
                col1 int,
                col2 int,
                col3 int,
                primary key(col1)
            );
            """)
conn.commit()

# DataFrame を作成します。
data = [('1','1','1'),('2','2','2')]
cols = ('col1','col2','col3')
pd_data = pd.DataFrame(data, columns=cols)

# StringIO を使用して、DataFrame を CSV 形式の文字列に変換します。
from io import StringIO

# インメモリバッファーを作成します。
buffer = StringIO()
        
# DataFrame を CSV 形式でバッファーに書き込みます。
pd_data.to_csv(buffer, index=False, header=False)
        
# バッファーの位置を先頭にリセットします。
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()

# データをクエリします。
cur.execute("SELECT * FROM df_data")
cur.fetchone()
cur.commit()