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

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


過去分
 [【AWS IoT】AWS IoT Core ルールを使ったデータストアを試してみた（その2）](https://qiita.com/tsumida/items/dffb9dae72a862909f0d)
 [【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で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を経由|


今回はこの4 「Amazon Elasticsearch Serviceにメッセージを送信する」をやっていきます。

###2-2-4.  「Amazon Elasticsearch Serviceにメッセージを送信する」
Amazon Elasticsearch ServiceはElasticsearchのマネージドサービスです。
Elasticsearchはテキストデータを検索、分析できるようにインデックス化して格納してくれるNoSQLデータストアです。
Apache lucene ベースで、Apache solrとはよく比較されますが、RESTFUL APIで作られているとう点やノンスキーマで入れられる　ようなことが特徴でしょうか。
可視化ツールkibanaとの親和性が高くAmazon Elasticsearch Serviceを使えば、kibanaも使えます。


####① Amazon Elasticsearch Service インスタンスの作成

※【補足】Elastiscserch Service インスタンス は停止して課金を止める　ようなことができず、課金を止めるなら消すしかないので注意です。

Amazon Elasticsearch Serviceのコンソールにアクセスし、新しいドメインの作成を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/9258f8d7-e270-2a2b-954c-6262f823a967.png)

 + デプロイタイプは「開発及びテスト」
 + バージョンは最新の「7.10(latest)」

を選び次へを押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/40880023-5352-5e84-255d-3701d425e2eb.png)


 + Elasticsearch ドメイン名にドメイン名を付けます。（今回は`jetson-nano-tsumida-yyyymmdd`）と入れました。
 + カスタムエンドポイントと自動調整はそのままにします。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/bb2c3452-c31b-51bb-e6d2-7d5421b56641.png)

 + データノードは`t3.small.elasticsearch`か`t2.small.elasticsearch`を選びます。※選ぶと下に無料枠の説明が表示されます。
 + データノードストレージはデフォルトのままとします。EBSの汎用SSDで10GB 

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/1580b8a5-1ecd-aeb0-be84-89c9046377f1.png)

 + 専用マスターノードは作らず、そのまま次に行きます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/f0a5a45e-11ad-5279-7590-fbf2885d2ddb.png)

 + ネットワークセキュリティは「パブリックアクセス」を選択します。

　　※後述する設定で接続できるIP制限はします。

 + 細かいアクセスコントロールを有効化　のチェックを外します。
 
　　※お試し用途であるため。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/a3d65e5c-8b50-3c36-8b57-c73905a5a1e0.png)

 + Kibana の SAML 認証とAmazon Cognito 認証はそのままにします。（使わない）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/94cbdd96-394f-9cef-ce66-4c290ff391e8.png)

 + ドメインアクセスポリシーは「カスタムアクセスポリシー」を選びます
 + http://checkip.amazonaws.com/ にアクセスし、自身の接続元IPをメモします。
 + IPv4アドレス を選び、上記のIPアドレスを入力、許可を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/457b3dea-b43a-cf9f-b0b3-0503163dc10d.png)

 + 暗号化、任意のクラスター設定はそのままにします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/c90e387e-e619-4a8d-c418-df0868dc81fe.png)

 + タグの追加もせずそのまま次にいきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/aa9340a8-ad6c-9cc8-4214-30b96f16e66e.png)

 + 確認画面で設定値を一通り確認して、確認を押します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/57880ab7-b120-ea0b-04aa-9479f8abc7c9.png)


これでインスタンスの作成手順は完了です。インスタンスができるまで結構時間がかかります。
※10分と出ますが20分くらいはかかります。

ドメインのステータスが「アクティブ」になれば完了です。作成完了画面で、KiabanaのURLはメモしておいてください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/d7245d53-5ce6-f97f-7f3e-a11a5d369603.png)




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

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

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

