5. シナリオ1: センサーデータをS3に保存する

BLEゲートウェイは、温湿度センサーが送信するBeaconを受信し、バイナリデータとしてAWS IoTに送信しています。AWS IoTのルールエンジンにご自身のセンサーデータのトピック名”sensor/<参加者番号>”を登録して、そのメッセージだけを「Amazon Kinesis Firehose」に送信します。Kinesis Firehoseでは、Lambda関数を使用して、受け取ったバイナリデータを活用が容易なJSON形式に変換し、Amazon S3へ保存します。

Amazon Kinesis Firehose は、受け取ったストリーミングデータを AWSの各種サービスに簡単にロードすることができます。ストリーミングデータをキャプチャ、変換して、Amazon Kinesis Analytics、Amazon S3、Amazon Redshift、および Amazon Elasticsearch Service に送信可能です。

_images/overview1.png

S3に保存するデータは、下記のJSON形式のデータになります。

{
  "model": "590080BC",       ← センサーのモデル番号
  "src_mac": "FFEAD7E638B8", ← センサーのMACアドレス
  "dst_mac": "AC83F3A041D2", ← ゲートウェイのMACアドレス
  "rssi": -47,               ← Beaconの電波強度
  "vbat": 3.34,              ← 電池電圧
  "tempreature": 26.97,      ← 温度
  "humidity": 42,            ← 湿度
  "@timestamp": "2017-05-11T08:34:43" ← タイムスタンプ
}

このシナリオでは、最初にデータの保存先であるS3とデータ変換用のLambda関数の設定を行い、最後にKinesis Firehoseの設定を行います。

5.1. S3バケットを作成

Amazon S3は、高い耐久性を持つ、容量無制限で安価なオブジェクトストレージ・サービスです。このシナリオでは、温湿度センサーのデータをS3に保存します。データをS3に保存すれば、AWSの各種データベースや、Amazon Athena、Amazon Redshift、Amazon EMR、Amazon Machine Lerningなどの各種分析サービスとの連携が容易になり、データレイクとして活用する事が出来ます。

データを保存するS3バケットを作成します。

サービス一覧から[S3]をクリックして開きます。

_images/s3.png

S3の設定画面が表示されれます。[バケットを作成する] ボタンを押します。

_images/s3-create-bucket.png

バケット名に、”sensor-data-<参加者番号>-<ユニークな文字>”、リージョンは、[米国東部(バージニア北部)] を選択し、[次へ]をクリックします。次の画面でも[次へ]をクリックして下さい。 S3のバケット名は、全世界でユニークな名前で有る必要があるため、「-<ユニークな文字>」をバケット名に追加し、他とコンフリクトしないようにして下さい。

項目 設定
バケット名 sensor-data-<参加者番号>-<ユニークな文字>
リージョン 米国東部(バージニア北部)
_images/s3-create-bucket-2.png

下記の画面が出るまで、[次へ]をクリックして下さい。最後に、[バケットを作成] をクリックすると、S3バケットが作成されます。

_images/s3-create-bucket-3.png

“sensor-data-<参加者番号>-<ユニークな文字>” のバケットがリージョン：[米国東部(バージニア北部)] に作成されているか確認して下さい。複数の参加者で1つのAWSアカウントを共用している場合、他のIAMユーザーのバケットも表示されます。

_images/s3-create-bucket-4.png

5.2. データ変換用のLambda関数を準備する

Amazon Kinesis Firehoseは、受け取ったデータをLambda関数を使用して処理することができます。このセクションでは、AWSにデータ変換用のLambda関数を登録します。

はじめに、Lambda関数を下記のリンクからPCにダウンロードして下さい。このzipファイルは、後ほどLambdaの設定画面でAWSにアップロードします。

https://s3-ap-northeast-1.amazonaws.com/awsiot-handson-dojo-jp/ranger-raw-data-to-json.zip

サービス一覧から[Lambda] をクリックして開きます。

_images/lambda.png

下記のような画面が表示されるので、[関数の作成] をクリックします。

_images/lambda-2.png

この画面が表示されない場合は、Lambdaの関数の画面から、[関数の作成] ボタンをクリックします。

_images/lambda-2-1.png

「設計図の選択」画面が表示されるので、[一から作成] をクリックします。

_images/lambda-3.png

トリガーの設定画面が表示されますが、何もせず、[次へ]をクリックします。

_images/lambda-4.png

先ほどダウンロードしたLambda関数のzipファイルをAWSにアップロードします。 下記のような画面が表示されるので、必要事項を入力して下さい。

