AWS Advent Calendar 2022 初日
普通に書いてもアレなんで、岡山県出身なので岡山弁口調で書きます
所要時間15~20分

記事のゴールは何じゃあ！？

AWS IoT CoreへPublishされるデータを分析したいことがあったんじゃ！
じゃけぇ、AWS IoT Analyticsを使って収集したデータをQuickSightで可視化する方法をハンズオン形式で紹介するでぇ！
この記事でAWS IoT Analyticsの構築が学べるどぉ！

対象者

AWS IoT Analyticsに興味あるんじゃ！って方
すぐにでもやってみたいんじゃ！って方

各サービス概要

AWS IoT Analytics

膨大な量の IoT データの高度な分析を簡単に実行および操作できる完全マネージド型サービス

クセが強いっ！

AWS IoT Core

ユーザーのすべての IoT デバイスとアプリケーションとの間で、低レイテンシーと高スループットで安全にメッセージを送受信できる。

ぼっけぇサービスじゃ！

QuickSight

クラウドネイティブのサーバーレスビジネスインテリジェンスサービス。データ駆動型の意思決定を行える。

カタカナのクセェ！

お題

例えばな、あなたは販売している車の運行を管理している人で、いろんな車がAWSへ接続しとったとしよう
車の状態や情報をAWS IoTへ毎秒MQTT通信でデータを送っとったら分析をやりとうなるわな

「いろんな車の走行距離を知りたいわー」
「グラフ化して分析したいわー」

ってなるわな
そんな時、どうすんなっちゅう話や

AWS IoT Analytics使ったらええがな！

これがベストプラクティスやないか

ほな、やってみるんじゃ！

1. AWS IoT Analyticsを作るんじゃ

1.1 S3バケット用意せな

今回は顧客マネージド型ストレージAmazon S3バケットを使用するハンズオンでいくどぉ

AWS IoT Analytics に組み込みのサービスマネージド型ストレージも選択できるけぇ、
S3を使わない場合はこのセクションは読み飛ばしてもかまわんけぇの

1.1.1 AWS IoT Analyticsに入ってきた生データを保存するバケット

まずAWS IoT Analyticsに入る生データを入れるバケットを用意じゃ
バケット名はお好みで

作成完了

IoT AnalyticsからS3への権限としてバケットポリシーを設定じゃ

設定するjson

{
    "Version": "2012-10-17",
    "Id": "PolicyID01",
    "Statement": [
        {
            "Sid": "Sid01",
            "Effect": "Allow",
            "Principal": {
                "Service": "iotanalytics.amazonaws.com"
            },
            "Action": [
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:ListMultipartUploadParts",
                "s3:AbortMultipartUpload",
                "s3:PutObject",
                "s3:DeleteObject"
            ],
            "Resource": [
                "arn:aws:s3:::${バケット名}",
                "arn:aws:s3:::${バケット名}/*"
            ]
        }
    ]
}

設定完了

1.1.2 AWS IoT Analyticsのパイプラインを通過したデータを保存するバケット

データストア経由で保存するS3バケットの準備をするで

IoT AnalyticsからS3への権限としてバケットポリシーを設定じゃ

設定するjson

{
    "Version": "2012-10-17",
    "Id": "PolicyID01",
    "Statement": [
        {
            "Sid": "Sid01",
            "Effect": "Allow",
            "Principal": {
                "Service": "iotanalytics.amazonaws.com"
            },
            "Action": [
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:ListMultipartUploadParts",
                "s3:AbortMultipartUpload",
                "s3:PutObject",
                "s3:DeleteObject"
            ],
            "Resource": [
                "arn:aws:s3:::${バケット名}",
                "arn:aws:s3:::${バケット名}/*"
            ]
        }
    ]
}

設定完了じゃ

1.2 チャネルを用意せな

続いては、チャネルを作成をしていくでぇ！
チャネルは、MQTTトピックからデータを収集し、未処理データメッセージをアーカイブしてから、データをパイプラインに流す機能じゃ

AWSマネジメントコンソールからAWS IoT Anaylticsのチャネル作成を選択じゃ

チャネル詳細

チャネル名を入力じゃ
名前はお好みで

ストレージオプション選択

ストレージオプションの選択は顧客マネージド型ストレージにし

1.1.1 AWS IoT Analyticsに入ってきた生データを保存するバケット

で用意したS3バケットを設定じゃ
S3を使わないならサービスドマネージド型ストレージを選択しといてな

選択したS3に対してAWS IoT Analyticsが使うアクセス権限のロールを作成する必要があるからロール名を入力せんと

ソース設定(オプション)

今回のハンズオンとしてはソースは使わんけど、
実稼働でIoT core経由でIoT AnalyticsにMQTTトピックのデータを流したい場合は対象トピックを設定する必要があるからソースの設定も紹介しとくで！

IoT Coreからチャネルへデバイスごとにデータ送るのに手間がかかるんで、今回はBatchPutMessage API使ってチャネルへデータ送信するやり方を後で説明するど

対象トピックからメッセージが受信できるかあとで確認からメッセージを表示をクリックじゃ

ここでのロール作成はAWS IoT Coreが対象のチャネルに対してデータ送信するのに必要なアクセス権限を付与してあげんといけんから忘れるなよ！

新しいタブページでAWSマネージドコンソールのAWS IoT Core画面を開いて、試しに対象トピックにメッセージを送信してみるで

AWS IoT Analyticsのソース設定画面で受信したメッセージが表示されたら成功じゃ！

