使用 IoT Greengrass，以具成本效益的方式直接將 AWS IoT 資料擷取至 Amazon S3

由 Sebastian Viviani （AWS） 和 Rizwan Syed （AWS） 建立

Summary

此模式示範如何使用 AWS IoT Greengrass 第 2 版裝置，以經濟實惠的方式將物聯網 （IoT） 資料直接擷取到 Amazon Simple Storage Service （Amazon S3） 儲存貯體。裝置會執行自訂元件來讀取 IoT 資料，並將資料儲存在持久性儲存體 （即本機磁碟區或磁碟區） 中。然後，裝置會將 IoT 資料壓縮為 Apache Parquet 檔案，並定期將資料上傳到 S3 儲存貯體。

您擷取的 IoT 資料數量和速度僅受限於您的邊緣硬體功能和網路頻寬。您可以使用 Amazon Athena 以符合成本效益的方式分析擷取的資料。Athena 使用 Amazon Managed Grafana 支援壓縮的 Apache Parquet 檔案和資料視覺化。

先決條件和限制

先決條件

限制

架構

目標技術堆疊

目標架構

下圖顯示設計用來擷取 IoT 感應器資料的架構，並將該資料存放在 S3 儲存貯體中。

該圖顯示以下工作流程：

注意：串流管理器 （StreamManager） 是受管元件。如需如何將資料匯出至 Amazon S3 的範例，請參閱 AWS IoT Greengrass 文件中的串流管理器。您可以使用本機MQTT代理程式作為元件，或像 Eclipse Mosquitto 這樣的其他代理程式。

工具

AWS 工具

其他工具

最佳實務

針對上傳的資料使用正確的分割區格式

對於 S3 儲存貯體 （例如 "myAwesomeDataSet/"或 "dataFromSource"） 中的根字首名稱沒有特定要求，但建議您使用有意義的分割區和字首，以便輕鬆了解資料集的目的。

我們也建議您在 Amazon S3 中使用正確的分割，以便在資料集上以最佳方式執行查詢。在下列範例中，資料會以 HIVE 格式分割，以便最佳化每個 Athena 查詢掃描的資料量。這可改善效能並降低成本。

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 Cloud 中的資料擷取成本。此案例中的定價範例是以發佈時的價格為基礎。價格可能變動。此外，您的成本可能會因您的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)