前提

以下の前提とします。

DynamoDBのハッシュキーにID（文字列）、レンジキーに時刻（文字列）を指定
データは時系列に格納されているものとする。
指定したカラムの集計を取るようにする。
とりあえず指定したIDで全件取得（時間による絞り込みはしない）

こんな感じのデータを想定。

id	score	timestamp
sensor1	0	2017-04-30T22:00:00.000
sensor1	0	2017-04-30T22:00:01.000
sensor1	1	2017-04-30T22:00:02.000
sensor1	0	2017-04-30T22:00:03.000
sensor2	1	2017-04-30T22:00:04.000

このうちsensor1の最新値を取るための実装をしてみます。結果としてscore:0が取得できるロジックを期待します。

実装

実行環境にはSERVERLESS FRAMEWORKを利用しました。こんな感じの実装でできました。

import boto3
import json
import decimal
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource('dynamodb')
table    = dynamodb.Table('test')

def run(event, context):

    res = table.query(
            KeyConditionExpression=Key('id').eq("sensor1")
        )

    return_response = max(res["Items"], key=(lambda x:x["timestamp"]))

    return json.dumps(return_response, default=decimal_default)

def decimal_default(obj):
    if isinstance(obj, decimal.Decimal):
        return float(obj)
    raise TypeError

まずは最新値を求めるというところで手抜き感満載の実装ですが、Aggregate部分を別クラスにしてしまって少しインテリジェンスに選択できるようにすれば、一気に用途が広がる気がします。

実行結果は以下の通りです。無事取得できてますね。

$ sls invoke local -f run
"{\"timestamp\": \"2017-04-30T22:03:00.000\", \"score\": 0.0, \"id\": \"sensor1\"}"

AWS LambdaでDynamoDBから取得した値の最新レコードを取得する

前提

実装