Raspberry Pi＋PythonでLoRaWAN Deviceにチャレンジしてみる（５）

#今回の内容
長かったチャレンジ投稿も、今回で終了です。
いままでS3バケットに保存していたデータを、Redshiftに書き込むところまで構築します。

#トポロジ（AWS側）

FirehoseにAmazon Lambda関数を設定することで、独自のカスタム変換を組み込むことができるようになります。
またその際のトポロジは、下図のようにFirehoseのI/O間を、Lambda関数にバイパスするようなイメージとなります。

<img src="https://qiita-image-store.s3.amazonaws.com/0/94476/ba6353d0-69fe-7064-c3b7-072f7452bf61.png" width="50%">

#Amazon Lambda関数の作成

FirehoseからバイパスされるAmazon Lambda関数を、予め以下の手順で作成します。

1. AWSマネジメントコンソールから`Lambda`を開き、`関数の作成`をクリック
    ![lambda_001.png](https://qiita-image-store.s3.amazonaws.com/0/94476/e5a32db2-155f-66c0-90df-e470d71af2a6.png)

2. Blueprints（設計図）の`kinesis-firehose-process-record-python`をクリック
    ![lambda_002.png](https://qiita-image-store.s3.amazonaws.com/0/94476/ecea48b3-a6bc-2dfa-d8c3-837fc3cb3348.png)

    **Blueprintについて**
    この`kinesis-firehose-process-record-python`は、Firehoseに送られたストリームデータをカスタム変換することを目的に、予め用意されているPython用のひな型となります。
    デフォルトのコードは以下の内容となり、`event`にセットされているPayloadをbase64デコードした後、再度base64エンコードして返しているだけなので、結果的に何も変換処理をしていません。
    つまりコード内の`# Do custom processing on the payload here`とコメントアウトされた部分に、`payload`に対するカスタムコードを書くことによって、独自の変換処理を追加できるというわけです。

    ```python:lambda_function.py
    from __future__ import print_function

    import base64

    print('Loading function')


    def lambda_handler(event, context):
        output = []

        for record in event['records']:
            print(record['recordId'])
            payload = base64.b64decode(record['data'])

            # Do custom processing on the payload here

            output_record = {
                'recordId': record['recordId'],
                'result': 'Ok',
                'data': base64.b64encode(payload)
            }
            output.append(output_record)

        print('Successfully processed {} records.'.format(len(event['records'])))

        return {'records': output}
    ```

3. `名前`に任意の関数名（今回は`lorawan-transform`）を入力、`ロール`は`既存のロールを選択`のまま、`既存のロール`には任意のIAMロール（CloudWatch Logsにアクセス権限があればよい。今回は`lambda_basic_execution`）を選択
    ![lambda_003.png](https://qiita-image-store.s3.amazonaws.com/0/94476/a0d9cc39-a5d5-fa0d-9d25-e813db3a645f.png)

    IAMロール`lambda_basic_execution`はデフォルトで登録されていると思いますが、もし存在しなければ以下を参考にIAMロールを作成してください。

    **信頼関係**
    Amazon LambdaにAssumeRoleを許可するポリシーオブジェクト

    ```json
    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Principal": {
            "Service": "lambda.amazonaws.com"
          },
          "Action": "sts:AssumeRole"
        }
      ]
    }
    ```

    **アクセス権限**
    CloudWatch Logsに出力を許可するインラインポリシー

    ```json
    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Action": [
                    "logs:CreateLogGroup",
                    "logs:CreateLogStream",
                    "logs:PutLogEvents"
                ],
                "Resource": "arn:aws:logs:*:*:*"
            }
        ]
    }
    ```


4. ページ最下部までスクロールし、`関数の作成`をクリック
    ![lambda_004.png](https://qiita-image-store.s3.amazonaws.com/0/94476/face3418-c6c9-3b78-ec6b-a9632412c0eb.png)

5. 作成後、ページ下部の`基本設定`にある`タイムアウト`値を、1分以上に変更（Firehose側の仕様のようです）
    ![lambda_005.png](https://qiita-image-store.s3.amazonaws.com/0/94476/43352843-aae5-b969-4c3e-dd6016ce02a0.png)

6. ページ上部にある`保存`をクリック
    ※もし`保存してテスト`をクリックした場合、テスト結果が`成功`してさえいれば、この時点では問題ありません。
    ![lambda_006.png](https://qiita-image-store.s3.amazonaws.com/0/94476/13dfa058-c1b4-c38d-3cd1-565fbf803df3.png)

#Kinesis Firehoseの設定変更

1. 配信ストリーム（`lorawan_stream`）を開き、`Edit`ボタンをクリック
![firehose_001.png](https://qiita-image-store.s3.amazonaws.com/0/94476/6a979dbb-ac8e-e186-8fe4-d99b4ba0b401.png)

2. `Data transformation`を`Enabled`に変更、`Lambda function`に作成した関数`lorawan-transform`を選択、`Lambda function version`に`Latest`を選択
![firehose_002.png](https://qiita-image-store.s3.amazonaws.com/0/94476/cb11ae65-57ba-6b2f-7e1a-a4777580aeb3.png)

3. 引き続き`IAM role`に`Firehose delivery IAM role`を選択
![firehose_003.png](https://qiita-image-store.s3.amazonaws.com/0/94476/f2509a08-a769-2a87-4b94-07314d18bac2.png)

4. 別ページが開くので、`IAMロール`に`lorawan_delivery_role`を選択、`ポリシー名`がデフォルトの`新しいロールポリシーの作成`のまま、`許可`をクリック
    ![firehose_004.png](https://qiita-image-store.s3.amazonaws.com/0/94476/2aee68d9-456e-bbd0-a515-f0ceac5955d3.png)

    この設定で、以下のアクセス権限インラインポリシーが、既存のロール（lorawan_delivery_role）に追加されます。

    ```
    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "s3:AbortMultipartUpload",
                    "s3:GetBucketLocation",
                    "s3:GetObject",
                    "s3:ListBucket",
                    "s3:ListBucketMultipartUploads",
                    "s3:PutObject"
                ],
                "Resource": [
                    "arn:aws:s3:::lorawan-XXXXXXXX",
                    "arn:aws:s3:::lorawan-XXXXXXXX/*",
                    "arn:aws:s3:::%FIREHOSE_BUCKET_NAME%",
                    "arn:aws:s3:::%FIREHOSE_BUCKET_NAME%/*"
                ]
            },
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "lambda:InvokeFunction",
                    "lambda:GetFunctionConfiguration"
                ],
                "Resource": "arn:aws:lambda:ap-northeast-1:<Account ID>:function:lorawan-transform:$LATEST"
            },
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "logs:PutLogEvents"
                ],
                "Resource": [
                    "arn:aws:logs:ap-northeast-1:<Account ID>:log-group:/aws/kinesisfirehose/lorawan_stream:log-stream:*"
                ]
            },
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "kinesis:DescribeStream",
                    "kinesis:GetShardIterator",
                    "kinesis:GetRecords"
                ],
                "Resource": "arn:aws:kinesis:ap-northeast-1:<Account ID>:stream/%FIREHOSE_STREAM_NAME%"
            },
            {
                "Effect": "Allow",
                "Action": [
                    "kms:Decrypt"
                ],
                "Resource": [
                    "arn:aws:kms:region:accountid:key/DUMMY_KEY_ID"
                ],
                "Condition": {
                    "StringEquals": {
                        "kms:ViaService": "kinesis.%REGION_NAME%.amazonaws.com"
                    },
                    "StringLike": {
                        "kms:EncryptionContext:aws:kinesis:arn": "arn:aws:kinesis:%REGION_NAME%:<Account ID>:stream/%FIREHOSE_STREAM_NAME%"
                    }
                }
            }
        ]
    }
    ```

5. `Save`ボタンをクリック
![firehose_005.png](https://qiita-image-store.s3.amazonaws.com/0/94476/de1df5fc-bac4-2bf0-a22b-62d893aede36.png)

6. `Confirm Changes`ダイアログが開くので`Confirm`をクリック
![firehose_006.png](https://qiita-image-store.s3.amazonaws.com/0/94476/87d15f22-245d-aebb-9a83-68d9c3fe38cb.png)

以上でFirehoseの設定が完了です。

しばらくするとCloudWatch Logsに、Lambdaの実行結果がロギングされるようになります。

![firehose_007.png](https://qiita-image-store.s3.amazonaws.com/0/94476/4e82a2bf-4c29-7bd0-f69e-641d01666663.png)

またこの時点ではAmazon Lambda関数内でデータが変換されていないので、S3バケットには[前回](https://qiita.com/nagase/items/9943687b1d50346f2835#%E3%82%B9%E3%82%AF%E3%83%AA%E3%83%97%E3%83%88%E3%81%AE%E5%AE%9F%E8%A1%8C)と同様のデータが蓄積されるだけです。

#カスタム変換処理の追加
作成したAmazon Lambda関数に、独自の変換処理を`# Do custom processing on the payload here`行以降に追加しました。
また追記に伴い必要になったモジュール（datetime、dateutil、pytz）も、新たにインポートしています。

```python:lambda_function.py
from __future__ import print_function

import base64
import json
from datetime import datetime
import dateutil.parser
from pytz import timezone

print('Loading function')


def lambda_handler(event, context):
    output = []
    jst = timezone('Asia/Tokyo')

    for record in event['records']:
        print(record['recordId'])
        payload = base64.b64decode(record['data'])

        # Do custom processing on the payload here
        list = []
        payload_dict = json.loads(payload)
        dt = dateutil.parser.parse(payload_dict['Time'])
        jst_dt = dt.astimezone(jst)
        list.append(payload_dict['DevAddr'])
        list.append(datetime.strftime(jst_dt, '%Y-%m-%d %H:%M:%S %Z'))
        list.append(payload_dict['payload_hex'].decode('hex'))
        payload = ','.join([str(i) for i in list])
        payload += '\n'
        print(payload)

        output_record = {
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(payload)
        }
        output.append(output_record)

    print('Successfully processed {} records.'.format(len(event['records'])))

    return {'records': output}
```

上記コードでは次の要素を変換し、カンマ区切りしたものを`payload`として返すようにしています。

<dl>
  <dt>DevAddr</dt>
  <dd>値の変換は無し</dd>
  <dt>Time</dt>
  <dd>ISO8601フォーマット'YYYY-mm-ddTHH:MM:SS.DDD+1:00'を、'YYYY-mm-dd HH:MM:SS JST'に変換</dd>
  <dt>payload_hex</dt>
  <dd>ASCIIコードを文字列に変換</dd>
</dl>

尚、追加モジュールのひとつ`pytz`ですが、Amazon Lambda実行環境にはデフォルトでインストールされていません。
`pytz`をインストールするには、一旦ローカル環境に`lambda_function.py`を準備し、以下のコマンドでカレントディレクトリ直下に`pytz`をインストール、`lambda_function.py`と一緒にzipにアーカイブした後、そのzipファイルをAmazon Lambdaにアップロードする必要があります。

```
$ pip install pytz -t .
$ zip -r upload.zip lambda_function.py pytz
```

以上の変換処理を追加することにより、以下のようなCSVフォーマットでS3バケット内に蓄積されるようになります。

```
01XXXXXX,2017-11-23 11:59:15 JST,695
```

#Redshiftクラスタの準備

出力先となるRedshiftクラスタがVPCにある場合、そのクラスタは次の要件を満たす必要があるようなので、予め確認しておきます。

- パブリックアクセスが有効であること  
    Firehoseはパブリック側からRedshiftに接続してきます。
- FirehoseのパブリックIPアドレスからアクセスできること  
    東京リージョンにあるFirehoseのCIDRは`13.113.196.224/27`ということなので、グループポリシーでこのパブリックIPアドレスから5439番ポートへのアクセスを許可しておきます。

今回は既存のRedshiftクラスタを利用した為、グループポリシーへ上記CIDRを追加するだけで済みました。

これで要件を満たしたので、次のSQLを実行してテーブル`lorawan.moisture`を作成します。

```
CREATE TABLE lorawan.moisture(
  devaddr VARCHAR,
  timestamp TIMESTAMP,
  humidity SMALLINT
);
```

#配信ストリームの作成（for Redshift）

まず最初に知っておかなければならないことは、「Firehoseの配信ストリーム作成後は、そのストリームの出力先（Destination）を変更することが出来ない」ということです。

つまりS3に出力するよう設定されたストリームでRedshiftに対し出力するよう設定変更することができないため、新たにRedshiftへの配信ストリームを作成し、そのストリームに対してAPI Gatewayから書き込む必要があります。

以降は、ここまで設定してきた環境を残したまま、新たにRedshiftへ出力する処理を追加することにします。

1. Firehoseの配信ストリーム新規作成ページを開き、`Delivery stream name`に任意のストリーム名（今回は`lorawan_stream_to_redshift`）を入力し、ページ下部の`Next`をクリック
    ![firehose_redshift_001.png](https://qiita-image-store.s3.amazonaws.com/0/94476/d262a067-c12d-f81c-a2b8-1e522ca2ef6d.png)

2. `Record transformation`に`Enabled`を選択、`Lambda Function`には上記で作成したLambda関数（`lorawan-transform`）を選択、`Lambda function version`は`$LATEST`を選択し、`Next`をクリック
    ![firehose_redshift_002.png](https://qiita-image-store.s3.amazonaws.com/0/94476/86744b62-f64d-e7ab-6b7c-b32317569e7a.png)

3. `Create　delivery stream`ページの各項目に適切な値を入力し、`Next`をクリック
    ![firehose_redshift_003.png](https://qiita-image-store.s3.amazonaws.com/0/94476/d073158f-e395-c3fa-346f-0980e269ffc5.png)
    > 
    > |セクション|項目|設定値|
    > |:--|:--|:--|
    > |Select destination|Destination|**Amazon Redshift**|
    > |Amazon Redshift destination|Cluster|Redshiftクラスタを選択
    > ||User name|クラスタへの接続ユーザー名|
    > ||Password|クラスタへの接続パスワード|
    > ||Database|データベース名|
    > ||Table|上記で作成した、センサーデータを保存するテーブル名|
    > |Intermediate S3 destination|Intermediate S3 bucket|Redshiftへデータをロードする前に、一時的に利用するS3バケット名（今回は`lorawan-buffer`を新規作成）|
    > |Amazon Redshift COPY command|COPY option|**DELIMITER ',' REGION 'ap-northeast-1'**<br />Lambda関数でCSV形式に変換しているため、デリミタにカンマを指定し、更に東京リージョンを指定|
    > ||COPY command|実際に`Redshiftへ投入されるCOPYコマンドが確認できる|

4. `Buffer interval`を`60(seconds)`に変更し、`IAM role`の`Create new, or Choose`ボタンをクリック
    ![firehose_redshift_004.png](https://qiita-image-store.s3.amazonaws.com/0/94476/c54c5524-9a4a-47e0-f0d2-ddde453a772e.png)

5. 別ページが開くので、`IAMロール`に既に作成済みのロール（`lorawan_delivery_role`）を選択、`ポリシー名`はデフォルトの`新しいロールポリシーの作成`のまま、`許可`をクリック
    ![firehose_redshift_005.png](https://qiita-image-store.s3.amazonaws.com/0/94476/e72f78e1-7306-6593-824a-705aea821db9.png)
    これで以下のアクセス権限インラインポリシーが、既存のロール（`lorawan_delivery_role`）に追加されます。
    
    ```
    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "s3:AbortMultipartUpload",
                    "s3:GetBucketLocation",
                    "s3:GetObject",
                    "s3:ListBucket",
                    "s3:ListBucketMultipartUploads",
                    "s3:PutObject"
                ],
                "Resource": [
                    "arn:aws:s3:::lorawan-buffer",
                    "arn:aws:s3:::lorawan-buffer/*",
                    "arn:aws:s3:::%FIREHOSE_BUCKET_NAME%",
                    "arn:aws:s3:::%FIREHOSE_BUCKET_NAME%/*"
                ]
            },
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "lambda:InvokeFunction",
                    "lambda:GetFunctionConfiguration"
                ],
                "Resource": "arn:aws:lambda:ap-northeast-1:<Account ID>:function:lorawan-transform:$LATEST"
            },
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "logs:PutLogEvents"
                ],
                "Resource": [
                    "arn:aws:logs:ap-northeast-1:<Account ID>:log-group:/aws/kinesisfirehose/lorawan_stream_to_redshift:log-stream:*"
                ]
            },
            {
                "Sid": "",
                "Effect": "Allow",
                "Action": [
                    "kinesis:DescribeStream",
                    "kinesis:GetShardIterator",
                    "kinesis:GetRecords"
                ],
                "Resource": "arn:aws:kinesis:ap-northeast-1:<Account ID>:stream/%FIREHOSE_STREAM_NAME%"
            },
            {
                "Effect": "Allow",
                "Action": [
                    "kms:Decrypt"
                ],
                "Resource": [
                    "arn:aws:kms:region:accountid:key/%SSE_KEY_ARN%"
                ],
                "Condition": {
                    "StringEquals": {
                        "kms:ViaService": "kinesis.%REGION_NAME%.amazonaws.com"
                    },
                    "StringLike": {
                        "kms:EncryptionContext:aws:kinesis:arn": "arn:aws:kinesis:%REGION_NAME%:<Account ID>:stream/%FIREHOSE_STREAM_NAME%"
                    }
                }
            }
        ]
    }
    ```

6. 元のページに戻ったら`Next`をクリック
    ![firehose_redshift_006.png](https://qiita-image-store.s3.amazonaws.com/0/94476/d1e8bf7e-c7a5-90f8-837d-c4c2e29d9c65.png)

7. 設定内容を確認し、`Create delivery stream`をクリック
    ![firehose_redshift_007.png](https://qiita-image-store.s3.amazonaws.com/0/94476/6d03b7bb-49d0-2dba-6289-dcbe9a269be4.png)

以上でFirehose配信ストリームの作成は完了です。

#API Gatewayの設定変更

第３回の[API Gateway作成](https://qiita.com/nagase/items/b80db2d7beb4c4cad4e8#api-gateway%E4%BD%9C%E6%88%90)で作成したAPI`LoRaWAN API`に、リソース`v2`と、POSTメソッドを追加で作成します。
![apigateway_004.png](https://qiita-image-store.s3.amazonaws.com/0/94476/b42f0a47-0e52-b92b-617d-7f5f1e21320c.png)

更に追加したPOSTの統合リクエストを開き、本文マッピングテンプレートで`DeliveryStreamName`の値に、上記で作成した配信ストリーム名を指定すれば設定完了です。
![apigateway_007.png](https://qiita-image-store.s3.amazonaws.com/0/94476/4d8b5ccb-7745-e66c-4a80-d99c1c5ec021.png)

あとはテストをして問題がないか確認し、更に`prod`ステージにデプロイすることも忘れないでください。

#アプリケーションサーバ設定変更

[第２回](https://qiita.com/nagase/items/a99248d6f13079e8abca#%E3%82%A2%E3%83%97%E3%83%AA%E3%82%B1%E3%83%BC%E3%82%B7%E3%83%A7%E3%83%B3%E3%82%B5%E3%83%BC%E3%83%90%E3%81%AE%E7%99%BB%E9%8C%B2)で登録したプラットフォーム側アプリケーションサーバのアドレスを、上記で作成したPOSTメソッドの呼出しURL(`https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/api/v2`)に変更します。

![apserver.png](https://qiita-image-store.s3.amazonaws.com/0/94476/a040cd9f-c8e2-1676-37ba-3d0d1eb232dc.png)

しばらく待つとLoRaWANデバイスから送信されたセンサー値が、Redshiftに書き込まれるようになりました。

```
dwh=# SELECT * FROM lorawan.moisture;
 devaddr  |      timestamp      | humidity
----------+---------------------+----------
 01XXXXXX | 2017-11-27 13:59:02 |      699
 01XXXXXX | 2017-11-27 14:09:13 |      700
(2 行)

dwh=#
```


**（2017/11/28 追記）**
１晩経過してデータも溜まっていたので、Tableauで可視化してみました。

![tableau.png](https://qiita-image-store.s3.amazonaws.com/0/94476/5f60be03-7f99-5960-a00c-496bfa5ea985.png)

ちなみにグラフ上の定値線ですが、300を下回れば乾燥していることを表し、700を超えると水のやりすぎを表しています。

**（2017/11/28 17:35追記）**
調子に乗って[ThingSpeak](https://thingspeak.com/channels/374353)も使い、世界中にデータを公開してみました。


#最後に
- 初回からいきなりシリアル通信＆ATコマンドでやるハメになりましたが、モデムを設定していた懐かしい時代の経験が、非常に役に立ちました。
- 当初`LoRaWAN`という字面だけで身構えてた部分もありましたが、デバイスのハードウェアそのものを開発する立場でなければ、そんなにハードルは高くなかったです。
- `Kinesis Firehose`は「いままでどうして使わなかったんだろ」と思うくらい、至極簡単にストリームデータをDestinationに流せることが解りました。最近になって東京リージョンで提供されるようになったようなので、今後活用していきたいと思います。
- プラットフォームとアプリケーションサーバ間の通信に、認証機構がまだ実装されていないということなので、今後実装してもらえるんじゃないかと期待しています。