概要

前回の続きです。今回はだいぶ長い記事になってしまいました。

前回の結果
- boto3のPaginatorを使って、Webのページングに使うインデックスが作れないか試してみた。
- Paginatorの仕組み上、フィルタを適用できず、またクエリの効率も悪い。
今回やってみること
- 自分でクエリを発行し、フィルタに対応したインデックス生成の処理を作る。
- Paginatorと同じような機能を盛り込んでみる。
結果
- 最終的には思い通りのものが作れたが、効率的に使うためには使い方を考える必要がある。
- インデックスからページを取得する場合の検索範囲を最小限に抑えるために、ページの起点と終点両方のキーが必要になる。
- Table.queryの出力はPaginatorよりシンプルになるが、json出力には変換が必要になる。（後述）
結論
- これ以上複雑なことをするならRDBMSの使用を考えた方がいいと思う。
環境
- python 3.6.7
- boto3 1.9.86

Webのページングに使うインデックスって？

これです。ページ分割された任意のページに飛ぶためのリンクをここではそう呼んでいます。**インデックスじゃなくてナビゲーションの方が適切だったかもしれません。**でも、もう二本目の記事なのでこのまま行きます。

テスト用のデータ

{"itemKey":{"S":"test-record"},"itemSortKey":{"N":"0"},"itemType":{"N":"1"}}
{"itemKey":{"S":"test-record"},"itemSortKey":{"N":"1"},"itemType":{"N":"2"}}
{"itemKey":{"S":"test-record"},"itemSortKey":{"N":"2"},"itemType":{"N":"3"}}
（以下itemSortKey149まで続く）

記事のために、こんな感じのデータを使います。パーティションキー"itemKey"は"test-recored"で全件固定に、ソートキー"itemSortKey"は0-149まで、計150件格納しています。項目"itemType"は順に1-2-3を繰り返しています（各itemTypeが50ずつ）。今回はitemTypeにフィルタをかけていきます。

フィルタに対応するための課題

フィルタに対応するには、boto3のPaginatorが使えないため、Paginatorが提供してくれた便利な機能を自力で作らなければいけません。加えて、生成したインデックスからページを取得する場合にも、DynamoDBの使用上考慮する点があります。

ページの切り出し：Paginatorは引数PageSizeに従って、1ページのサイズを切り出してくれました。これはやらなきゃ話になりません。
一度のクエリで取りきれないケース：Paginatorはページ毎に自動的にLastEvaluatedKeyをExclusiveStartKeyに設定してクエリを投げてくれました。
最大件数の制限：Paginatorは引数MaxItemsで、取得する合計のレコード数（各ページの合計）を制限してくれました。今回の用途で使う機会はあまりないと思いますが、考えてみます。
生成されたインデックスからページを取得する際に、PaginatorのようにLimitが使えません。これについてはインデックスにページ終点のキーも一緒に含めて、検索範囲を限定するようにします。

実際に作ってみる

以下のコードで、各ページへのインデックスを生成します。前回よりはpythonらしいコードになりましたが、まだまだ。しかも処理が複雑になった分コードが汚いですが許してください。

FullQueryIndexer.py

import boto3
import simplejson

# インデックス生成クラスの呼び出し側の処理
def main():
    # 1.ClientでなくTableを使う
    table = boto3.resource('dynamodb').Table('page.test')
    query_params = {
        'TableName' : 'page.test',
        'ProjectionExpression' : 'itemSortKey',
        'KeyConditionExpression' : 'itemKey = :keyVal',
        'FilterExpression' : 'itemType = :filterVal',
        'ExpressionAttributeValues' : {
            ':keyVal': "test-record",
            ':filterVal': 1
        },
        'Limit': 5, # LastEvaluatedKeysの動作を確認したいのでLimitを付ける
        'PaginationConfig':{
            'PageSize': 7, # Paginatorと同じ引数にしておく
            #'MaxItems': 40 MaxItemsも指定可能 
        }
    }

    indexer = FullQueryIndexer(table=table, query_params = query_params)
    indexes = indexer.find_indexes()

    # 結果の出力
    results = simplejson.loads(simplejson.dumps(indexes))
    for result in results:
        print(result)


