前回 【AWS IoT】AWS IoT Core ルールを使ったデータストアを試してみた（その1）

の続きです。

前提

• AWS IOT CORE 初級ハンズオンを参考にしています。
• 送信元デバイスは　NVIDIA Jetson Nano 開発者キット B01　を利用
• AWS IoT Core のコンパネで「モノ」を作成して、「モノ」用の証明書も作成してデバイスに保存
• デバイスにはAWS IoT Device SDK をインストール
• クライアントプログラムはAWS IOT CORE 初級ハンズオンのダミーデバイス用プログラム( device_main.py）を利用
• デバイスからIoT Coreに向けてMQTTで10秒おきにデータ送信　→ 30秒おきに変更
• データはこんな感じ

{
  "DEVICE_NAME": "jetson-nano-tsumida",
  "TIMESTAMP": "2021-03-17T22:03:48",
  "TEMPERATURE": 18,
  "HUMIDITY": 57
}

2-1.データのストア先と使ったルール（アクション）

※試してみたいものをチョイスしたリスト

今回はこの2 「Amazon Kinesis ストリームにメッセージを送信する」　でLamdaでDyanamoDBに保存するをやっていきます。

2-2-2. 「Amazon Kinesis ストリームにメッセージを送信する」　でLamdaでDyanamoDBに保存する

Dyanamo DBはアプリケーションが参照するためのNoSQL データストアです。
kinesis Data Streams に流すことで大量データのリアルタイム処理が効率的に行えます。

① Amazon Kinesis Data Streamsの作成

まずKinesis streams コンソールでストリームを作成します。

データストリームの作成を押して

データストリーム名に名前をつけます。（今回はjetson-nano-tsumidaと入れました。）
開いているシャード数は1にします。シャード数分コストはかかりますが処理性が上がります。

ステータスがアクティブになれば完了です。

② DynamoDB テーブルの作成

次にDyanamoDBのテーブルを作ります。

テーブルの作成を押します。

DyanamoDB テーブルの作成画面で
テーブル名に名称（今回はjetson-nano-tsumida-yyyymmddと入力）
パーティションキーに　deviceid　と入力、ソートキーの追加にチェックして timestamp を入力
これでテーブルを作ります。

これはすぐできると思います。

③ IAMロールの作成

Lambda用のロールと、AWS IoT 用のロールを作ります。

まずLambda用のロールです。
IAM のロールメニューで「ロールの作成」を押します。

一般的なユースケースのLambdaを選んで次の「ステップ：アクセス権限」を押します。

以下の３つにチェックを入れ「次のステップ：タグ」を押します。

• AWSLambdaBasicExecutionRole
• AmazonKinesisReadOnlyAccess
• AmazonDynamoDBFullAccess

※今回はセキュリティ面で絞るべきところは考慮外とします。

タグは付けずに次に行きます。
ロール名に名前を付け（今回はjetson-nano-tsumida-lambda とつけました）
ポリシーにチェックした3つが入っているかを確認してロールの作成を押します。

次にAWS IoT 用のロールです
同様にIAMのロール画面でIoTを選び、「次のステップ:アクセス権限」を押します。

標準で必要な3つのポリシーがついてるのでそのまま次に行きます。

タグは付けずに次に行きます。
ロール名に名前を付け（今回はjetson-nano-tsumida-iot とつけました）ロールの作成を押します。

④ Lmabdaの設定

次はLambdaを設定していきます。

関数の作成を押す

一から作成を選ぶ
関数名に名前を付ける（今回は jetson-nano-tsumida-yyyymmdd とつけました。）
ランタイムはpython3.8を選択

デフォルトの実行ロールの変更　を押し、既存のロールを使用する　にチェックをして
③ IAMロールの作成で作成したLambda用のロールを選び、関数の作成を押します。

次にトリガーを追加します。「＋トリガーを追加」ボタンを押します。

トリガーの選択でKinesisを選びます。

Kinesis ストリームは、① Amazon Kinesis Data Streamsの作成 で作成したストリームを選びます。

次にコードをタブでコードを書きます。

lambda_function.pyをダブルクリックしてコードを書きます。

※今回のlambdaのコードは、AWS IoT ハンズオン用に用意されているここを参照してください。

Deployを押します。

次にこのpythonのコードが使う環境変数（Dynamoテーブルなど）を設定します。
上タブの設定、左メニューの環境変数を選び編集を押す。

キーに　TABLE_NAME
値に②DynamoDB テーブルの作成で作ったテーブル名を入れ保存する

これで Kinesis → lambda → DynamoDBの部分の設定完了です。
次にIoT Coreのルールとアクションを設定します。

⑤ IoT Core ルール アクションの設定

前回と同様にルールとアクションを設定していきます。
左メニュー「ACT」の下にあるルールをクリックし、画面右上の「作成」をクリックします。

ルールの作成画面が出てくるのでルールに名前を付け、クエリを書きます。

ルール名はjetson_nano_tsumida_kinesisとしました。クエリは以下です。

　select * from 'data/jetson-nano-tsumida'

次に「アクションの追加」をクリックして、
「Amazon Kinesis ストリームにメッセージを送信する」を選択して「アクションの設定」をクリックします。

アクションの設定画面が表示されるので
ストリーム名に　① Amazon Kinesis Data Streamsの作成　で作ったもの
パーティションキーに　${newuuid()} を入力
ロールは③ IAMロールの作成　で作ったAWS IOT 用のロールを選び「ロールの更新」を押します。

アタッチされたポリシーと表示されたらアクションの追加を押します。

元の画面に戻るので「ルールの作成」を押します。

⑥ データを確認する

AWS IOT CORE 初級ハンズオンのダミーデバイス用プログラム(device_main.py）でAWS IoT Coreに向けてデータを送り続けた状態にしています。※前回と変更して30秒おきに送っています。

• Kinesis Data streams に流れたデータ確認

kinesis のコンソールで 作成したデータストリームのストリームメトリクスを見てみます。
レコードを取得 -合計（数）とput レコード - 合計（バイト）が増えていれば、データが流れている　という事になります。

• DynamoDB にたまったデータの確認

Dynamo のコンソールのテーブルで作成したテーブルをクリックし、「項目」タブを選ぶと送信されたデータが確認できます。
ちゃんと届入れいれば、1件ごとに1行表示されます。

データをクリックするとこんな感じで中が見れます。

DyanamoDBへの溜め方は以上です。

⑤ 今回のまとめ（IoT Core →　ルール　→　kinesis Data Streams → Lambda → DyanamoDB)

IoT Core に届いたデータをKinesis Data streams を経由してDynamoDBへ入れてみました。
IoT CoreからDyanamoへは直接データを入れることもできますが、Kinesis Data Streams とLambdaを使うことで
DyanamoDBへストアする際に並列処理をしたりできるので、処理性能を上げることができます。

次回は
timestream とElastic Searchへ入れてみた　を書きます。

追記：次回分リンク。長くなったのでElasticsearchはさらに次回に
【AWS IoT】AWS IoT Core ルールを使ったデータストアを試してみた（その3）