AWS IoT Greengrass
开发人员指南
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

IoT Analytics

IoT Analytics 连接器将本地设备数据发送到 AWS IoT Analytics。您可以使用此连接器作为中央枢纽来收集 Greengrass 核心设备上的传感器以及已连接的 Greengrass 设备中的数据。连接器将数据发送到当前 AWS 账户和区域中的 AWS IoT Analytics 通道。它可以将数据发送到默认目标通道和动态指定的通道。

注意

AWS IoT Analytics 是一项完全托管的服务,可让您收集、存储、处理和查询 IoT 数据。在 AWS IoT Analytics 中,可以进一步分析和处理该类数据。例如,它可用于训练用于监控机器运行状况的 ML 模型或测试新的建模策略。有关更多信息,请参阅 AWS IoT Analytics 用户指南 中的什么是 AWS IoT Analytics?

连接器接受输入 MQTT 主题上的格式化和非格式化数据。它支持两个预定义的主题,其中以内联方式指定目标通道。它还可以接收有关订阅中配置的客户定义的主题的消息。这可用于路由设备中发布到固定主题的消息或处理资源有限的设备中的非结构化或堆栈相关的数据。

此连接器使用 BatchPutMessage API 将数据(以 JSON 或 base64 编码字符串的形式)发送到目标通道。连接器可以将原始数据处理为符合 API 要求的格式。连接器将输入消息缓存在每个通道队列中并异步处理批次。它提供了可让您控制队列和批处理行为以及限制内存消耗的参数。例如,您可以配置最大队列大小、批处理间隔、内存大小和活动通道数。

ARNarn:aws-cn:greengrass:区域::/connectors/IoTAnalytics/versions/1

要求

此连接器具有以下要求:

  • AWS IoT Greengrass 核心软件 v1.7 or later。

  • Python 版本 2.7 已安装在核心设备上并且已添加到 PATH 环境变量。

  • 此连接器只能在受支持的 AWS 区域中使用。有关更多信息,请参阅 限制

  • 创建并配置所有相关的 AWS IoT Analytics 实体(通道、管道、数据存储、数据集)和工作流程。有关更多信息,请参阅 AWS IoT Analytics 用户指南 中的 AWS CLI控制台过程。

    注意

    目标 AWS IoT Analytics 通道必须使用与此连接器相同的账户并位于其所在的相同 AWS 区域中。

  • 一个添加到 Greengrass 组角色的 IAM 策略,该策略允许对目标通道执行 iotanalytics:BatchPutMessage 操作,如以下示例所示。通道必须位于当前 AWS 账户和区域中。

    { "Version": "2012-10-17", "Statement": [ { "Sid": "Stmt1528133056761", "Action": [ "iotanalytics:BatchPutMessage" ], "Effect": "Allow", "Resource": [ "arn:aws-cn:iotanalytics:区域:account-id:channel/channel_1_name", "arn:aws-cn:iotanalytics:区域:account-id:channel/channel_2_name" ] } ] }

    有关更多信息,请参阅IAM 用户指南中的添加和删除 IAM 策略

参数

MemorySize

要分配给此连接器的内存量(以 KB 为单位)。

在控制台中显示名称:Memory size (内存大小)

必需:true

类型:string

有效模式:^[0-9]+$

PublishRegion

在其中创建 AWS IoT Analytics 通道的 AWS 区域。使用与连接器相同的区域。

注意

这必须还与在组角色中指定的通道的区域匹配。

在控制台中显示名称:Publish region (发布区域)

必需:false

类型:string

有效模式:^$|([a-z]{2}-[a-z]+-\\d{1})

PublishInterval

将一批收到的数据发布到 AWS IoT Analytics 的时间间隔(以秒为单位)。

在控制台中显示名称:Publish interval (发布间隔)

必需:false

类型:string

默认值:1

有效模式:$|^[0-9]+$

IotAnalyticsMaxActiveChannels

连接器主动监控的 AWS IoT Analytics 通道的最大数量。此值必须大于 0,并且至少等于您希望在某个指定时间发布到连接器的通道的数量。

您可以使用此参数通过限制连接器可在某个给定时间管理的队列总数来限制内存消耗量。在发送所有排队的消息后,队列将被删除。

在控制台中显示名称:Maximum number of active channels (活动通道的最大数量)

必需:false

类型:string

默认值:50

有效模式:^$|^[1-9][0-9]*$

IotAnalyticsQueueDropBehavior

当队列已满时从通道队列中丢弃消息的行为。