項目 設定
名前 ranger-raw-data-to-json-<参加者番号>
ランタイム Python 2.7
コード エントリ タイプ ZIPファイルをアップロード
関数パッケージ ranger-raw-data-to-json.zip (先ほどダウンロードしたもの)
_images/lambda-5.png

「Lambda 関数ハンドラおよびロール」のロールで、[カスタムロールの作成] を選択すると、別画面でIAMの設定画面が出てきます。

項目 設定
ロール カスタムロールの作成
_images/lambda-custom-role.png

下記の設定を入力し、ロールの設定を行って下さい。

IAMの設定画面では、[新しいIAMロールの作成] を選択し、ロール名を入力し、[ポリシードキュメントを表示] をクリックして下さい。このIAMロールでは、Lambda関数のアクセス権限を設定します。Kinesis Firehoseからデータを受け取り、データ変換したデータをKinesis Firehoseへデータを書き戻すことを可能にし、CloudWatchログへのログの出力を許可しています。

項目 設定
IAM ロール 新しいIAMロールの作成
ロール名 lambda_KinesisFirehoseAccess-<参加者番号>
_images/lambda-role.png

[編集] をクリックし、下記のポリシードキュメントをコピペして下さい。[許可] をクリックするとIAMの設定画面は閉じます。

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

_images/lambda-role-2.png

ロールが[既存のロール] に変更され、ロール名として、”lambda_KinesisFirehoseAccess-<参加者番号>” が表示されていることを確認し、[次へ] をクリックして下さい。

_images/lambda-6.png

確認画面が表示されますので、下までスクロースし、[関数の作成] をクリックして下さい。

_images/lambda-create.png

Lambda関数の作成に成功すると、下記の様な画面が表示され、アップロードしたLambda関数のソースコードが表示されます。

_images/edit-lambda-func.png

BLEゲートウェイは、BLEフィルター設定で、指定したBLEセンサーのデータだけをAWS IoTへ送信するようになっていますが、他の参加者の温湿度センサーのBeaconも受信し、トピック “sensor/<参加者番号>”として、AWS IoTに送信しています。

AWS IoTに送信するデータの中に、BLEセンサーのMACアドレスが付加されていますので、この値を利用して、Lambda関数でBLEセンサーのBDアドレスを確認させ、ご自身の温湿度センサーが送信したデータだけをLambda関数で処理するようにします。

このLambda関数は、”my_temphumid_sensor_mac” に記載されているMACアドレスを持つデータだけを処理します。アップロードしたLambda関数の”my_temphumid_sensor_mac”を、ご自身の温湿度センサーのMACアドレスに書き換えましょう。

ご自身の温湿度センサーのラベルに記載されているMACアドレスを確認し、Lambda関数内のの”my_temphumid_sensor_mac” を書き換え、[保存] ボタンを押して下さい。

_images/src_mac.png _images/edit-lambda-func-2.png

5.3. Kinesis Firehoseの設定

Lambda関数の登録が終わりましたので、Amaozn Kinesis Firehoseの設定を行います。Kinesis Firehoseのストリーム名を指定し、データの吐き出し先をS3とします。また、データ変換を有効にして、先ほど登録した変換用のLambda関数を指定します。S3へデータを書き出す周期は、60秒周期とします。

サービス一覧から[Kinesis]をクリックして開きます。

_images/kinesis.png

Kinesisの画面が開きますので、[Firehose コンソールに移動] をクリックして下さい。

_images/firehose.png

[Create Delivery Stream] をクリックして下さい。

_images/create-delivery-stream.png

Step 1 で、”Delivery stream name” として、”sensor-data-to-s3-<参加者番号>” を設定し、他の設定は、そのままにして [Next] をクリックします。

項目 設定
Delivery stream name sensor-data-to-s3-<参加者番号>
_images/create-delivery-stream-1.png _images/create-delivery-stream-2.png

Step 2 では、Lambda関数を用いた、データ変換の設定を行います。

Lambdaでデータ変換を行いますので、Data transformation を Enabled にして、Lambda関数を選んで、[Next] をクリックして下さい。

項目 設定
Data transformation Enabled
Lambda function ranger-raw-data-to-json-<参加者番号>
_images/create-delivery-stream-3.png

Step 3 で、Destination として Amazon S3 を選択し、下記の情報を入力し、[Next] をクリックします。