# インデックスを生成するクラス
class FullQueryIndexer:
    def __init__(self, *, table, query_params):
        self.table = table
        self.query_params = query_params

    def find_indexes(self):
        # クエリからPaginationConfigを抜き出す
        config = self.query_params.pop('PaginationConfig')

        # 2.処理状態の記録用クラス
        class IndexerContext:
            def __init__(self):
                self.indexes = [] # 作成されたインデックス
                self.total_items = 0 # 合計件数
                self.remaining = [] # インデックス作成途中のリスト
                self.last_evaluated_key = None # 再度クエリを投げる場合のキー
                # Paginatorと同じ方針のパラメータ
                self.page_size = config['PageSize']
                self.max_items = config.get('MaxItems', float('Inf'))

        context = IndexerContext()

        # 3.クエリ結果がなくなるまで繰り返し
        while True:
            context = self.find_from_query(query_params = self.query_params, context = context)
            
            # 3.1.これ以上クエリの必要がなければ終了
            if context.last_evaluated_key is None:
                break
            self.query_params['ExclusiveStartKey'] = context.last_evaluated_key

        return context.indexes

    # クエリ1回ごとの処理
    def find_from_query(self,*, query_params, context):
        # クエリの発行
        results = self.table.query(**query_params)

        # 4.前回残したレコードに今回のクエリの結果を継ぎ足す
        result_items = context.remaining
        result_items += results.pop('Items')
        result_size = len(result_items)
        context.remaining = [] # 次回のためにクリア

        # LastEvaluatedKeyをクエリ結果から抜き出す
        context.last_evaluated_key = results.get('LastEvaluatedKey', None)

        # 5.MaxItemsの処理 レコード数がMaxItemsを超える場合は切り詰める
        over_size = (context.total_items + len(result_items)) - context.max_items
        if over_size > 0:
            result_items = result_items[0: -over_size]
            # これ以上クエリを投げない
            context.last_evaluated_key = None

        # 6.クエリ結果からのインデックス生成
        for point in range(0, len(result_items), context.page_size):

            # 6.1.ページサイズ分を切り取る（長さが足りなくてもok）
            page_items = result_items[point : point + context.page_size]
            item_size = len(page_items)

            # 6.2.取得したページが1ページの長さに満たない場合
            # 次のクエリで結果が取れるなら続ける
            if item_size < context.page_size and context.last_evaluated_key is not None :
                context.remaining = page_items # 処理途中のリストを引き継ぐ
                return context

            # 6.3.インデックスとして追加
            context.indexes.append({
                'PageNo' : len(context.indexes)+ 1,
                'StartKey' : page_items[0],
                'TerminateKey' : page_items[-1]
            })
            context.total_items += item_size

        return context

解説

検索結果のレコードは50件になるので、ページサイズを7にして、最終ページに1件だけのページが残る、少し複雑目の結果が出るようにしています。なお、**このコードを少しいじればフィルタ対応Paginatorとしても使えるようになるはずです。¹**もちろん、フィルタをかけなくても動作します。
以下、ソースコード中のコメントに番号が振ってある部分についての解説です。

Paginatorはboto3.client('dynamodb')を使いますが、今回はboto3.resource('dynamodb')からTableを取得して使います。clientからのクエリとTableからのクエリでは戻り値に違いがあります（後述）。
クエリを複数回投げる間、状態を維持したいので、クラスにまとめています。MaxItemsは省略されたら無限大に置き換えています。
クエリの発行回数が1-n回で不定になるので、do-while的なループにしています。
1. クエリの処理が終わった後、クエリがLastEvaluatedKeyを返していなければ終了です。
ここからクエリ毎の処理に入っていますが、前回のクエリで、1ページ分に満たない要素が残った場合、次に発行されるクエリで1ページ分のサイズになる可能性があるため、前回の結果の残りに新しい結果を継ぎ足しています。
クエリにMaxItemsが指定されていた場合、今回のクエリでMaxItemsをオーバーしているかもしれないので、オーバーしていたら切り詰めます。同時に、クエリから取得したLastEvaluatedKeyも削除してしまうことで、次のクエリを発行しないようにします。
ここからはクエリ結果から取り出した1ページ毎の処理です。
1. rangeでページの起点を作り出して、そこから1ページ分先まで、クエリ結果からスライスを切り出します。スライスは長さをオーバーしてもエラーを起こさないので、何も考えずに切り出します。²
2. 切り出したページが1ページ分のサイズを持っていない場合、クエリがLastEvaluatedKeyを返していれば、現在のページの内容を確保したうえでreturnし、次のクエリを発行します。上の5の処理で、取得した件数が既にMaxItemsに達している場合、LastEvaluatedKeyを削除してあるので、この処理には入らずに、件数が足りないページのインデックスを生成して処理を終わります。
3. インデックスを生成して、総件数をカウントしています。今のページの先頭と末尾の要素を設定するだけです。PageNoは、ウェブ上の表示に使い易いように1スタートにしています。