在控制台中显示名称:Queue drop behavior (队列丢弃行为)

必需:false

类型:string

有效值:DROP_NEWESTDROP_OLDEST

默认值:DROP_NEWEST

有效模式:^DROP_NEWEST|DROP_OLDEST$

IotAnalyticsQueueSizePerChannel

在提交或丢弃消息前要在内存(每个通道)中保留的消息的最大数量。此值必须大于 0。

在控制台中显示名称:Maximum queue size per channel (每个通道的最大队列大小)

必需:false

类型:string

默认值:2048

有效模式:^$|^[1-9][0-9]*$

IotAnalyticsBatchSizePerChannel

要在一个批处理请求中发送到 AWS IoT Analytics 通道的消息的最大数量。此值必须大于 0。

在控制台中显示名称:Maximum number of messages to batch per channel (每个通道中要批处理的消息的最大数目)

必需:false

类型:string

默认值:5

有效模式:^$|^[1-9][0-9]*$

IotAnalyticsDefaultChannelName

此连接器用于已发送到客户定义的输入主题的 AWS IoT Analytics 通道的名称。

在控制台中显示名称:Default channel name (默认通道名称)

必需:false

类型:string

有效模式:^[a-zA-Z0-9_]$

创建连接器示例 (CLI)

以下 CLI 命令创建一个 ConnectorDefinition,其初始版本包含 IoT Analytics连接器。

aws greengrass create-connector-definition --name MyGreengrassConnectors --initial-version '{ "Connectors": [ { "Id": "MyIoTAnalyticsApplication", "ConnectorArn": "arn:aws-cn:greengrass:区域::/connectors/IoTAnalytics/versions/1", "Parameters": { "MemorySize": "65535", "PublishRegion": "us-west-1", "PublishInterval": "2", "IotAnalyticsMaxActiveChannels": "25", "IotAnalyticsQueueDropBehavior": "DROP_OLDEST", "IotAnalyticsQueueSizePerChannel": "1028", "IotAnalyticsBatchSizePerChannel": "5", "IotAnalyticsDefaultChannelName": "my_channel" } } ] }'

注意

该连接器中的 Lambda 函数具有很长的生命周期。

在 AWS IoT Greengrass 控制台 中,您可以从该组的 Connectors (连接器)页面添加一个连接器。有关更多信息,请参阅 Greengrass 连接器入门(控制台)

输入数据

此连接器接受有关预定义的和客户定义的 MQTT 主题的数据。发布者可以是 Greengrass 设备、Lambda 函数或其他连接器。

预定义主题

连接器支持以下两个结构化的 MQTT 主题,这些主题允许发布者以内联方式指定通道名称。

  • iotanalytics/channels/+/messages/put 主题上的格式化消息。这些输入消息中的 IoT 数据必须格式化为 JSON 和 base64 编码字符串。

  • iotanalytics/channels/+/messages/binary/put 主题上的非格式化消息。此主题上接收的输入消息将被视为二进制数据,并且可包含任何数据类型。

     

    要发布到预定义主题,请将 + 通配符替换为通道名称。例如:

    iotanalytics/channels/my_channel/messages/put
客户定义的主题

连接器支持 # 主题语法,可让其接受您在订阅中配置的任何 MQTT 主题上的输入消息。我们建议您指定主题路径,而不是在订阅中仅使用 # 通配符。 这些消息将发送到您为连接器指定的默认通道。

客户定义的主题上的输入消息将被视为二进制数据。它们可以使用任何消息格式,并且可以包含任何数据类型。您可以使用客户定义的主题来路由设备中发布到固定主题的消息。您还可以使用这些主题来接受无法将数据处理为要发送到连接器的格式化消息的设备中的输入数据。

有关订阅和 MQTT 主题的更多信息,请参阅输入和输出

组角色必须允许对所有目标通道执行 iotanalytics:BatchPutMessage 操作。有关更多信息,请参阅 要求

主题筛选条件:iotanalytics/channels/+/messages/put

使用此主题将格式化消息发送到连接器并动态指定目标通道。此主题还允许您指定在响应输出中返回的 ID。连接器验证 ID 对于其发送到 AWS IoT Analytics 的出站 BatchPutMessage 请求中的每个消息是否是唯一的。将丢弃具有重复 ID 的消息。

发送到此主题的输入数据必须使用以下消息格式。

消息属性
request

要发送到指定通道的数据。

必需:true

类型:包含以下属性的 object

message

