PolarDB for PostgreSQL の DynamoDB 互換機能を使用する方法について説明します。これには、機能の有効化、エンドポイントとアカウントの設定、接続の確認が含まれます。 - PolarDB

このガイドでは、クラスターを作成し、PolarDB for PostgreSQL およびの DynamoDB 互換機能に初めて接続する方法について説明します。

前提条件

DynamoDB 互換機能を使用する前に、ご利用のクラスターが次の要件を満たしていることを確認してください。

カーネルバージョン：
- PostgreSQL 16、マイナーカーネルバージョン 2.0.16.13.16.0 以降。
データベースプロキシバージョン： 2.3.63.1 以降。

説明

PolarDBコンソールに移動し、設定と管理 > バージョン管理 ページでクラスターのバージョンを確認します。バージョンが要件を満たしていない場合は、マイナーバージョンをアップグレード。

DynamoDB 互換機能の有効化

クラスターの DynamoDB 互換機能は、次の 2 つの方法のいずれかで有効にできます。

既存のクラスター

ご利用のクラスターが前提条件を満たしている場合は、クラスターの 基本情報 ページに移動します。DynamoDB 互換 設定を見つけて 有効化 をクリックします。

新規クラスター

ご利用のクラスターが前提条件を満たしていない場合、または新規クラスターでこの機能を使用したい場合は、PolarDBコンソールで新しいクラスターを作成できます。

PolarDB クラスターのカスタム購入ページに移動します。
データベースエンジンを PostgreSQL 16 に設定し、[DynamoDB 互換性] チェックボックスをオンにします。
リージョン、ゾーン、ノードスペックなど、その他のオプションを設定します。

専用の DynamoDB アカウントの作成

専用の DynamoDB アカウントを作成して、API アクセス用の認証情報 (AccessKey) を取得します。

ご利用のクラスターの 設定と管理 > アカウント管理 ページで、アカウントの作成 をクリックします。
[アカウントタイプ] を DynamoDB アカウント に設定し、アカウント名とパスワードを設定します。
説明
このパスワードはコンソールでのアカウント管理専用であり、API アクセスには使用しません。
たとえば、アカウント名を dynamodb_user に設定します。アカウント名は、英字で始まり、英字または数字で終わる必要があります。また、小文字の英字、数字、アンダースコア (_) のみを含み、16 文字を超えてはなりません。パスワードは 8～32 文字で、大文字、小文字、数字、特殊文字のうち少なくとも 3 種類を含める必要があります。
アカウントが作成されたら、アカウントリストでそのアカウントを見つけます。アカウント名は Access Key ID です。SecretAccess Key を取得するには、[キー] 列の [表示] をクリックします。

DynamoDB エンドポイントの設定

DynamoDB 互換機能を有効にすると、システムは自動的にデフォルトの DynamoDB エンドポイントを作成します。カスタムエンドポイントを作成することもできます。

クラスターの 基本情報 ページで、データベースへの接続 セクションにある、システムによって作成されたデフォルトの DynamoDB を見つけます。
(任意) カスタムクラスターエンドポイントの作成 をクリックし、エンドポイントタイプを DynamoDB に設定します。その後、アタッチするノードを指定し、このエンドポイントのデータ整合性レベルをプリセットできます。

接続方法とホワイトリスト

サーバーまたはクライアントの場所に基づいて接続方法を選択します。

プライベート接続 (推奨) アプリケーションが Alibaba Cloud ECS にデプロイされている場合は、ECS インスタンスと PolarDB クラスターが同じ VPC 内にあることを確認してください。また、ECS インスタンスの IP アドレスまたはセキュリティグループをクラスターのホワイトリスト必要があります。
インターネット接続 ローカル環境で開発・テストを行う必要がある場合、または ECS インスタンスと PolarDB クラスターが同じ VPC にない場合は、クラスターの 基本情報 ページの データベースへの接続 セクションで、DynamoDB のインターネットエンドポイントを申請できます。また、ローカル環境の IP アドレスまたは ECS インスタンスのパブリック IP アドレスをクラスターのホワイトリスト必要があります。
[DynamoDB エンドポイント] カードで、対象のエンドポイントの横にある [パブリックエンドポイントの申請] をクリックします。すると、システムは対応するインターネットエンドポイントとポート情報を生成します。
説明
インターネット経由でクラスターにアクセスすると、セキュリティリスクが生じます。インターネットエンドポイントは、一時的な開発およびテスト目的でのみ使用することを推奨します。終了したら、インターネットエンドポイントを解放してください。

