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


IoT　始めたばかり人向けの記事です。
AWS IoT Coreのルール機能で、デバイスから送信したデータをいくつかのストア先に溜めてみました。
ルール機能の概要やストア方法をまとめてみました。


#前提
+ [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秒おきにデータ送信
+ データはこんな感じ

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

#1. AWS IoT Core ルールでできる事

AWS IoT Coreのルール機能を使うと、MQTTで受け取ったデータを他のAWS マネージドサービスへ比較的簡単に連携できます。
##1-1.アクション
連携先は「アクション」としてあらかじめ用意されたものから選択できます。
####アクションで選択できるもののイメージ（2021年3月現在）
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/6ce27711-f9af-bdd8-3cdb-b2e6b485a795.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/a925eccb-0e81-5db8-026d-99d54c535178.png)

##1-2.ルールクエリステートメント

AWS IoT Coreのルール機能では、AWS IoTのメッセージブローカーからデータをサブスクライブする際のフィルターを簡単に書ける機能もあります。「ルールクエリステートメント」がそれにあたります。SQLライクなクエリを書くことでフィルターすることができます。

#### ルールクエリステートメント のイメージ
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/dc9d5832-d428-aec0-8682-bd1d35a74e27.png)
例えば

```SQL
　select * from [topic名] where temperature > 50
```
のように記載することで
[topic名]宛に送信されたもののうち、temperature が 50より大きいものだけを「アクション」で定義した先に送る事が出来ます。


##1-3.エラーアクション

ルール処理中にエラーが発生した場合、エラーアクションの先にデータを送ることも可能です。
通常のアクションと同じ候補から選べますので、何かあったらとりあえずCloud Watchに流しとく
みたいな使い方も可能です。

#### エラーアクションのイメージ
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/e507a097-b27e-ff14-0766-97f10f356c48.png)


#2. AWS IoT Core ルールでやってみた事
[AWS IOT CORE 初級ハンズオン](https://aws-iot-core-for-beginners.workshop.aws/)で試せるものもやってみつつ、他にアクションで試してみたいものをチョイスしてデータを入れてみました。

##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-2.ルール　アクションの設定

###2-2-1. アクション：Amazon Kinesis Firehose ストリームにメッセージを送信する　でS3に保存する
まずS3に保存するパターンです。アクション「S3バケットにメッセージを格納する」で直接書く事もできるにはできるのですが、メッセージをまとめたり圧縮したりしてS3に入れてくれるAmazon Kinesis FirehoseのアクションでS3に入れてみました。

####① S3バケットの作成
まず今回のデータを入れるS3バケットを作ります。
バケット名を入力してリージョン選んだら今回はそのままの設定で作ります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/8b1bde7e-c163-bf66-cf87-74dd89bfd73c.png)


####② Amazon Kinesis Firehose の設定
次にAmazon Kinesis Firehose　を設定します。Kinesis の画面で「配信ストリームの作成」をクリックして
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/8c5645a9-a078-2cfa-8868-c179ee163c40.png)

配信ストリームに名前を付けます。ここではiot-storage-test とつけました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/8eb92004-c6a0-b41c-0301-8bff086e5ab3.png)

Data Transform （あらかじめ作ったlambdaを指定する）やRecord Format conversion（Apache Parquet などの機能へ連携）などETLのTする機能を設定する事もできるようですが今回はスキップします。

Step3で保存先を選択できますのでここでS3を選びます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/76f6d5b6-f437-0f49-74dc-0f1a641d2b9c.png)

次に[① S3バケットの作成](#①-S3バケットの作成)で作ったS3バケットを選んでNEXT画面に行きます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/59b55dc0-ecfa-46aa-1686-489bde26fa7c.png)

次にバッファサイズとバッファインターバルを設定します。このバッファサイズを超える量まで溜め込むか、インターバルの時間を超えると、溜めてたものを1ファイルにまとめてS3に吐きだしてくれます。ここでは5MBと60secondsに設定してみました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/ad420a0c-5cc6-c520-26d5-b950f37d0b87.png)

他、圧縮方式、暗号化の有無、エラーログをCloud Watchに送るか　などを設定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/a72e511f-2a5c-49ba-be9f-7da11f5d0b09.png)

最後にIAMロールを作るか選んであげて完了です。※IAMロールの説明は今回は除きます。

####③ IoT Core ルール アクションの設定
ここまでできたらIoT Core の ルール アクションを設定していきます。
左メニュー「ACT」の下にあるルールをクリックし、画面右上の「作成」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/02182e82-3b5b-e6ac-18ca-9e72d235ce2c.png)

ルールの作成画面が出てくるのでルールに名前を付けます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/489e52c6-6d08-2a0a-13c9-ee2177f662e3.png)

小ネタなんですが、このルール名、-（ハイフン）を受け付けず、_（アンダースコア）はOKなんです。ここまで聞くと普通に聞こえるかもしれませんが、AWSの他のサービスで名前つけようとすると全く逆だったりするんです。なぜここだけハイフンNGでアンスコOKにしたのか問いただしたい。知ってる人がいたら教えてほしいです。