采用 JSON 或 base64 编码字符串形式的设备或传感器数据。

必需:true

类型:string

id

请求的任意 ID。此属性用于将输入请求映射到输出响应。如果指定,响应对象中的 id 属性将设置为该值。如果您忽略此属性,则连接器将生成一个 ID。

必需:false

类型:string

有效模式:.*

示例输入
{ "request": { "message" : "{\"temp\":23.33}" }, "id" : "req123" }
主题筛选条件:iotanalytics/channels/+/messages/binary/put

使用此主题将未格式化的消息发送到连接器并动态指定目标通道。

连接器数据不会解析在此主题上收到的输入消息。它会将这些消息视为二进制数据。在将消息发送到 AWS IoT Analytics 之前,连接器会对其进行编码和格式化以符合 BatchPutMessage API 要求:

  • 连接器对原始数据进行 base64 编码并将已编码的负载包含在出站 BatchPutMessage 请求中。

  • 连接器为每个输入消息生成并分配一个 ID。

    注意

    连接器的响应输出不包含这些输入消息的 ID 关联。

消息属性

无。

主题筛选条件:#

使用此主题将任何消息格式发送到默认通道。在您的设备发布到固定主题或您希望将数据发送到无法将数据处理为连接器支持的消息格式的设备中的默认通道时,这尤其有用。

您可以在创建用于将此连接器连接到数据源的订阅中定义主题语法。我们建议您指定主题路径,而不是在订阅中仅使用 # 通配符。

连接器数据不会解析已发布到此输入主题的消息。所有输入消息将被视为二进制数据。在将消息发送到 AWS IoT Analytics 之前,连接器会对其进行编码和格式化以符合 BatchPutMessage API 要求:

  • 连接器对原始数据进行 base64 编码并将已编码的负载包含在出站 BatchPutMessage 请求中。

  • 连接器为每个输入消息生成并分配一个 ID。

    注意

    连接器的响应输出不包含这些输入消息的 ID 关联。

消息属性

无。

输出数据

此连接器将状态信息发布为输出数据。 此信息包含其接收并发送到 AWS IoT Analytics 的每个输入消息的 AWS IoT Analytics 所返回的响应。

主题筛选条件

iotanalytics/messages/put/status

示例输出:成功
{ "response" : { "status" : "success" }, "id" : "req123" }
示例输出:失败
{ "response" : { "status" : "fail", "error" : "ResourceNotFoundException", "error_message" : "A resource with the specified name could not be found." }, "id" : "req123" }

注意

如果连接器检测到可重试的错误(例如,连接错误),它将在下一个批处理中重试发布操作。 指数回退由 AWS 开发工具包进行处理。带可重试的错误的请求将添加回通道队列,以根据 IotAnalyticsQueueDropBehavior 参数进行进一步发布。

限制

此连接器受以下限制的约束。

  • 由适用于 Python (boto3) 的 AWS 开发工具包施加的针对 AWS IoT Analytics batch_put_message 操作的所有限制。

  • 由 AWS IoT Analytics BatchPutMessage API 施加的所有限制。有关更多信息,请参阅 AWS 一般参考 中的 AWS IoT Analytics 限制

    • 每个通道每秒 100000 条消息。

    • 每个批次 100 条消息。

    • 每条消息 128 KB。

    此 API 使用通道名称(而不是通道 ARN),因此,不支持将数据发送到跨区域或跨账户通道。

  • 由 AWS IoT Greengrass 核心施加的所有限制。有关更多信息,请参阅 AWS 一般参考 中的 AWS IoT Greengrass 核心限制

    以下限制可能尤其适用:

    • 设备发送的消息的最大大小为 128 KB。

    • Greengrass 核心路由器中的最大消息队列大小为 2.5 MB。

    • 主题字符串的最大长度为 256 字节的 UTF-8 编码的字符。

  • 此连接器只能在 AWS IoT GreengrassAWS IoT Analytics 都支持的 AWS 区域中使用。目前包括以下区域:

    • 美国东部(俄亥俄州) - us-east-2

    • 美国东部(弗吉尼亚北部) - us-east-1

    • 美国西部(俄勒冈) - us-west-2

    • 亚太区域(东京) - ap-northeast-1

    • 欧洲(法兰克福) - eu-central-1

    • 欧洲(爱尔兰) - eu-west-1

许可证

IoT Analytics连接器包含以下第三方软件/许可:

此连接器按照 Greengrass 核心软件许可协议发布。

另请参阅