接続と検証

Boto3 SDK を使用した次の Python コードで設定を検証します。この例では、Alibaba Cloud Linux 3.2104 LTS 64 ビット オペレーティングシステムを実行している ECS インスタンスを使用します。

プロジェクトディレクトリに移動します。この例では /home/testDynamoDB を使用します。
```
mkdir /home/testDynamoDB
cd /home/testDynamoDB
```
/home/testDynamoDB ディレクトリに、仮想環境 (venv) を作成して、プロジェクトの依存関係を分離し、グローバルパッケージとの競合を防ぎます。
```
python3 -m venv myenv
```
仮想環境をアクティベートします。
```
source myenv/bin/activate
```
必要な Python の依存関係をインストールします。
```
pip3 install boto3
```

Python ファイルを作成し、次のコードをコピーします。endpoint_url、aws_access_key_id、aws_secret_access_key を、ご利用の PolarDB クラスターのエンドポイントとDynamoDB アカウントの認証情報に置き換えます。

vim main.py

import boto3
from botocore.exceptions import ClientError
# 1. 接続詳細の設定
# PolarDB クラスターの DynamoDB 互換機能のエンドポイント。
endpoint_url = "http://<your-polardb-ddb-endpoint>:<port>"
# DynamoDB アカウントの Access Key ID。
aws_access_key_id = "<your-access-key-id>"
# DynamoDB アカウントの Secret Access Key。
aws_secret_access_key = "<your-secret-access-key>"
# PolarDB の DynamoDB 互換機能では、region_name を "public" に設定する必要があります。
region_name = "public"
# 2. DynamoDB リソースクライアントの作成
dynamodb = boto3.resource(
    'dynamodb',
    endpoint_url=endpoint_url,
    aws_access_key_id=aws_access_key_id,
    aws_secret_access_key=aws_secret_access_key,
    region_name=region_name,
)
table_name = 'usertable'
# 3. データベース操作のための関数の定義
def create_table():
    """テーブルを作成するか、すでに存在する場合は取得します。"""
    try:
        table = dynamodb.create_table(
            TableName=table_name,
            KeySchema=[{'AttributeName': 'userid', 'KeyType': 'HASH'}], # パーティションキー
            AttributeDefinitions=[{'AttributeName': 'userid', 'AttributeType': 'S'}]
            # ProvisionedThroughput は任意であり、PolarDB では無視されます。
        )
        print("テーブルを作成中... アクティブになるのを待っています。")
        table.meta.client.get_waiter('table_exists').wait(TableName=table_name)
        print("テーブルはアクティブです！")
        return table
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceInUseException':
            print("テーブルはすでに存在します。")
            return dynamodb.Table(table_name)
        else:
            raise
def put_items(table):
    """テーブルにいくつかの項目を挿入します。"""
    users = [
        {'userid': 'user01', 'name': 'Alice', 'age': 24},
        {'userid': 'user02', 'name': 'Bob', 'age': 30},
        {'userid': 'user03', 'name': 'Charlie', 'age': 28}
    ]
    for user in users:
        table.put_item(Item=user)
    print("3 つの項目を挿入しました。")
def scan_table(table):
    """テーブル内のすべての項目をスキャンして表示します。"""
    response = table.scan()
    items = response.get('Items', [])
    print(f"{len(items)} 個の項目をスキャンしました：")
    for item in items:
        print(item)
def delete_table(table):
    """テーブルを削除します。"""
    table.delete()
    print("テーブルを削除中... 削除されるのを待っています。")
    table.meta.client.get_waiter('table_not_exists').wait(TableName=table_name)
    print("テーブルが正常に削除されました。")
# 4. 検証プロセスの実行
if __name__ == '__main__':
    try:
        user_table = create_table()
        put_items(user_table)
        scan_table(user_table)
    finally:
        # エラーが発生した場合でも、テーブルがクリーンアップされるようにします。
        if 'user_table' in locals():
            delete_table(user_table)

スクリプトを実行すると、次のような出力が表示されます。

python3 main.py

Creating table... Waiting for it to become active.
Table is active!
Inserted 3 items.
Scanned 3 items:
{'age': Decimal('24'), 'name': 'Alice', 'userid': 'user01'}
{'age': Decimal('30'), 'name': 'Bob', 'userid': 'user02'}
{'age': Decimal('28'), 'name': 'Charlie', 'userid': 'user03'}
Deleting table... Waiting for it to be removed.
Table deleted successfully.

この出力は、接続が成功し、設定が検証されたことを示します。