```SQL
　select parse_time("yyyy-MM-dd'T'HH:mm:ss.SSSZ", timestamp()) as timestamp, * from 'data/jetson-nano-tsumida'
```

※IoT Coreで受け取った時間をペイロードの先頭につけています。parse_timeはタイムスタンプを人間が判読可能な日付/時刻形式にフォーマットします。（このフォーマットはタイムゾーンを表すUTC+0000も付与されます。）

elastichserchは時間データをタイムゾーン情報なしに送ると、UTCとして受け取ります。
例えば今回のデバイスからの送信データ`"TIMESTAMP": "2021-06-30T15:00:00"` は
"2021-06-30T15:00:00+0000" として入ります。
これをasia/tokyo TIMEZONEのクライアントで見ると9時間進んで見える（7/1 の0時になる）ので注意です。

今回はあえて小文字の`timestamp`として（送信データそのままの）`TIMESTAMP`と比較できるようにしています。


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


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/2145d0ad-7a08-f0a7-2535-5265f27901f5.png)




 + ドメイン名は[① Amazon Elasticsearch Service インスタンスの作成](#①-amazon-elasticsearch-service-インスタンスの作成) で入力したElasticsearch ドメイン名 を選択してください。
 + IDには`${newuuid()}` を入力します。
 + 索引には`timestamp`を入力します。
 + タイプにも`timestamp`を入力します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/fc2e19e4-97bd-33bf-6e0e-1b5b019ca512.png)


 + ロールの作成 を押しロール名を付けます。（今回は`jetson_nano_tsumida_es`と入れました。）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/dc5ea8b2-e95f-f989-4ea4-82d3829873b4.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/3473df61-5a89-0d50-382d-3647ed19c44a.png)

 + ルール画面に戻るのでルールの作成を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/024e004a-b104-18f2-a394-57b6e86e1164.png)

これでルールアクションの設定は完了です。



####③ kibanaでデータを確認する
Elasticsarchに保存されたデータをKibanaで確認していきます。

[① Amazon Elasticsearch Service インスタンスの作成](#① Amazon Elasticsearch Service インスタンスの作成)で作成後に表示されたKibanaのURLにアクセスします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/bdd3322c-90a6-57c4-983e-52269d933d01.png)

真ん中をKibanaをクリックするとこの画面に飛びます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/3088f8ac-f540-08b4-679e-439e8ede8a61.png)


 + 三 をクリックして「Discover」を押します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/11e6e329-96c4-66b6-d80d-87788fd7062f.png)


 + 「Crate index pattern」をクリックします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/b47aec6c-f602-41ff-57b3-a387c153de32.png)


+ index pattern name に `timestamp*` と入力して「Next step」を押します。(*は自動で入ります）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/060b690f-c62a-e0c7-6027-9a003d3b611d.png)

※現在Elasticsearchに入っているデータ（index）一覧が表示されるので、Indexを選んでkibanaに登録する　といった流れです。indexは　timestamp-yyyymmdd のように日別で分割して作ったりしますので、timestamp-* というインデックス名のパターンで登録できます。

+ time field は`timestamp` （タイムゾーン付で送ったデータ）を選択し、「Create index pattern」を押します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/97297952-ff15-f16b-9543-429f81e2fdfd.png)


 + index patternが登録されると以下のように表示されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/04ffd32f-d5dd-9d3d-9a80-40378567812c.png)


 + もう一度、三 をクリックして「Discover」を押します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/959191e7-1be0-887a-0de6-9f0f815c3979.png)

このようにデータが届いていることが確認できました。

グラフの下に表示されているのは、実際に届いたデータ１件１件が表示されています。>を押して開いてみます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/428bd5a3-be78-674a-7dd0-77c2cf1a93cd.png)

ルールクエリのところで書きましたが、大文字TIMESTAMPはタイムゾーンなしで送信したもの、小文字timestampはtimezone付きで送信したものです。

