AWSとBLE人感センサーで会議室の利用状況を可視化してみる[第3話]

# 概要
エヌシーアイ総合システムのtaguchyです。
普段は主にBtoCのWEBシステムや、BtoB向けWEB APIの開発などをしています。
最近はIoT関連でもお仕事させていただいております。

さて、今回は第三話ということで、今日は以下を行います。
(前回は[こちら](https://qiita.com/taguchy/items/8bd153b91105cc11a04e "AWSとBLE人感センサーで会議室の利用状況を可視化してみる[第2話]"))

1. センサーデータをJSONに変換するLambdaを作成
2. Kinesis Firehoseを利用してS3, ElasticSearchに格納
3. Kibanaにてセンサーデータを参照

## 利用デバイス
センサーはレンジャーシステムズ株式会社様の「monoコネクト」を利用して、以下のデバイスを使用します。

* BLE/WiFiゲートウェイ：１台
* 人感センサー：３台

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/3e1b38cd-b664-653b-45e1-605499711b99.jpeg" width=60%>


## AWS構成
Amazon Kinesis Data Firehoseを利用して、AWS IoTからS3とElasticsearchにデータを流します。

![AWS構成.png](https://qiita-image-store.s3.amazonaws.com/0/197194/4e1d7788-95e2-86f2-449b-f4d04224b59d.png)

# センサーデータをJSONに変換するLambdaを作成
AWS Cloud9を使ってLambdaを作成します。
nodeを書くのは初めてですが頑張ります。

![Cloud9_1.png](https://qiita-image-store.s3.amazonaws.com/0/197194/2a1e31f9-aacf-8e16-4d94-32f9f1d01f2f.png)

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/cea5b0cd-d91e-3000-387f-9e5520fa55ee.png" width="683px">

csvtojsonやasyncなどのモジュールを利用したいので、画面中央下のコンソールにてnpmコマンドを実行しました。

```
yuki-taguchi:~/environment $ cd convertSensorDataToJSON/
yuki-taguchi:~/environment/convertSensorDataToJSON $ npm init
・
・
yuki-taguchi:~/environment/convertSensorDataToJSON $ npm install csvtojson
・
・
yuki-taguchi:~/environment/convertSensorDataToJSON $ npm install async
・
・
yuki-taguchi:~/environment/convertSensorDataToJSON $ npm install date-utils
・
・
yuki-taguchi:~/environment/convertSensorDataToJSON $ ls
convertSensorDataToJSON  node_modules  package.json  template.yaml
yuki-taguchi:~/environment/convertSensorDataToJSON $ ls node_modules/
async  csvtojson  date-utils  first-chunk-stream  is-utf8  lodash  strip-bom
```

変換用に以下のLambda関数を作成しました。
Kibanaで時系列データとして扱うために"@timestamp"フィールドを作成しています。

```
'use strict';
console.log('Loading function');
const csv = require('csvtojson');
const asyn = require('async');
require('date-utils');

exports.handler = (event, context, callback) => {
        // all record convert.
        asyn.map (event.records, function (record, callback) {
            // need to base64 decode.
            let sensor_data = Buffer.from(record.data, 'base64').toString('utf8');
            // "$GPRP" is not need.
            var sensor_data_non_gprp = sensor_data.replace(/^(\$GPRP,)/g, "");
            // split payload.
            let payload = sensor_data_non_gprp.match(/^(.{12},.{12},.*,)(.*)/);
            let split = payload[2].match(/^(02010612FF0D0082BC)(.{4})(.{2})(.{8})(.{4})(.{2})(.{6})/); 
            let fix = split[1];
            let battery = split[2];
            let status = split[3];
            let reserve = split[4];
            let user = split[5];
            let fix2 = split[6];
            let reserve2 = split[7];
            
            // make @timestamp
            let now = new Date();
            
            let sensor_data_split_payload = payload[1]
                                + fix + "," + battery + ","
                                + status + "," + reserve + ","
                                + user + "," + fix2+ ","
                                + reserve2 + "," + now.toFormat('YYYY-MM-DDTHH24:MI:SS');
            csv({
                noheader: true,
                headers: ['sensor_mac','gw_mac', 'rssi', 'fix', 'battery', 'status', 'reserve', 'user', 'fix2', 'reserve2', '@timestamp']
            }).fromString(sensor_data_split_payload)
                .on('json',(jsonObj)=>{
                // Called for the number of rows.
                console.log(jsonObj);
                let jsonData = Buffer.from(JSON.stringify(jsonObj), 'utf8').toString('base64');
                callback (null, {
                                recordId : record.recordId,
                                result : 'Ok',
                                data : jsonData
                                }
                );
            });
        }, function (err, results) {
            if (err) {
                throw err;
            }
            console.log(results);
            callback (null, {records: results});
        });
};
```

ソースの修正が完了したら、Cloud9右側の"AWS Resources"にて"↑"をクリックし、デプロイします。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/0e615545-58a0-f39d-bf44-47b4e4e2f0c7.png" width="683px">

最後にLambdaのロール設定を行います。
LambdaのAWSコンソールからLambdaの実行ロールを調べます。
※Lambdaのタイムアウト設定を3分に伸ばしました。
　（Kinesisの設定画面で注意文言が出る為ですが、実際は必要ないかもしれません。）

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/46c91e72-2ded-3fa3-da44-d35aee208c5c.png" width="683px">

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/13599ec0-df04-8c31-8ea8-906d0af5ebcf.png" width="683px">

KinesisFirehoseから呼び出すため、KinesisFirehoseへのフルアクセスポリシーを設定します。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/f610c295-7ab3-7523-fafd-84b450cf4253.png" width="683px">

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/4f20813f-db00-7a52-d9cf-b47fbc37677c.png" width="683px">

以上でセンサーデータをJSONに変換するLambdaが作成できました。

# Kinesis Firehoseを利用してS3, ElasticSearchに格納する
## S3バケットの作成
今回データの可視化はElasticSearchで行うため、S3はバックアップ用途として使用します。
機会があればS3に入ったデータをAthenaなどで見ることもしてみたいです。

バケットを作成します。
![AWS_S3.png](https://qiita-image-store.s3.amazonaws.com/0/197194/55b47d3e-759f-e2f8-d2bd-34ae5dfa1bd1.png)

## ElasticSearchの設定
今回はデバイスデータを以下のルールで保存します。
1. 24時間毎に保存
2. 保存したデータは1か月保持

１日のデータ量がどの程度になるのかが把握しきれなかっ
ため、まずはデフォルト設定でElasticSearchを立ち上げます。
ElasticSearchのバージョン6.0は、現時点(2018/1/5)ではKinesisFirehoseが対応していないようだったので、5.5に変更しました。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/dfbb9b83-c367-b630-f05b-2be22d70a6a5.png" width="683px">

スナップショットの時間はAM1:00としています。
<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/f1099e5b-e176-22da-5aaa-2e93af15562e.png" width="683px">
<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/f3aef8c7-c4db-7ae1-6426-07528b49770e.png" width="683px">

ElasticSearchはVPC内に作成することもできますが、ALBの設定などに手間取ることを避け、今回は社内IPのみに限定したネットワーク公開としました。

## Kinesis Firehoseの設定
FirehoseにてLambdaを利用してPayloadデータをJSON変換しつつ、S3およびElasticSearchに転送する設定を行います。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/d07a1dc5-6dd7-1cde-0c64-a5216fdaaafb.png" width="683px">

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/a192b0af-a3c2-74f9-4463-ea58d3904a09.png" width="683px">

データ変換設定にて、事前に作成したLambdaを指定します。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/3c6d5ad9-1735-da1f-b309-832c5033af50.png" width="683px">
<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/3fd59b89-a57e-d17f-e0c8-d8d1cfbb6096.png" width="683px">

Firehoseにてデータのバッファを設定します。
今回は60秒置きにバッファしたデータをLambdaにて変換し、S3およびElasticSearchに流し込む設定としました。
※もっと小さい数字にしようかと思いましたが60秒が一番小さい数字でした。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/fb412560-8fa6-398f-43a1-9c523794db0e.png" width="683px">

Firehoseのロールはボタンポチポチで勝手に作成されました。とても楽です。
ElasticSearchへの設定もS3とほぼ同様に行います。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/79512485-2085-dfc1-3c65-56199e9bff30.png" width="683px">

作成完了です。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/afbaebe1-7a8f-61b0-c1e6-edbc0efa23bc.png" width="683px">


## AWSIoTのルール設定
KinesisFirehoseを利用してデータを転送するルール設定を行います。
AWSIoTのACTを選択してルール設定を行いました。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/8e006aa4-acf0-f452-ba62-6f52d1550b00.png" width="683px">

トピックのみフィルターをかけます。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/d3f3931a-e9dd-7c8c-45ba-0064cce3fa42.png" width="683px">

事前に作成したKinesisFirehoseをアクションに追加します。
ロールは既存のロールもしくは新規で作成します。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/aebb568c-8dff-76d8-a136-b88633296f8b.png" width="683px">

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/0f7b0099-97f5-45ee-8d0f-3e57628079ac.png" width="683px">

AWSIoTのルール設定が完了しました。

## 設定の確認

ElasticSearchのコンソールにて、indeの作成とデータが格納されているか確認できます。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/ad3d0c09-d3f3-c339-17ce-281faf727667.png" width="683px">

以上でAWSIoTからKinesisFirehoseを利用してS3, ElasticSearchに格納できました。

# Kibanaにてセンサーデータを参照
ElasticSearchのコンソールに表示されているKibanaへアクセスします。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/6cb06597-c47b-c66c-20eb-fc6476d8fc87.png" width="683px">

「index name or pattern」に設定したindex名を設定します。

<img src="https://qiita-image-store.s3.amazonaws.com/0/197194/d2ea49ba-78b4-26fb-612f-4026f1b1641e.png" width="683px">

以上でKibanaのindex設定が完了しました。

# 次回更新予定
記事が長くなってしまったので、Kibanaでセンサーデータを参照する部分は予定外ですが次回とします。