DynamoDBに値が入っている前提で、
- とある時間の
- あるIDにおける
- 集計値
を取ろうする際の実装。（AWS Lambda Python 2.7を使っています）
前提
以下の前提とします。
- 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\"}"