IoT Greengrass を使用してコスト効率よく AWS IoT データを Amazon S3 に直接取り込む

作成者: Sebastian Viviani (AWS) と Rizwan Syed (AWS）

[概要]

このパターンは、AWSIoT Greengrass バージョン 2 デバイスを使用して、モノのインターネット (IoT ) データを Amazon Simple Storage Service (Amazon S3) バケットに直接コスト効率よく取り込む方法を示しています。デバイスは、IoT データを読み取り、データを永続ストレージ (つまり、ローカルディスクまたはボリューム) に保存するカスタムコンポーネントを実行します。次に、デバイスは IoT データを Apache Parquet ファイルに圧縮し、そのデータを定期的に S3 バケットにアップロードします。

取り込むIoTデータの量と速度は、エッジハードウェアの機能とネットワーク帯域幅によってのみ制限されます。Amazon Athena を使用すれば、取り込んだデータをコスト効率よく分析することができます。Athena は、圧縮された Apache Parquet ファイルと「Amazon Managed Grafana」によるデータの視覚化をサポートしています。

前提条件と制限

前提条件

制約事項

アーキテクチャ

ターゲットテクノロジースタック

ターゲットアーキテクチャ

次の図は、IoT センサーデータを取り込み、そのデータを S3 バケットに保存するように設計されたアーキテクチャを示しています。

この図表は、次のワークフローを示しています：

注：ストリームマネージャー (StreamManager) は管理対象コンポーネントです。Amazon S3 にデータをエクスポートする方法の例については、 AWS IoT Greengrass ドキュメントの「ストリームマネージャー」を参照してください。ローカルMQTTブローカーは、コンポーネントとして、または Eclipse Mosquitto などの別のブローカーとして使用できます。

ツール

AWS ツール

その他のツール

ベストプラクティス

アップロードされたデータには適切なパーティション形式を使用してください。

S3 バケットのルートプレフィックス名 (たとえば、"myAwesomeDataSet/"や"dataFromSource") には特定の要件はありませんが、データセットの目的がわかりやすいように、わかりやすいパーティションとプレフィックスを使用することをお勧めします。

また、クエリがデータセットで最適に実行されるように、Amazon S3 では適切なパーティション分割を使用することをお勧めします。次の例では、各 Athena クエリでスキャンされるデータ量が最適化されるように、データを HIVE形式でパーティション化しています。これにより、パフォーマンスを向上させ、コストを削減できます。

s3://<ingestionBucket>/<rootPrefix>/year=YY/month=MM/day=DD/HHMM_<suffix>.parquet

エピック

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3DataUpload",
            "Effect": "Allow",
            "Action": [
                "s3:List*",
                "s3:Put*"
            ],
            "Resource": [
                "arn:aws:s3:::<ingestionBucket>",
                "arn:aws:s3:::<ingestionBucket>/<prefix>/*"
            ]
        }
    ]
}

{
  "region": "<region>",
  "parquet_period": <period>,
  "s3_bucket": "<s3Bucket>",
  "s3_key_prefix": "<s3prefix>"
}

{
  "logsUploaderConfiguration": {
    "systemLogsConfiguration": {
    ...
    },
    "componentLogsConfigurationMap": {
      "<com.iot.ingest.parquet>": {
        "minimumLogLevel": "INFO",
        "diskSpaceLimit": "20",
        "diskSpaceLimitUnit": "MB",
        "deleteLogFileAfterCloudUpload": "false"
      }
      ...
    }
  },
  "periodicUploadIntervalSec": "300"
}

トラブルシューティング

関連リソース

追加情報

コスト分析

次のコスト分析シナリオは、このパターンでカバーされるデータインジェストアプローチが AWS クラウドのデータインジェストコストにどのように影響するかを示しています。このシナリオの料金例は、公開時の価格に基づいています。料金は変更されることがあります。さらに、コストは、AWSリージョン、AWSサービスクォータ、およびクラウド環境に関連するその他の要因によって異なる場合があります。

入力信号セット

この分析では、IoT の取り込みコストを他の利用可能な代替手段と比較するための基礎として、以下の入力信号セットを使用します。

このシナリオでは、システムは 125 個の信号を受信します。各信号は 8 バイトで、40 ミリ秒 (25 Hz) ごとに発生します。これらの信号は、個別に受信することも、共通のペイロードにまとめて送信することもできます。これらの信号は、必要に応じて分割してパックすることができます。レイテンシーも決定できます。レイテンシーは、データの受信、蓄積、および取り込みにかかる時間で構成されます。

比較のため、このシナリオの取り込みオペレーションは us-east-1AWSリージョンに基づいています。コスト比較は AWSサービスにのみ適用されます。ハードウェアや接続などの他のコストは、分析には考慮されません。

コスト比較

次の表は、各取り込み方法の月額コストを米ドル (USD) で示しています。

*サービスクォータを満たすには、データをダウンサンプリングする必要があります。つまり、この方法ではデータの一部が失われるということです。

代替方法

このセクションでは、以下の代替方法の等価コストを示します。

Parquetファイルジェネレーター

次のコード例は、Python で記述された Parquet ファイルジェネレーターの構造を示しています。このコード例は説明のみを目的としており、ご使用の環境に貼り付けても動作しません。

import queue
import paho.mqtt.client as mqtt
import pandas as pd

#queue for decoupling the MQTT thread
messageQueue = queue.Queue()
client = mqtt.Client()
streammanager = StreamManagerClient()

def feederListener(topic, message):
    payload = {
        "topic" : topic,
        "payload" : message,
    }
    messageQueue.put_nowait(payload)

def on_connect(client_instance, userdata, flags, rc):
    client.subscribe("#",qos=0)

def on_message(client, userdata, message):
    feederListener(topic=str(message.topic), message=str(message.payload.decode("utf-8")))


filename = "tempfile.parquet"
streamname = "mystream"
destination_bucket= "amzn-s3-demo-bucket"
keyname="mykey"
period= 60

client.on_connect = on_connect
client.on_message = on_message
streammanager.create_message_stream(
            MessageStreamDefinition(name=streamname, strategy_on_full=StrategyOnFull.OverwriteOldestData)
        )


while True:
   try:
       message = messageQueue.get(timeout=myArgs.mqtt_timeout)
   except (queue.Empty):
       logger.warning("MQTT message reception timed out")

   currentTimestamp = getCurrentTime()
   if currentTimestamp >= nextUploadTimestamp:
       df = pd.DataFrame.from_dict(accumulator)
       df.to_parquet(filename)
       s3_export_task_definition = S3ExportTaskDefinition(input_url=filename, bucket=destination_bucket, key=key_name)
       streammanager.append_message(streamname, Util.validate_and_serialize_to_json_bytes(s3_export_task_definition))
       accumulator = {}
       nextUploadTimestamp += period
   else:
        accumulator.append(message)