項目 設定
Destination Amazon S3
Delivery stream name sensor-data-to-s3-<参加者番号>
S3 bucket sensor-data-<参加者番号>-<ユニークな文字>
_images/create-delivery-stream-3-1.png

Step 4 で、下記を入力し、[Create new, or Choose] ボタンを押します。

項目 設定
Buffer interval 60 (Elasticsearch Serviceに60秒周期でデータが追加されます)
Error Logging Disable
IAM role [Create new, or Choose] ボタンを押す
_images/create-delivery-stream-3-2.png

IAM ロール の設定画面が開きます。

“新しい IAMロールの作成” を選択し、ロール名を入力して、[許可] をクリックすると、Firehoseの設定画面に戻ります。

項目 設定
IAM ロール 新しい IAMロールの作成
ロール名 firehose_delivery_role_s3-<参加者番号>
_images/firehose-iam-role.png

設定したIAM ロールが選択されていることを確認し、[Next] をクリックして下さい。

_images/create-delivery-stream-4.png

最後に、[Create Delivery Stream] をクリックして下さい。

_images/create-delivery-stream-6.png

Kinesis Firehose Delivery Streamsの作成が開始されました。

_images/create-delivery-stream-7.png

5.4. AWS IoT ルールの設定

AWS IoTでは、受信したデータをルールエンジンを用いて、他のAWSの各種サービスと容易に連携できます。ルールエンジンの設定は、SQLステートメントとアクションで構成されます。SQLステートメントのトピックフィルターでデータを抽出し、アクションとしてAWSの各種サービスを指定することができます。

このシナリオでは、BLEゲートウェイが送信するトピック “sensor/<参加者番号>” を、トピックフィルターとして設定し、送付先(アクション)を先ほど作成したKinesis Firehose Delivery Streamsとします。

サービス一覧から[AWS IoT]をクリックして開きます。

_images/iot-servicemenu@2x.png

ルールを作成します。メニューから[Rules]をクリックします。以下の様な画面が表示された場合は、[Create rule]をクリックします。

_images/create-rule.png

あるいは、ルール(Rule)一覧が表示された場合は、[Create]をクリックします。

_images/create-rule-2.png

下記の項目を入力し、[Add action]をクリックします。

設定項目 値
Name iotrule_firehose_s3_<参加者番号>
Attribute *
Topic filter sensor/<参加者番号>

“*”は、半角の＊(アスタリスク)です。

Topic filterは、BLEゲートが送信するトピックになります。

_images/create-rule-3.png

“Send messages to an Amazon Kinesis Firehose stream” を選択し、[Confgure action] をクリックします。

_images/select-action.png

Stream nameで、先ほど作成した、Kinesis Firehose Delivery Streamsの “sensor-data-to-s3-<参加者番号>” を選択し、[Create a new role] をクリックします。

_images/configure-action.png

IAM role nameに、”IOT_to_Kinesis_Firehose-<参加者番号>” と入力し、[Create a new role] をクリックします。このIAMロールでは、Kinesis Firehoseへのデータの送信の権限を設定します。

_images/create-role.png

IAM role nameで、”IOT_to_Kinesis_Firehose-<参加者番号>” を選択し、[Update role] をクリックします。 “Successfully updated role.” と表示されたら、[Add action] をクリックします。

_images/add-action.png

Create a rule 画面に戻ります。[Create rule] をクリックして、ルールを作成します。

_images/create-rule-4.png

5.5. S3に保存されたデータを確認する

サービス一覧から[S3]をクリックして開きます。

_images/s3.png

“sensor-data-<参加者番号>-<ユニークな文字>” のバケットをクリックします。

_images/s3-check.png

データが保存されたフォルダまで辿って下さい。

60秒周期でファイルが保存される設定になっているため、ファイルが現れるまで数分かかることがあります。画面を更新しながら、待って下さい。

_images/s3-check-2.png

ファイルをクリックすると、ファイルの概要が表示されます。[ダウンロード] をクリックしてダウンロードして内容を確認して下さい。

_images/s3-check-3.png

下記の形式のセンサーデータが確認できます。

{"dst_mac": "F0B3865C39C3", "@timestamp": "2017-05-17T01:13:28", "humidity": 44, "vbat": 3.31, "tempreature": 25.25, "rssi": -44, "model": "590080BC", "src_mac": "E9070D5A15FB"}

これで、シナリオ1は、終わりです。バイナリデータをAWS IoTで受信し、Kinesis FirehoseでLambda関数を用いて、JSON形式に変換し、S3に保存することができました。