使用物联网 Greengrass 经济实惠地将物联网数据直接摄入亚马逊 S3 AWS

由塞巴斯蒂安·维维亚尼 (AWS) 和 Rizwan Syed () 创作 AWS

Summary

此模式向您展示了如何使用物联网 G AWS reengrass 版本 2 设备经济高效地将物联网 (IoT) 数据直接摄取到亚马逊简单存储服务 (Amazon S3) 存储桶中。设备运行自定义组件，用于读取物联网数据，并将数据保存在永久存储（即本地磁盘或卷）中。然后，设备将物联网数据压缩为 Apache Parquet 文件，并定期将数据上传至 S3 存储桶。

您采集的物联网数据的数量和速度仅受边缘硬件功能以及网络带宽的限制。您可使用 Amazon Athena 经济高效地分析您摄取的数据。Athena 支持使用 Amazon Managed Grafana 进行压缩 Apache Parquet 文件和数据可视化。

先决条件和限制

先决条件

限制

架构

目标技术堆栈

目标架构

下图显示了一种架构，该架构旨在摄取物联网传感器数据，并将该数据存储至 S3 存储桶。

图表显示了以下工作流：

注意：流管理器 (StreamManager) 是托管组件。有关如何将数据导出到 Amazon S3 的示例，请参阅 AWS IoT G reengrass 文档中的流管理器。你可以使用本地MQTT经纪商作为组件，也可以使用其他经纪商，比如 Eclipse Mos quitto。

工具

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服务配额以及与您的云环境相关的其他因素。

输入信号集

该分析使用以下一组输入信号为基础，将物联网摄取成本与其他可用替代方案进行比较。

在此情况下，系统接收 125 个信号。每个信号为 8 字节，每 40 毫秒 (25 Hz) 会出现一次。这些信号可单独发出，也可以分组至公共有效载荷中。您可根据需要选择拆分和打包这些信号。您还可确定延迟。延迟由接收、累积和摄取数据时间段组成。

为便于比较，此场景的摄取操作以区域为基础。us-east-1 AWS成本比较仅适用于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)