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

前回 [【AWS IoT】AWS IoT Core ルールを使ったデータストアを試してみた（その1）](https://qiita.com/tsumida/items/c03cd6eef3341c7915e9)

の続きです。

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

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

##2-1.データのストア先と使ったルール（アクション）
※試してみたいものをチョイスしたリスト

|#  | ストア先           |使ったルール（アクション）                               | 途中経由したサービス |
|:-:|:-                 |:-                                                      |:-                   |
|1  | S3                | Amazon Kinesis Firehose ストリームにメッセージを送信する|Amazon Kinesis Firehose |
|2  | DynamoDB          | Amazon Kinesis ストリームにメッセージを送信する         |Amazon Kinesis Data Streams,Lambda |
|3  | timestream        | Timestreamテーブルにメッセージを書き込む                |なし|
|4  | Elasticsearch     | Amazon Elasticsearch Serviceにメッセージを送信する|なし|
|5  | AWS IoT Analytics | IoT Analyticsにメッセージを送る|IoT Analytics内でchannel,pipeline,storageを経由|


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


###2-2-2.  「Amazon Kinesis ストリームにメッセージを送信する」　でLamdaでDyanamoDBに保存する
Dyanamo DBはアプリケーションが参照するためのNoSQL データストアです。
kinesis Data Streams に流すことで大量データのリアルタイム処理が効率的に行えます。


####① Amazon Kinesis Data Streamsの作成
まずKinesis streams コンソールでストリームを作成します。

データストリームの作成を押して
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/f68c8c16-24ed-938d-dd79-b1de383b9bc5.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/6b975015-7460-58a9-69e6-4a254dc7fe83.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/bb0470b2-9ce0-70a8-7428-4a56a6289419.png)


####② DynamoDB テーブルの作成
次にDyanamoDBのテーブルを作ります。

テーブルの作成を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/d7cf9238-524b-a0cf-7922-4900c5867af8.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/a5e857c8-44f4-bf72-bbc1-ce2bcaa01457.png)

これはすぐできると思います。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/d9fbdd81-e23f-d3c7-e05b-b6e1d144006f.png)


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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/aacf1206-1549-01ca-9dbb-d5e9de222c69.png)

一般的なユースケースのLambdaを選んで次の「ステップ：アクセス権限」を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/cbc9b972-c1a9-d857-d06f-57abe3f56578.png)

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

 + AWSLambdaBasicExecutionRole
 + AmazonKinesisReadOnlyAccess
 + AmazonDynamoDBFullAccess

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/cecfe6cc-b173-4bd6-f33b-354d0eaff289.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/a954596b-120e-a596-0087-0487eda36a55.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/89b2114b-2331-6d21-8331-bf0b92643a16.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/c39f9dfe-59a9-7a88-e039-9be638af4c06.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/897d348a-f705-e88f-d8d3-674d0f28e4f9.png)

標準で必要な3つのポリシーがついてるのでそのまま次に行きます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/69f8384c-e4ca-87c6-5edf-40e34980d2fc.png)

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


####④ Lmabdaの設定
次はLambdaを設定していきます。

関数の作成を押す
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/bf2e5b27-636a-7b4f-2c6d-b7d2c7edee7b.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/307ce460-8823-8194-f1c9-28ec92265b92.png)

デフォルトの実行ロールの変更　を押し、既存のロールを使用する　にチェックをして
[③ IAMロールの作成](#③-IAMロールの作成)で作成したLambda用のロールを選び、関数の作成を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/44609ab3-5bb1-331a-49df-e2f3ec8f2012.png)


次にトリガーを追加します。「＋トリガーを追加」ボタンを押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/5a2754c1-0949-7704-b359-28e86dbe2835.png)


トリガーの選択でKinesisを選びます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/675f2660-b846-4501-cd22-55884e0bf45c.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/1f4941cc-0b44-7e11-2666-a1818be563c0.png)


次にコードをタブでコードを書きます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/698e97c0-0172-c3fd-fb18-50922871d771.png)

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

※今回のlambdaのコードは、AWS IoT ハンズオン用に用意されている[ここ](https://awsj-iot-handson.s3-ap-northeast-1.amazonaws.com/aws-iot-core-workshop/lambda_function.py)を参照してください。

Deployを押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/b262979f-27e5-13c8-549e-516e6ab93624.png)

次にこのpythonのコードが使う環境変数（Dynamoテーブルなど）を設定します。
上タブの設定、左メニューの環境変数を選び編集を押す。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/3b299ed3-756d-83ae-943c-29baeec69bdd.png)

キーに　TABLE_NAME
値に②DynamoDB テーブルの作成で作ったテーブル名を入れ保存する
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/e2483840-5e74-4c15-dee6-6486a5cec05f.png)

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


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

ルールの作成画面が出てくるのでルールに名前を付け、クエリを書きます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/1ac3e9f9-1e43-d65a-c811-4e38eadadd06.png)


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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/db8a235f-50cf-d6aa-4c4b-0b71c72cf891.png)


アクションの設定画面が表示されるので
ストリーム名に　① Amazon Kinesis Data Streamsの作成　で作ったもの
パーティションキーに　`${newuuid()}` を入力
ロールは③ IAMロールの作成　で作ったAWS IOT 用のロールを選び「ロールの更新」を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/ee5cd673-1583-708f-5bcc-345f2dcd36b5.png)

アタッチされたポリシーと表示されたらアクションの追加を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/c23e5b1a-5eba-681c-a571-1c533b0cae11.png)

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



####⑥ データを確認する
[AWS IOT CORE 初級ハンズオン](https://aws-iot-core-for-beginners.workshop.aws/)のダミーデバイス用プログラム(device_main.py）でAWS IoT Coreに向けてデータを送り続けた状態にしています。※前回と変更して30秒おきに送っています。


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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/8d1dd3a7-5a9a-228c-9e4f-a2a1eb36abe4.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/4433dffb-9392-cd2f-fd7f-6787332d7d24.png)

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

 Dynamo のコンソールのテーブルで作成したテーブルをクリックし、「項目」タブを選ぶと送信されたデータが確認できます。
ちゃんと届入れいれば、1件ごとに1行表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/43d25df2-10d2-7384-5d1d-3b76bfc1d62f.png)

データをクリックするとこんな感じで中が見れます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/4c0d31e3-396d-9259-3d69-496397a9855e.png)



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）](https://qiita.com/tsumida/items/e5d628cb2a34c4f0f6b5)