次にルールクエリステートメントを設定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/c08c1390-be6c-308e-abd0-c842c03a0572.png)

黒いコンソールみたいなところにこんなSQLを書いてあげるだけです。フィルターしたい場合は`where TEMPERATURE > 50` のようなWhere句を後ろにつけます。

```SQL
　select * from 'data/jetson-AN0001'
```

今回使ったプログラムはメッセージのTopicを 「data/[デバイス名]」として送っているのでこのようなクエリになっています。

次に「アクションの追加」をクリックして、
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/27c5eb6e-86e6-f63e-4069-38d2391b73af.png)

「Amazon Kinesis Firehose ストリームにメッセージを送信する」を選択して一番下にある「アクションの設定」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/e30da8bf-8a95-abcb-0105-b65c8b936fde.png)

「Amazon Kinesis Firehose ストリームにメッセージを送信する」の設定画面がでてくるので、[② Amazon Kinesis Firehose の設定](#②-Amazon-Kinesis-Firehose-の設定)で設定した名前をストリーム名で選択します。Kinesis Firehoseはデータをバッファしてまとめてくれますが、データを何で区切るかをセパレータで指定します。今回は「\n (改行)」を選びました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/5a1d84ee-b0dc-f21a-12ba-2b61eea7f393.png)

「バッチモード」はリクエスト内に複数のデータを埋め込んで送るPutRecordBatch用のオプションなので今回はチェックしません。最後に適切なIAMロールを選ぶか、ここで作成して設定完了です。


####④ データを送って確認してみる。
[AWS IOT CORE 初級ハンズオン](https://aws-iot-core-for-beginners.workshop.aws/)のダミーデバイス用プログラム(device_main.py）でAWS IoT Coreに向けてデータを送ってみます。

こんな感じで実行されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/8e28d4ed-057f-3526-8d5e-852a5545da2e.png)


実際にメッセージがIoT Coreに届いているかはIoT Coreのメニュー「テスト」で簡単に確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/247a09d4-a30f-530b-05c9-7c452d754c4f.png)
この画面で「トピックのフィルター」に送信したトピック名を入力（上画像の例ではdata/jetson-AN0001）して、サブスクライブボタンを押すだけです。
メッセージがそのトピック宛に届いていれば、画面下部にリアルタイムに表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/60a45bd0-b1f2-d3e0-72d6-97d87551bb01.png)
※編集の都合上トピック名を修正しています。

ここでもう一つ小ネタです。一度作った「モノの名前」は、名前の編集といった形では修正できません。もし変えたい場合、新しい名前でモノを作って置き換える必要があります。

https://docs.aws.amazon.com/ja_jp/iot/latest/developerguide/iot-moisture-create-thing.html
>作成後にモノの名前は変更できません。モノの名前を変更するには、新しいモノを作成して、新しい名前を付け、古いモノを削除する必要があります。

実際に運用する場合、モノの命名ルールは最初にちゃんと決めたほうが良さそうですね。

では、最終目的S3にデータが入ったか見てみます。自動で年／月／日／時　でフォルダが切られています。

+ 年
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/32664963-f91b-80bd-e5f8-8aa1f6ae1e5c.png)

+ 月
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/41fb9c7e-660f-6f0a-8e59-b3b84442fefe.png)

+ 日
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/2c393270-0a2a-d722-eb79-21b7d14a045c.png)

+ 時間
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/a7c0bd28-5c97-11c4-1713-320cd890dc88.png)

+ ファイル
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/df9b6fa0-610b-b4bb-2f03-c01da7df638f.png)


ファイルをダウンロードして中を見ると、送信したデータがまとめられて保存されているのが分かりました。

```
{"DEVICE_NAME": "jetson-sumida-AN0001", "TIMESTAMP": "2021-03-18T23:22:55", "TEMPERATURE": 24, "HUMIDITY": 52}
{"DEVICE_NAME": "jetson-sumida-AN0001", "TIMESTAMP": "2021-03-18T23:23:25", "TEMPERATURE": 18, "HUMIDITY": 57}
{"DEVICE_NAME": "jetson-sumida-AN0001", "TIMESTAMP": "2021-03-18T23:23:55", "TEMPERATURE": 15, "HUMIDITY": 58}
```

S3への送信確認はこれで終わりです。

####⑤ 今回のまとめ（IoT Core →　ルール　→　kinesisFirehose → S3）
ルールのアクションとサービスを設定するだけで簡単にS3に入れる事ができました。
クライアントプログラムを除けば、コードはとても簡単なクエリのみです。
ルールのアクション機能を活用すると簡単にサービス連携ができそうです。


記事が長くなったので
アクション「Amazon Kinesis ストリームにメッセージを送信する」でDynamoDBに保存する
を次回書きます。

追記：次回分リンク
[【AWS IoT】AWS IoT Core ルールを使ったデータストアを試してみた（その2）](https://qiita.com/tsumida/items/dffb9dae72a862909f0d)