Boto3を利用したQLDBへのデータ追加処理

概要

QLDB を Python で操作するにあたり、Boto3 を利用したデータ追加方法のまとめとなります。

QLDB

Amazon QLDB は、フルマネージド型の台帳データベースで、SQLライクなクエリーを実行することにより、データ操作を実施することができます。

Boto3

QLDB を Python から操作するには、Boto3 を利用することで可能となります。
Boto3 には、QLDB 関連で２つのサービスが提供されており、QLDB と QLDBSession が存在します。
データ操作するには、QLDBSession を利用することになります。

データ追加処理

データを追加するには、QLDBSession の send_command 関数を呼び出し、下記の操作を実行する必要があります。

1. 初期化
2. セッションの開始
3. トランザクションの開始
4. クエリーの実行
5. コミット処理
6. セッションの終了

1.初期化

QLDBSession を初期化します。

client = boto3.client('qldb-session',
    aws_access_key_id=[アクセスキーID],
    aws_secret_access_key=[シークレットアクセスキー]
)

2.セッションの開始

トランザクションを開始し、セッショントークンを取得します。
- ledger_name には、操作したい QLDB のテーブル名を指定します。

ledger_name = 'LedgerName'
response = client.send_command(
    StartSession={
        'LedgerName': ledger_name
    },
)
session_token = response['StartSession']['SessionToken']

トランザクションの開始

トランザクションを開始し、トランザクションIDを取得します。
- トランザクションを開始するには、取得したセッショントークンを渡します。

response = client.send_command(
    SessionToken=session_token,

    StartTransaction={}
)
transaction_id = response['StartTransaction']['TransactionId']

3.クエリーの実行

クエリーを実行します。
- クエリーを実行するには、実行するステートメントと取得したトランザクションIDを渡します。
- QLDB では、PartiQL クエリ言語を利用する必要があります。
- https://docs.aws.amazon.com/ja_jp/qldb/latest/developerguide/ql-reference.html

text = "{'TestId': 1, 'TestName': 'Table', 'TestDate': `2019-10-01T12:00:00Z`}"
statement = 'INSERT INTO Test ' + text

client.send_command(
    SessionToken=session_token,

    ExecuteStatement={
        'TransactionId': transaction_id,
        'Statement': statement,
    },
)

4.コミット処理

実行したクエリーをデータベースに反映するために、コミット処理をする必要があります。
- コミット処理には、トランザクションID とコミットダイジェスト(後述)を渡します。

コミットダイジェストの作成方法

AWS から提供されている Amazon QLDB ドライバーを利用すれば、コミット処理にコミットダイジェストは必要ありませんが、その他の言語 (Boto3等)で利用するには、コミットダイジェストを独自に作成する必要があります。
コミットダイジェストは、実行したクエリーとトランザクションIDをハッシュ化した値となります。
- ハッシュ化するにあたり、AWS から提供されている Amazon Ion Python と Amazon Ion Hash Python を利用します。
- https://github.com/amzn/ion-python
- https://github.com/amzn/ion-hash-python

def to_qldb_hash(value):
    value = loads(dumps(value))
    ion_hash = value.ion_hash('SHA256')

    return ion_hash


def hash_comparator(h1, h2):
    h1_array = array('b', h1)
    h2_array = array('b', h2)

    if len(h1) != 32 or len(h2) != 32:
        raise ValueError("Invalid hash")

    for i in range(len(h1_array) - 1, -1, -1):
        difference = h1_array[i] - h2_array[i]
        if difference != 0:
            return difference
    return 0


def join_hashes_pair_wise(h1, h2):
    if len(h1) == 0:
        return h2

    if len(h2) == 0:
        return h1

    if hash_comparator(h1, h2) < 0:
        concatenated = h1 + h2
    else:
        concatenated = h2 + h1

    return concatenated 

def create_commit_digest(statement, transaction_id):
    concatenated = join_hashes_pair_wise(to_qldb_hash(statement), to_qldb_hash(transaction_id))
    new_hash_lib = sha256()
    new_hash_lib.update(concatenated)
    return new_hash_lib.digest()

コミット処理

コミットダイジェストを作成し、コミット処理を実行します。

commit_digest = create_commit_digest(statement, transaction_id)
response = client.send_command(
    SessionToken=session_token,

    CommitTransaction={
        'TransactionId': transaction_id,
        'CommitDigest': commit_digest
    },
)

5.セッションの終了

セッショントークンを渡し、作成したセッションを終了します。

client.send_command(
    SessionToken=session_token,

    EndSession={},
)

まとめ

Boto3 を利用した QLDB へのデータ追加方法に説明しました。インターネット上に情報があまりなかったので、まとめとして記載しました。今回は、Python でのやり方について記載をしましたが、他の言語で提供されている API でも参考にできるかなと思います。また、Java を利用可能であるなら、提供されているドライバーを用いた開発が一番簡単かなと思いました。ドライバーのサンプルも充実しているので。データ取得処理は気分が向いたら書こうかなと思います。