.parquet`
## 操作说明
### 设置您的环境
| Task | 说明 | 所需技能 |
| --- | --- | --- |
| 创建 S3 存储桶。 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | 应用程序开发人员 |
| 添加 IAM 权限至 S3 存储桶。 | 要向用户授予您之前创建的 S3 存储桶和前缀写入权限，请将以下 IAM policy 添加至您的 Amazon IoT Greengrass 角色：{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "S3DataUpload",
"Effect": "Allow",
"Action": [
"s3:List*",
"s3:Put*"
],
"Resource": [
"arn:aws:s3:::",
"arn:aws:s3::://*"
]
}
]
}有关更多信息，请参阅 Aurora 文档中的[创建 Amazon S3 资源 IAM policy](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/AuroraMySQL.Integrating.Authorizing.IAM.S3CreatePolicy.html)。接下来，更新 S3 存储桶的资源策略（如果需要），以允许使用正确的 AWS [主体](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-bucket-user-policy-specifying-principal-intro.html)写入。 | 应用程序开发人员 |
### 构建和部署 Amazon IoT Greengrass 组件
| Task | 说明 | 所需技能 |
| --- | --- | --- |
| 更新组件的配方。 | 当您根据以下示例[创建部署](https://docs.aws.amazon.com/greengrass/v2/developerguide/create-deployments.html)时[更新组件配置](https://docs.aws.amazon.com/greengrass/v2/developerguide/update-component-configurations.html)：{
"region": "",
"parquet_period": ,
"s3_bucket": "",
"s3_key_prefix": ""
}将 `` 替换为您的 AWS 区域、将 `` 替换为定期间隔、将 `` 替换为 S3 存储桶，将 `` 替换为前缀。 | 应用程序开发人员 |
| 创建组件。 | 请执行以下操作之一：[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | 应用程序开发人员 |
| 更新 MQTT 客户端。 | 示例代码不使用身份验证，因为该组件在本地连接至代理。如果您的场景不同，请按需要更新 MQTT 客户端部分。此外，执行下列操作：[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | 应用程序开发人员 |
### 将该组件添加至 Amazon IoT Greengrass Version 2 核心设备中
| Task | 说明 | 所需技能 |
| --- | --- | --- |
| 更新核心设备部署。 | 如果 Amazon IoT Greengrass Version 2 核心设备的部署已经存在，[请修改部署](https://docs.aws.amazon.com/greengrass/v2/developerguide/revise-deployments.html)。如果部署不存在，请[创建新部署](https://docs.aws.amazon.com/greengrass/v2/developerguide/create-deployments.html)。要为组件指定正确的名称，请根据以下内容[更新新组件的日志管理器配置](https://docs.aws.amazon.com/greengrass/v2/developerguide/log-manager-component.html)（如果需要）：{
"logsUploaderConfiguration": {
"systemLogsConfiguration": {
...
},
"componentLogsConfigurationMap": {
"": {
"minimumLogLevel": "INFO",
"diskSpaceLimit": "20",
"diskSpaceLimitUnit": "MB",
"deleteLogFileAfterCloudUpload": "false"
}
...
}
},
"periodicUploadIntervalSec": "300"
}最后，完成对 Amazon IoT Greengrass 核心设备部署的修订。 | 应用程序开发人员 |
### 验证是否将数据摄取至 S3 存储桶
| Task | 说明 | 所需技能 |
| --- | --- | --- |
| 查看 Amazon IoT Greengrass 卷日志。 | 检查以下各项：[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | 应用程序开发人员 |
| 检查 S3 存储桶。 | 验证数据是否正在上传至 S3 存储桶。您可以看到每个时间段在上传的文件。您还可以通过查询下一部分中的数据，验证数据是否已上传至 S3 存储桶。 | 应用程序开发人员 |
### 设置来自 Athena 的查询
| Task | 说明 | 所需技能 |
| --- | --- | --- |
| 创建数据库和表。 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | 应用程序开发人员 |
| 授予 Athena 数据访问权限。 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | 应用程序开发人员 |
## 问题排查
| 问题 | 解决方案 |
| --- | --- |
| MQTT 客户端无法连接 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) |
| MQTT 客户端订阅失败 | 验证 MQTT 代理权限。如果你有来自 AWS 的 MQTT 代理，请参阅 [MQTT 3.1.1 代理 (Moquette)](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-moquette-component.html) 和 [MQTT 5 代理 (EMQX)](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-emqx-component.html)。 |
| 无法创建 Parquet 文件 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) |
| 对象未上传至 S3 存储桶 | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) |
## 相关资源
+ [DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html)（Pandas 文档）
+ [Apache Parquet 文档](https://parquet.apache.org/docs/)（Parquet 文档）
+ [开发 Amazon IoT Greengrass 组件](https://docs.aws.amazon.com/greengrass/v2/developerguide/develop-greengrass-components.html)（Amazon IoT Greengrass 开发人员指南，第 2 版）
+ [将 Amazon IoT Greengrass 组件部署至设备](https://docs.aws.amazon.com/greengrass/v2/developerguide/manage-deployments.html)（Amazon IoT Greengrass 开发人员指南，第 2 版）
+ [与本地物联网设备互动](https://docs.aws.amazon.com/greengrass/v2/developerguide/interact-with-local-iot-devices.html)（Amazon IoT Greengrass 开发人员指南，第 2 版）
+ [MQTT 3.1.1 代理 (Moquette)](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-moquette-component.html)（Amazon IoT Greengrass 开发人员指南，第 2 版）
+ [MQTT 5 代理 (EMQX)](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-emqx-component.html)（Amazon IoT Greengrass 开发人员指南，第 2 版）
## 附加信息
**成本分析**
以下成本分析场景演示了此模式中涵盖的数据摄取方法如何影响 Amazon Web Services Cloud 中的数据摄取成本。此场景中的定价示例基于发布价格。价格可能会发生变化。此外，您的费用可能会有所不同，具体取决于您的 AWS 区域、AWS 服务限额以及与云环境相关的其他因素。
*输入信号集*
该分析使用以下一组输入信号为基础，将物联网摄取成本与其他可用替代方案进行比较。
|
|
| 信号数量 | 频率 | 每个信号的数据 |
| --- |--- |--- |
| 125 | 25 Hz | 8 字节 |
在此情况下，系统接收 125 个信号。每个信号为 8 字节，每 40 毫秒 (25 Hz) 会出现一次。这些信号可单独发出，也可以分组至公共有效载荷中。您可根据需要选择拆分和打包这些信号。您还可确定延迟。延迟由接收、累积和摄取数据时间段组成。
为了便于比较，此场景的摄取操作基于 `us-east-1` AWS 区域。成本比较仅适用 Amazon Web Services。硬件或连接等其他成本未计入分析。
*成本比较*
下表显示了每种摄取方法的每月费用（以美元为单位）。
|
|
| 方法 | 月度成本 |
| --- |--- |
| AWS 物联网 SiteWise *\$1* | 331.77 美元 |
| 带有数据处理包的 AWS IoT SiteWise Edge（将所有数据保存在边缘） | 200 美元 |
| AWS IoT Core 和 Amazon S3 访问原始数据规则 | 84.54 美元 |
| 边缘 Parquet 文件压缩并上传至 Amazon S3 | 0.5 美元 |
\$1必须对数据进行缩减采样，才能符合服务限额。这意味着使用此方法会丢失数据。
*替代方法*
本节显示了以下替代方法等效成本：
+ **AWS IoT SiteWise** — 每个信号都必须以单独的消息形式上传。因此，每月的消息总数为 125×25×3600×24×30，相当于每月 81 亿条消息。但是，AWS IoT 每个属性每秒 SiteWise 只能处理 10 个数据点。假设将数据缩减到 10 Hz，则每月的消息数量将减少到125×10×3600×24×30，即 32.4 亿条。如果您使用发布者组件，该组件以 10 个为一组（每百万封邮件 1 美元）打包，则每月的费用为每月 324 美元。假设每条消息为 8 字节 (1 Kb/125)，则为 25.92 Gb 的数据存储空间。这增加了每月 7.77 美元的费用。第一个月的总费用为 331.77 美元，每月增加 7.77 美元。
+ **带有数据处理包的 AWS IoT SiteWise Edge，包括在边缘完全处理的所有模型和信号（即无需云端接入）**— 您可以使用数据处理包作为替代方案，以降低成本并配置在边缘计算的所有模型。即使没有进行实际计算，也可以仅用于存储和可视化。在这种情况下，必须为边缘网关使用强大硬件。每月的固定费用为 200 美元。
+ **MQTT 直接接入 AWS IoT Core by MQTT 以及物联网规则，以将原数据存储至 Amazon S3** — 假设所有信号都发布在公共负载中，则发布到 AWS IoT Core 的消息总数为 25×3600×24×30，即每月 6,480 万条。按每百万条消息 1 美元算，每月费用为 64.8 美元。按每百万条规则激活 0.15 美元，每条消息一条规则，每月增加 19.44 美元的费用。按照 Amazon S3 中每 Gb 存储空间 0.023 美元成本，每月再增加 1.5 美元（每月增加以反映新数据）。第一个月的总费用为 84.54 美元，每月增加 1.5 美元。
+ **在边缘压缩 Parquet 文件中的数据并上传至 Amazon S3（建议的方法**）— 压缩率取决于数据的类型。使用针对 MQTT 测试的相同工业数据，整个月的总产出数据为 1.2 Gb。每月的费用为 0.03 美元。其他基准测试中描述的压缩率（使用随机数据）约为 66％（更接近最坏的情况）。总数据量为 21 Gb，每月花费 0.5 美元。
**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)
```