以符合成本效益的方式，使用 IoT 資料直接導入 Amazon S3 AWS

創建者塞巴斯蒂安·維維亞尼 (AWS) 和里茲萬·西德 () AWS

Summary

此模式說明如何使用 IoT Greengrass 第 2 版裝置，以符合成本效益的方式將物聯網 (IoT) 資料直接導入 Amazon 簡易儲存服務 (Amazon S3) 儲存貯體。AWS裝置執行可讀取 IoT 資料的自訂元件，並將資料儲存在永久性儲存裝置 (亦即本機磁碟或磁碟區) 中。然後，裝置會將 IoT 資料壓縮為 Apache Parquet 檔案，並定期將資料上傳到 S3 儲存貯體。

您擷取的 IoT 資料量和速度僅受邊緣硬體功能和網路頻寬的限制。您可以使用 Amazon Athena 以符合成本效益的方式分析擷取的資料。Athena 支持使用 Amazon 託管 Graf ana 壓縮 Apache 實木複合地板文件和數據可視化。

先決條件和限制

先決條件

限制

架構

目標技術堆疊

目標架構

下圖顯示旨在擷取 IoT 感應器資料並將該資料存放在 S3 儲存貯體的架構。

該圖顯示以下工作流程：

注意：串流管理員 (StreamManager) 是受管理的元件。如需如何將資料匯出至 Amazon S3 的範例，請參閱 AWS IoT Greengrass 文件中的串流管理員。您可以使用本地MQTT代理作為組件或其他經紀人，如 Eclipse 蚊子。

工具

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此案例中的定價範例是以發佈時的價格為基礎。價格可能變動。此外，您的費用可能會因您的AWS地區、AWS服務配額以及與雲端環境相關的其他因素而有所不同。

輸入信號設置

此分析使用下列一組輸入訊號做為比較 IoT 擷取成本與其他可用替代方案的基礎。

在這種情況下，系統接收 125 個信號。每個信號為 8 個字節，並且每 40 毫秒（25 Hz）發生一次。這些信號可以單獨出現或分組在一個共同的有效載荷中。您可以根據需要選擇拆分和打包這些信號。您還可以確定延遲。延遲包括接收、累積和擷取資料的時間段。

為了進行比較，此案例的擷取作業是以「us-east-1AWS區域」為基礎。費用比較僅適用於AWS服務。其他成本 (例如硬體或連線能力) 不會計入分析中。

成本比較

下表顯示每個擷取方法的每月成本，以美元 (USD) 為單位。

* 資料必須縮減取樣才能符合服務配額。這意味著此方法存在一些數據丟失。

替代方法

本節顯示下列替代方法的等效成本：

拼花文件生成器

下列程式碼範例會示範以 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)