出力されるインデックス

clientを使っていたPaginatorは'itemSortKey':{'N':'0'}のように、型の情報を持った出力をしていましたが、Table.queryではシンプルな構造になります。しかし、数値に使われるDecimal型などはそのままではjson.dumpsできないので、こちらの方の記事のような方法で変換する必要があります。³変換した後のjsonが以下です。

{'PageNo': 1, 'StartKey': {'itemSortKey': 0}, 'TerminateKey': {'itemSortKey': 18}}
{'PageNo': 2, 'StartKey': {'itemSortKey': 21}, 'TerminateKey': {'itemSortKey': 39}}
{'PageNo': 3, 'StartKey': {'itemSortKey': 42}, 'TerminateKey': {'itemSortKey': 60}}
{'PageNo': 4, 'StartKey': {'itemSortKey': 63}, 'TerminateKey': {'itemSortKey': 81}}
{'PageNo': 5, 'StartKey': {'itemSortKey': 84}, 'TerminateKey': {'itemSortKey': 102}}
{'PageNo': 6, 'StartKey': {'itemSortKey': 105}, 'TerminateKey': {'itemSortKey': 123}}
{'PageNo': 7, 'StartKey': {'itemSortKey': 126}, 'TerminateKey': {'itemSortKey': 144}}
{'PageNo': 8, 'StartKey': {'itemSortKey': 147}, 'TerminateKey': {'itemSortKey': 147}}

変換前の戻り値はこの後の例で出てきます。

生成されたインデックスからページを取得する

前回と比べて、終端キーが増えていますが、こちらはあまり変わりません。このクラスの呼び出しのコードは省略します。

PageGetter.py

class PageGetter:
    def __init__(self, table):
        self.table = table

    def get_page(self, index):
        query_params = {
            'TableName': 'page.test',
            'Select': 'SPECIFIC_ATTRIBUTES',
            'ProjectionExpression': 'itemKey,itemSortKey,itemType',
            'FilterExpression': 'itemType = :filterVal',
            'ExpressionAttributeValues': {
                ':keyVal': 'test-record',
                ':filterVal': 1
            },
            'KeyConditionExpression' : 'itemKey = :keyVal AND itemSortKey BETWEEN :StartKey AND :TerminateKey'
        }

        # 1.KeyConditionExpressionを設定する（パーティションキーと起点、終端キー）
        query_params['ExpressionAttributeValues'][':StartKey'] = index['StartKey']['itemSortKey']
        query_params['ExpressionAttributeValues'][':TerminateKey'] = index['TerminateKey']['itemSortKey']

        return self.table.query(
            **query_params
        )

出力の一部

1ページ目と最終ページを出力してみました。最終ページも取り損ねていませんね。別途ScannedCountも取得しましたが、必要最小限で抑えられています。
**しかし、この処理はページサイズ分のレコードを返すことを保証しません。**途中でテーブルのレコードに増減が起きると、ページサイズより多いレコードや、少ないレコードが返る場合があります。これは対策が思いつきません。
ここではjson.dumpsではなく、Table.queryが返す生の出力をprintしてみました。

1 ページ目 ScannedCount: 19
{'itemKey': 'test-record', 'itemSortKey': Decimal('0'), 'itemType': Decimal('1')}
{'itemKey': 'test-record', 'itemSortKey': Decimal('3'), 'itemType': Decimal('1')}
{'itemKey': 'test-record', 'itemSortKey': Decimal('6'), 'itemType': Decimal('1')}
{'itemKey': 'test-record', 'itemSortKey': Decimal('9'), 'itemType': Decimal('1')}
{'itemKey': 'test-record', 'itemSortKey': Decimal('12'), 'itemType': Decimal('1')}
{'itemKey': 'test-record', 'itemSortKey': Decimal('15'), 'itemType': Decimal('1')}
{'itemKey': 'test-record', 'itemSortKey': Decimal('18'), 'itemType': Decimal('1')}