TIMESTAMP:Jul 1, 2021 @ 02:19:47.000
timestamp:Jun 30, 2021 @ 17:19:47.320

実際に送信した時間は17:19:47です。大文字TIMESTAMPは9時間進んでいる（誤った時間になっている）ことがわかります。エッジからの送信時にはタイムゾーンを付けて送るのが良いと思います。（UTCに変換して、2021-06-30T08:19:47Z として送るなど）

####④ kibanaでデータを可視化する

次はKibanaでグラフを作ってみます。

 + 三 をクリックして「Visualize」を押します。

 + 「Create new visualization」を押します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/c5cc81f2-81fc-6a72-6443-79fc640c417b.png)

 + 「Line」を選びます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/1160efe3-af75-4bc4-d060-8bd21627ad69.png)

 + 「timestamp*」を選びます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/c86870b0-8d86-5116-10f3-9b23da8975be.png)

+ グラフ画面が表示されすので、右の Y-axisを設定していきます。

Aggregationに `Avarage`
Fieldに `TEMPERATURE`

を入れます。送信された温度データの時間単位の平均値をだすという意味です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/02247b94-9bf2-7eba-a71a-0bde67b6a70f.png)

+ 「+Add」押してY-axisを選び以下を追加します。

Aggregationに `Avarage`
Fieldに `HUMIDITY`

を入れます。送信された湿度データの時間単位の平均値をだすという意味です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/0773450c-8ac2-570d-b22a-dac3d3bae11b.png)


+ Bucket下の「+Add」押してX-axisを選び以下を追加します。

Aggregationに `Date Histogram`
Fieldに `timestamp`
minimum intervalに `Minute`

を入れます。横軸を時間にして10分単位で集計する　という意味です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/c5644f28-2146-39a0-10a1-1c0e6a8665e0.png)

 + 「Update」を押すとグラフが表示されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1001712/e64ff7dc-dc9e-8bdc-caf6-796249c4c56c.png)

Kibanaによる可視化はここまでです。


#####小ネタ
elasticsearch に入ったIndexを確認したり、削除したりするにはElasticsearchのapiを呼びます。

PCのターミナルからcurlを実行します。※curlは必要に応じて入れてください。

 + Index一覧を取得する

```
curl -XGET 'https://[elasticsearchのエンドポイント]/_cat/indices?v'
```

※elasticsearchのエンドポイントはelasticsearchのコンソールで確認できます。
 + 結果

```
$ curl -XGET 'https://[elasticsearchのエンドポイント]/_cat/indices?v'
health status index     uuid                   pri rep docs.count docs.deleted store.size pri.store.size
green  open   .kibana_1 CAFIybgnSf6EC9QG0hW8mQ   1   0         26            5     27.7kb         27.7kb
yellow open   timestamp 0wFtu4F6RNGYiRcPS61OVg   5   1        118            0    194.6kb        194.6kb

```


sampleで入れたデータなどを削除する場合は削除リクエストを投げることで消せます。

 + Indexを削除する


```
curl -XDELETE 'https://[elasticsearchのエンドポイント]/timestamp?pretty'
```

+ 結果

```
$ curl  -XDELETE 'https://[elasticsearchのエンドポイント]/timestamp?pretty'
{
  "acknowledged" : true
}
```




####⑤ 今回のまとめ（IoT Core →　ルール　→　Elasticsearch )

IoT Core に届いたデータを直接Elasticsearchに入れてみました。これも設定自体はとても簡単でした。すぐにできます。
Elasticsearchは文字列データの検索が強いので、そういった要件があれば検討してみてもよいと思います。kibanaで可視化も簡単にできるので便利です。
ただ他のデータストアよりコストはかかるので注意です。（データ量だけでなく、EC2インスタンスを常時起動している分の料金がかかる）

今回はここまでにします。次回はラスト 5「IoT Analyticsにメッセージを送る」を投稿します。

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