チャネル完成

確認画面から設定確認して作成すればチャネル完成！

1.3 データストアを用意せな

パイプラインから処理されたメッセージをデータストアに保存する必要があるんで、データストアを設定するでい！

AWSマネジメントコンソールからAWS IoT Anaylticsのデータストア作成を選択じゃ

データストア詳細

データストアIDはお好みで

ストレージタイプ設定

ストレージオプションの選択は顧客マネージド型ストレージにし

1.1.2 AWS IoT Analyticsのパイプラインを通過したデータを保存するバケット

で用意したS3バケットを設定じゃ

他のAWSサービスからも利用されることがある場合は、顧客マネージド型S3ストレージを利用するんじゃ
AWS IoT Analyticsからのみであればサービスマネージド型ストレージを利用しといてな

選択したS3に対してAWS IoT Analyticsが使うアクセス権限のロールを作成する必要があるからロール名を入力せな

データ形式

デフォルトのJSONを設定

データパーティション追加

ここはデフォルト通りで有効化せずに次じゃ

データストア完成

確認画面から設定確認して作成すればデータストア完成！

1.4 パイプラインを用意せな

パイプラインはメッセージをデータストアに保存する前にメッセージを処理する機能じゃ！

AWSマネジメントコンソールからAWS IoT Anaylticsのパイプライン作成を選択

パイプラインID, ソースをセットアップ

パイプライン名を入力し
前述で記載した
パイプラインソースはチャネル名を設定して
パイプライン出力はデータストアIDを設定じゃ

メッセージ属性推論

デフォルトのまま次じゃ

メッセージの強化、変換、フィルタ

デフォルトのまま、次じゃ次じゃ

パイプライン完成

確認画面から設定確認して作成すればパイプライン完成！

2. データセットを作るんじゃ

データストアからデータを取得するのにデータセットが必要じゃ！

2.1 データをぶち込めぇ！

本来ならIoT Coreを経由してIoT Analyticsへデータ送信するのがあるべき姿じゃけど、ポチポチTopicに対してPublishするのは大変じゃ
今回はハンズオンなんで手っ取り早くSDKを利用してChannelへデータを流し込むど！

Cloud9からでもできるけど、今回はローカル環境からぶち込むんじゃ
ダミーデータとして、車ごとの走行距離を30日分としてチャネルへメッセージを送信するで

※予めローカル環境にaws profileを設定しておく

sample-car.py

import boto3
from boto3.session import Session
import copy
from datetime import datetime, date, timedelta
import random
import uuid
import json

# profile_nameはIAMで作成したcredentialsのprofile名を指定
# Cloud9経由でやる場合は指定は↓はコメントアウト
session = boto3.Session(profile_name='')
iota = session.client('iotanalytics')

# Cloud9経由でやる場合は指定は↓コメント外す
# iota = boto3.client('iotanalytics')

# AWS IoT Analyticsに送信するデータ構造
DATA_TEMPLATE = {
  "timestamp": "",
  "device_id": "",
  "mileage": 0
}

# AWS IoT Analyticsのチャンネル名(自分の環境に合わせて修正)
CHANNEL_NAME="iot_analytics_channel20221201"

# 過去30日分のダミーデータを作成。
now = datetime.now()
n_days_ago = now - timedelta(days=10)
start_time = int(n_days_ago.timestamp())
end_time = int(now.timestamp())

cnt = 0
total = 0
messages = []

# 現在まで、1分ごとのデータを作成
while end_time > start_time:
  dt = datetime.fromtimestamp(start_time)
  item = copy.copy(DATA_TEMPLATE)
  item["timestamp"] = dt.strftime("%Y-%m-%d %H:%M:%S")
  item["device_id"] = 'car' + str(random.randrange(1, 100))
  item["mileage"] = random.randrange(1, 10000)

  messages.append({'messageId':str(uuid.uuid4()),'payload': json.dumps(item)})
  cnt = cnt + 1

  # batch putで送れるデータは、デフォルトだと100件まで
  # https://docs.aws.amazon.com/iotanalytics/latest/userguide/limits.html
  if cnt >= 100:
    total = total + cnt
    print("batch put:", total)
    ret = iota.batch_put_message(channelName=CHANNEL_NAME, messages=messages)
    cnt = 0
    messages = []

  start_time = start_time + 60

if cnt > 0:
  total = total + cnt
  print("batch put:", total)
  ret = iota.batch_put_message(channelName=CHANNEL_NAME, messages=messages)

# pythonファイルを置いているディレクトリ直下で実行
$ python3 sample-car.py

2.2 データセット作るんじゃ

データに対してクエリを実行するためにデータセットを作成するでぇ！

AWSマネジメントコンソールからAWS IoT Anaylticsのデータセット作成を選択

タイプ選択

今回はSQLを利用するんで「SQLを作成」を選択するど

データセット設定の詳細

データセット名の入力と作成したデータストアを指定じゃ

SQLクエリ作成

Pythonからチャネルへ流した車それぞれのメッセージデータに走行距離の平均値、最小値、最大値を検索できるようにSQL文を設定しておくんじゃ

テストクエリを実行して、実際にクエリ結果が返るか確認しとく

SQL文

SELECT date_trunc('hour', date_parse(timestamp, '%Y-%m-%d %H:%i:%S')) as dt_hour,
device_id,
avg(mileage) as mile_avg,
min(mileage) as mile_min,
max(mileage) as mile_max
FROM iot_analytics_datestore20221201
GROUP BY 1, 2