最終ページ ScannedCount: 1
{'itemKey': 'test-record', 'itemSortKey': Decimal('147'), 'itemType': Decimal('1')}

処理の効率を考える

やりたいことはこれでできるのですが、パーティション内のレコード全件検索というのは、テーブルのサイズによっては重い処理です。
私は元々、あまり大きくないテーブルで使うつもりだった⁴のであまり心配しなくてもいいんですが、それでも他の目的に応用する時のために、効率的な方法を少し考えます。

キーにLSIやGSIを設定する。⁵
- DynamoDBでのクエリは、取得する項目に関わらずレコード全体が検索対象になって、RCUを消費します。今回のようなケースではキーとフィルタする対象以外必要ないので、LSIやGSIを使って検索対象を絞り込んだ方が良さそうです。ただし、LSIやGSIの追加によって、必要なWCUも増加します。
インデックスを生成するタイミングを減らす。
- 最近のWebアプリケーションであれば、いちど生成したインデックスはフロントエンドで保持してもらうことで、ある程度は生成の回数を減らせるはずです。
- 生成したインデックスを別のテーブルに格納しておくことも有効です。この場合、インデックスの内容に変化のある、テーブルへの書き込み処理を行ったタイミングでインデックスを再生成すれば、概ね最新の結果を保てると思います。

こんなケースには向いていない

ページを辿るインデックスはWebではよくあるUIですし、便利ではありますが、DynamoDBがこういう使い方に向いているかどうかは悩ましいところです。向いていないケースを考えてみます。

自由度の高い検索
- フィルターをかけられるといっても、色んな項目を対象にしたり、「ユーザが自由に検索条件を指定できる」ような用途だと、その都度インデックス生成の処理が必要になるので、重い処理になります。これはよく言われる、**DynamoDBをRDBMS的に使うべきではない。**という話です。
変更の頻繁なテーブル
- テーブルに対して、頻繁に追加や削除が行われると、インデックスから取得できるページの内容が不正確になるケースが増えます。そういう場合はWebページにインデックスを置くことそのものが不向きな気がします。⁶
大きすぎるテーブルやパーティション
- パーティション内を全件検索するので、あまりに大きいテーブルを対象にするのはやはり無理があります。どうしてもやるなら、一度に取得するページ数を制限して、必要なら続きのページを取得する。というような方法が考えられますが、今回は考えないことにしました。

結論

目的は達成できましたが、DynamoDBの使い方としては、RDBMSを使えと言われるぎりぎり手前の方法じゃないかと思います。
そうは言っても、DynamoDBとRDBMSでは管理の手間や性能とコスト⁷も違ってくるので、DynamoDBでできるだけの方法を覚えておくのも悪くはないかなと思いました。

元々のPaginatorにも、フィルタをかけて処理したい状況って、なくもない気がします。 ↩
スライスに取らなくても、リスト上の位置を計算すれば無駄なメモリを使わずに済むのですが、長さが足りない場合のコードが読みにくいのと、注釈1のようにPaginatorとして使うような場合なら、これでいいだろう。ということで楽をしました。この過程でこの記事が生まれました。 ↩
最初はsimplejsonの使用を勧めていましたが、追加のライブラリを入れなくても、json.dumpsにdefaultの変換用関数を与えればいけます。「この記事」にありました。 ↩
ひとつのパーティションをひとりで使用して、多くもないレコードを時々追加と削除するようなアプリケーションを作る過程で出て来たアイディアなので、ここで作っている内容は大体オーバースペックです。フィルタも使うつもりはなかったのですが、やってみたくなったので書いています。 ↩
記事中で何度も違う意味で「インデックス」という言葉を使ってしまったので、LSI/GSIのことをインデックスと呼ばないことで、無理矢理混同を避けています。 ↩
RDBMSを使う場合でも、limitとoffset等を使って検索するケースが多いでしょうから、間のレコードの増減があるとやっぱり正しくない結果が返るので、仕方ないかもしれません。 ↩
ここなんですよねぇ…代わりにRDSを使えと言われても、仕事ならとにかく、個人で遊ぶにはコストも重要なので…無理矢理にDynamoDBでやりました。 ↩