【AWS Lambdaの基本コードその5】 DynamoDBにデータを登録

投稿日 2017年9月28日
著者 aws-recipe-user
カテゴリーデプロイメント＆マネージメント

こんにちは、Michaelです。
今回は、AWS LambdaからDynamoDBのテーブルにデータを登録(Put)してみます。

今回の構成

Lambdaが起動されると、外部から入力されたデータをDynamoDBのテーブル「invoke_history」に登録します。Lambdaで実行された時間を記録し、ソートキー「time」に登録することでキーの重複を回避します。

IAMロールの設定

LambdaからDynamoDBにデータを登録するため、以下のポリシーを作成し、Lambdaに適用するIAMロールにアタッチします。このポリシーでは、テーブルの「読み取り」、「書き込み」、「更新」の最小限のポリシーのみ許可しています。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "dynamodb:GetItem",
                "dynamodb:PutItem",
                "dynamodb:UpdateItem",
                "dynamodb:DescribeTable"
            ],
            "Effect": "Allow",
            "Resource": "*"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Action": [

"dynamodb:GetItem",

"dynamodb:PutItem",

"dynamodb:UpdateItem",

"dynamodb:DescribeTable"

"Effect": "Allow",

"Resource": "*"

}

]

}

コード

コードは以下のようになります。

# coding: utf-8

# ①ライブラリのimport
import datetime
import decimal
import boto3
from boto3.dynamodb.conditions import Key, Attr

# ②Functionのロードをログに出力
print('Loading function')

# ③DynamoDBオブジェクトを取得
dynamodb = boto3.resource('dynamodb')

# ④Lambdaのメイン関数
def lambda_handler(event, context):
    
    # ⑤テーブル名を指定
    table_name = "invoke_history"
    
    # ⑥実行時間の記録
    time_utc = datetime.datetime.today()
    time_unix = time_utc.timestamp()
    event["time"] = time_unix

    # ⑦float型をdecimal型に変換
    payload = {key: val if type(val) != float else decimal.Decimal(str(val)) for key, val in event.items()}
    
    # ⑧DynamoDBテーブルのオブジェクトを取得
    dynamotable = dynamodb.Table(table_name)

    try:
        # ⑨DynamoDBへのデータ登録
        res = dynamotable.put_item(
            Item = payload
        )
        print("Succeeded.")
        return
    
    except Exception as e:
        print("Failed.")
        print(e)
        return

# coding: utf-8

# ①ライブラリのimport

import datetime

import decimal

import boto3

from boto3.dynamodb.conditions import Key, Attr

# ②Functionのロードをログに出力

print('Loading function')

# ③DynamoDBオブジェクトを取得

dynamodb = boto3.resource('dynamodb')

# ④Lambdaのメイン関数

def lambda_handler(event, context):

# ⑤テーブル名を指定

table_name = "invoke_history"

# ⑥実行時間の記録

time_utc = datetime.datetime.today()

time_unix = time_utc.timestamp()

event["time"] = time_unix

# ⑦float型をdecimal型に変換

payload = {key: val if type(val) != float else decimal.Decimal(str(val)) for key, val in event.items()}

# ⑧DynamoDBテーブルのオブジェクトを取得

dynamotable = dynamodb.Table(table_name)

try:

# ⑨DynamoDBへのデータ登録

res = dynamotable.put_item(

Item = payload

)

print("Succeeded.")

return

except Exception as e:

print("Failed.")

print(e)

return

コードの説明

コードの内容は以下の通りです。

項番	項目	内容
①	ライブラリのimport	処理に必要なライブラリを指定してインポートします。
②	Functionのロードをログに出力	Lambdaのコードがロードされたことが分かるようにログにメッセージを出力します。
③	DynamoDBオブジェクトを取得	DynamoDBと接続できるようにインスタンスを生成します。
④	Lambdaのメイン関数	トリガーにより実行される関数。
⑤	テーブル名を指定	データを登録するテーブル名を設定します。
⑥	実行時刻の記録	Lambdaの実行時刻をUnixtimeで記録します。
⑦	float型をdecimal型に変換	DynamoDBではfloat型がサポートされないため、decimal型に変換します。
⑧	DynamoDBテーブルのオブジェクトを取得	DynamoDBのテーブルを操作できるようにインスタンスを生成します。
⑨	DynamoDBへのデータ登録	DynamoDBのテーブルにデータを登録します。

実行結果

テスト実行にあたり、テストイベントとして以下のJSONデータを登録しておきます。

{
  "client_id": "device001",
  "message": "Lambda test"
}

{

"client_id": "device001",

"message": "Lambda test"

}

Lambdaのコードエディタの上部にある[テスト]をクリックすると、テスト実行されます。
テストを実行すると、テストイベントのデータとLambda実行時刻がDynamoDBに登録されていることが確認できました。

まとめ

LambdaとMySQLといったRDBMSとの組み合わせは、コネクション数等の問題からアンチパターンとされるため、DynamoDBとの組み合わせはよく使われます。
DynamoDBはスキーマレスでRDSより設計がしやすく、ディスク容量の制限もないため、IoTのような無尽蔵にデータが入力されるケースでもディスク容量を気にせず利用できます。
次回は、DynamoDBのデータをLambdaから読み取るコードを紹介します。
次回もお楽しみに！

この記事を書いた人

aws-recipe-user

記事一覧