本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # InfluxDB 当您要查询设备遥测以了解运行趋势或实时监控时间序列数据时，请选择此操作。您可以使用客户端或服务器端批处理将多个点合并到一个写入请求中。 Amazon IoT Core 将每条消息转换为InfluxDB行协议并将其写入指定的数据库和表。有关该格式的更多信息，请参阅文档[中的 ](https://docs.influxdata.com/influxdb3/enterprise/reference/line-protocol/) InfluxDB行协议。 InfluxData 有关托管集群的信息，请参阅 Inf [ luxDB 的 ](https://docs.amazonaws.cn/timestream/latest/developerguide/influxdb3.html) Amazon Timestream。 **Topics** + [先决条件](#influxdb-rule-action-prerequisites) + [InfluxDB 操作目的地](#influxdb-action-destination) + [InfluxDB 术语映射](#influxdb-terminology-mapping) + [参数](#influxdb-rule-action-parameters) + [批处理](#influxdb-rule-action-batching) + [Per-element 阵列有效载荷的模板](#influxdb-per-element-templates) + [InfluxDB 记录内容](#influxdb-record-content) + [错误操作](#influxdb-error-actions) + [示例](#influxdb-rule-action-examples) ## 先决条件 此规则操作具有以下先决条件： + **InfluxDB 操作目的地 ** — 创建一个 InfluxDB 操作目的地，为您的InfluxDB实例指定终端节点、版本和证书。 Amazon IoT Core 在发送流量之前验证端点所有权。请参阅[InfluxDB 操作目的地](#influxdb-action-destination)。 + **一个 IAM 角色**， Amazon IoT 可以假定写入您的 InfluxDB 数据库，并对存储您的 InfluxDB 证书的密钥执行`GetSecretValue`操作。有关更多信息，请参阅 [授予 Amazon IoT 控制其所需的访问权限](iot-create-role.md)。 + **InfluxDB 凭证存储在 Amazon Secrets Manager** — 对于 InfluxDB V3，在创建 InfluxDB 集群时，适用于 InfluxDB 的 Amazon Timestream 会自动预置密钥管理器密钥。密钥包含集群凭证。对于 InfluxDB V2，从您的 InfluxDB 实例生成 All Access 或自定义 API 令牌。将令牌值作为纯文本机密存储在中。 Amazon Secrets Manager有关创建令牌的更多信息，请参阅 InfluxData 文档[中的](https://docs.influxdata.com/influxdb/v2/admin/tokens/create-token/)创建令牌。在运行时，规则操作调用`secretsmanager:GetSecretValue`以检索这些凭据，然后再对您的InfluxDB端点进行身份验证。 + **消息负载中的时间戳 **-消息负载中的每个对象都必须包含一个名为`timestamp`（区分大小写）的键和一个整数 Unix 纪元值。时间戳单位也可以在物联网规则定义中设置。 Amazon IoT Core 不会为InfluxDB操作生成时间戳，除非使用InfluxDB作为错误操作。 **注意** `time`不能使用诸如`ts`或之类的别名来代替`timestamp`。 + **HTTPS 连接 ** — 必须可以通过 HTTPS 访问您的 InfluxDB 实例。 Amazon IoT Core 支持的出站端口是：443、8443、8086（InfluxDB V2 的默认端口）和 8181（InfluxDB V3 的默认端口）。 + **JSON-format 有效负载 ** — InfluxDB 操作仅处理 JSON 有效负载。如果您的设备发布二进制或 Protobuf 数据，请在操作运行之前使用规则 SQL 中的[`decode()`](binary-payloads.md#binary-payloads-protobuf)函数将其转换为 JSON。 ## InfluxDB 操作目的地 在使用InfluxDB规则操作之前，必须先创建InfluxDB操作目标。目标定义了您的InfluxDB实例的连接参数。 Amazon IoT Core 然后验证端点所有权。 ### 创建目的地 使用 `CreateTopicRuleDestination` API 创建 InfluxDB 目标： ``` { "destinationConfiguration": { "influxDBConfiguration": { "endpoint": "https://my-instance.timestream-influxdb.us-west-2.amazonaws.com:8086", "influxDBVersion": "V2", "secretId": "arn:aws:secretsmanager:us-west-2:111122223333:secret:my-influxdb-credentials-AbCdEf" } } } ``` ### 目的地参数 | 参数 | Type | 必需 | 描述 | | --- | --- | --- | --- | | endpoint | 字符串 | 是 | 您的 InfluxDB 实例的 HTTPS 终端节点 URL。不支持 HTTP。支持的端口：443、8086、8181、8443。 | | influxDBVersion | 字符串 | 是 | InfluxDB 版本。有效值：V2、V3。 | | secretId | 字符串 | 是 | 包含您的InfluxDB令牌的 Amazon Secrets Manager 密钥的名称或ARN。 | | secretType | 字符串 | 否 | 机密值的类型。有效值：SecretString、SecretBinary。 | | secretKey | 字符串 | 否 | 包含身份验证令牌的秘密 JSON 中的密钥。仅当密钥是具有多个密钥的 JSON 对象时才需要。 | ### 端点所有权验证 当您创建InfluxDB操作目标时，使用您提供的 Amazon IoT Core 凭据对InfluxDB API进行身份验证来验证端点所有权： + **InfluxDB V2**： Amazon IoT Core 调用端点。`/api/v2/me` + **InfluxDB V3**： Amazon IoT Core 调用列表数据库端点 () `GET /api/v3/configure/database`。 成功响应 (2xx) 将目标状态设置为。`ENABLED`失败时， Amazon IoT Core 将状态设置为`ERROR`。要重试验证，请在状态设置为`IN_PROGRESS`的情况下调用`UpdateTopicRuleDestination`。 ## InfluxDB 术语映射 以下术语在 InfluxDB V2 和 V3 之间有所不同。 | Amazon IoT Core 参数 | InfluxDB V2 术语 | InfluxDB V3 术语 | | --- | --- | --- | | databaseName | 存储桶 | 数据库 | | tableName | 测量值 | 表 | **注意** 如果您要从 Amazon Timestream 规则操作迁移，则`dimensions`参数映射到 InfluxDB 操作中的标签，查询结果属性映射到字段。 ## 参数 使用 InfluxDB 操作创建 Amazon IoT 规则时，必须指定以下信息： `destinationArn` InfluxDB 操作目标的 ARN。请参阅[InfluxDB 操作目的地](#influxdb-action-destination)。支持[替换模板](iot-substitution-templates.md)：否 `roleArn` 授予访问密钥管理器密钥 Amazon IoT 权限的 IAM 角色的 ARN。请参阅[先决条件](#influxdb-rule-action-prerequisites)。支持[替换模板](iot-substitution-templates.md)：否 `databaseName` 要写入记录的InfluxDB数据库（在InfluxDB v2 * 中称为*存储桶，在InfluxDB v3 * 中称为*数据库）的名称。支持[替代模板](iot-substitution-templates.md)：否。要将数据路由到不同的数据库，请为每个数据库创建单独的规则操作。 `tableName` 要写入记录的表（*在InfluxDB v2中称为*度量值，*在InfluxDB v3中称为*表）的名称。支持[替换模板](iot-substitution-templates.md)：是。 `organization` InfluxDB 组织名称。InfluxDB v2 是必需的。如果你在InfluxDB v3中加入这个参数，它将被忽略。支持[替代模板](iot-substitution-templates.md)：否。 `tags` 每个点的元数据，指定为地图。每个地图键都是一个标签名称，每个地图值都是相应的标签值。为提高查询性能对标签进行了索引。 + 在InfluxDB V3中，每个标签名称在表中必须是唯一的，并且不能重复字段名称。 + 标签值支持消息范围和每个元素的替换模板。 `timestampUnit` 有效载荷中时间戳值的精度。有效值：`s`（秒）\|`ms`（毫秒）\|（微秒）\|`us`（纳秒）。`ns`默认值：`ms`。支持[替代模板](iot-substitution-templates.md)：否。 `batchConfig` （可选） Server-side 批处理配置。有关更多信息，请参阅 [批处理](#influxdb-rule-action-batching)。 + `maxBatchSize`— 每批的最大分数。有效范围：1—500。 + `maxBatchOpenMs`— 保持批次打开状态的最大时间（以毫秒为单位）。有效范围：5—1,000。 + `maxBatchSizeBytes`— 刷新前的最大总大小（以字节为单位）。有效范围：100—131,072。 + `batchAcrossTopics`：布尔值。当时`true`，该批次包含来自不同主题的消息的积分。默认值：`false`。 ## 批处理 InfluxDB 操作支持两种批处理模式。 ### Client-side 批处理 您的物联网设备将时间序列数据批处理为 JSON 数组，并将其作为单个 MQTT 消息发布。通过客户端批处理，每个数组元素在单个写入请求中成为一个线路协议点。您不需要额外的配置。 ### Server-side 批处理 在将单个消息写入InfluxDB之前，使用服务器端批处理对它们进行分组。使用参数配置服务器端批处理。`batchConfig`当首先达到任何配置的限制（`maxBatchSize`、或`maxBatchSizeBytes`）时`maxBatchOpenMs`，将刷新该批次。 **注意** 可以同时配置服务器端批处理和客户端批处理（JSON 数组有效负载）。 InfluxDB 操作是根据出站负载大小进行计量的，以 5 KiB 为增量。 Line-protocol 还会计量转换失败次数。 ## Per-element 阵列有效载荷的模板 当您的物联网设备以 JSON 数组的形式发送批处理的时间序列数据时，您可以使用 ** “每元素替换模板” ** 来解析每个单独数组元素的值。这会将每个数据点路由到不同的表或应用特定元素的标签。有关更多信息，请参阅[Per-element模板](https://docs.amazonaws.cn/iot/latest/developerguide/per-element-templates.html)。 ### 二的语法 [替代模板](iot-substitution-templates.md) + `${expression}`— 在消息范围内针对传入的设备消息进行解析。每条消息计算一次；相同的值适用于数组中的每个点。 + `@{expression}`— 针对规则的 SQL SELECT 语句生成的有效负载中的单个元素在元素范围内进行解析。 Re-evaluated 对于每个数组元素，因此每个点可以得到不同的值。有关更多信息，请参阅[`@{expression}`参考](https://docs.amazonaws.cn/iot/latest/developerguide/per-element-expression.html)。 使用 `@{...}` in `tableName` 和 tag 值分别针对每个数组元素解析表达式。 ### 示例 给定以下设备负载（JSON 数组）： ``` [ {"measurement_type": "temperature", "room": "kitchen", "timestamp": 1700000000000, "value": 23.5}, {"measurement_type": "humidity", "room": "bedroom", "timestamp": 1700000001000, "value": 60.1} ] ``` 以及以下操作配置： ``` { "influxDB": { "destinationArn": "arn:aws:iot:us-west-2:111122223333:ruledestination/influxdb/abc123", "roleArn": "arn:aws:iam::111122223333:role/iot-influxdb-role", "databaseName": "sensor_data", "tableName": "@{measurement_type}", "tags": { "room": "@{room}" }, "timestampUnit": "ms" } } ``` 生成的线路协议输出包含两点： ``` temperature,room=kitchen value=23.5 1700000000000 humidity,room=bedroom value=60.1 1700000001000 ``` **注意** 引用的字段将从线路协议字段集中移除，因此它也不会显示为字段。`@{...}`在此示例中，`measurement_type`成为表名并`room`成为标签，因此两者都不出现在字段集中，`value`这是唯一的字段。 ### 限制 + `@{...}`仅在 InfluxDB 操作配置（`tableName`和标签值）中支持。 + 您不能`@{...}`在规则 SQL（SELECT/WHERE子句）、错误操作定义或任何其他规则操作中使用。 + 内部仅支持字段引用`@{...}`。不支持函数。 + 每个值最多支持一个`@{...}`标记。单个值中的多个标记会产生 API 异常。 + 不能将`${...}`和混合`@{...}`在同一个值中。混合会产生 API 异常。 + 单个 JSON 对象被视为一个元素数组。 ## InfluxDB 记录内容 对于后SQL查询结果中的每条记录，生成的InfluxDB行协议点包含以下组件： | 组件 | 来源 | | --- | --- | | 表 | tableName参数值 | | 标签 | 参数中的键值对 tags | | 字段 | 其余有效负载属性未用作标签、表名或时间戳 | | Timestamp | 从有效载荷中提取的timestamp密钥 | ### 数据类型转换 Amazon IoT Core 将 JSON 值转换为 InfluxDB 线路协议类型，如下所示： | JSON 类型 | 线路协议类型 | 示例 | | --- | --- | --- | | 整数（−2³ 到 2³ −1） | 有符号整数（i后缀） | 42 → 42i | | 整数（2³ 到 2−1） | 无符号整数（u后缀） | 9223372036854775808 → 9223372036854775808u | | 外部整数（−2³ 到 2−1） | 已拒绝（触发错误操作） | — | | 浮点数/十进制 | IEEE-754 64 位浮点 | 23.5 → 23.5 | | 布尔值 | t 或者 f | true → t | | 字符串 | 引用的字符串 | "active" → "active" | | Null | 省略（未写入） | — | | 对象或数组 | 压缩的 JSON 字符串（去掉空格） | {"a":1} → "{\\"a\\":1}" | ### 命名限制 + 字段键和标签键不能为空或以下划线 (`_`) 开头。 + 在InfluxDB V3中，表名和标签密钥必须以字母或数字开头。 + 字段键、标签键和标签值中的逗号、等号和空格会自动转义。 + 测量：逗号和空格会自动转义。 + 标签在序列化之前按键的字母顺序排序，以提高摄取性能。 + 输出中省略空标签值。 ### 保留的密钥 在线路协议转换期间，将从设置的字段中删除以下有效负载密钥： + `timestamp`— 用作点时间戳。 + 由`tableName`（通过`${...}`或`@{...}`）引用的密钥-用作表名。 + 标签值（通过`${...}`或`@{...}`）引用的密钥-用作标签值。 ## 错误操作 如果InfluxDB操作失败，则会触发配置的错误操作。 ### 使用 InfluxDB 作为错误操作 您可以将InfluxDB操作配置为任何规则的错误操作。整个有效载荷被写入到单`tableName`一记录中。 Message-scope 错误操作的`tableName`和支持替换模板 (`${...}`) `tags`。 ### 错误操作输出 触发错误操作时，输出负载包含:`ruleName`、、`topic`、`cloudwatchTraceId`、`clientId``sourceIp`、`base64OriginalPayload`（Base64-encoded 原始消息）和一个`failures`数组，其中每个条目都有`failedAction``failedResource`、和`errorMessage`。 在服务器端批处理中，负载使用`payloadsWithMetadata`，每条不同的入站 MQTT 消息都有一个条目。每个失败的`affectedIds`值都指的是这些消息条目的`id`值；它们不是积分索引。即使客户端的JSON数组生成多个InfluxDB点，它也是一条入站消息。有关完整的有效负载格式，请参阅[批处理的错误操作](http_batching.md#batching_errors)。 | 故障场景 | 说明 | | --- | --- | | 目标已禁用或出错 | InfluxDB 操作目标未启用。验证端点所有权验证成功。 | | 目标 ARN 无效 | 指定的目的地不存在。 | | 角色 ARN 无效 | IAM 角色不存在或缺乏权限。 | | 密钥检索失败 | 密钥或配置的密钥secretKey不存在，或者规则操作角色无法检索或解密密密钥。 | | 缺少时间戳 | 有效载荷不包含密timestamp钥。负载中的每个对象都必须包含一个具有整数 Unix 纪元值的timestamp字段。 | | 时间戳值无效 | 时间戳值不是整数（例如，字符串、浮点数或 ISO-8601 日期）。该值必须是整数 Unix 纪元，单位由timestampUnit指定。 | | 有效载荷无效（无字段） | 删除保留密钥后，有效负载不包含线路协议的有效字段。 | | 字段键无效 | 字段键为空或以开头\_。 | | 客户机批次包含无效点 | JSON 数组中的一个或多个元素未通过行协议验证。 | | 标签 \+ 字段超过列限制 | 标签和字段键的组合超过了最大列数 (250)。 | | 连接失败 | Amazon IoT Core 无法连接到 InfluxDB 端点。 | | 身份验证失败 | InfluxDB 令牌无效或已过期。在中更新密钥 Amazon Secrets Manager。 | | 未找到资源 | InfluxDB 中不存在指定的数据库、表或组织。 | | 字段类型冲突 | 一个或多个字段与现有架构冲突。整个批量写入失败。 | | InfluxDB 服务器错误 | InfluxDB 中出现内部错误。 | | InfluxDB 服务不可用 | InfluxDB 暂时不可用。规则引擎以指数回退方式重试。 | **重要** 批处理中任何一点的字段类型冲突都会导致整个批量写入失败。InfluxDB不会部分提交积分——要么所有点都成功要么整个写入都被拒绝。 重试可重试错误 (503) 时采用指数回退。对于 HTTP 401 响应，请从中 Amazon IoT Core 重新加载令牌 Amazon Secrets Manager 并重试请求一次。 Non-retryable 错误（404、422）会立即触发错误操作。有关重试限制，请参阅[Amazon IoT Core 服务配额](https://docs.amazonaws.cn/general/latest/gr/iot-core.html#limits_iot)。 ## 示例 ### InfluxDB 规则操作 ``` { "topicRulePayload": { "sql": "SELECT * FROM 'devices/+/telemetry'", "ruleDisabled": false, "awsIotSqlVersion": "2016-03-23", "actions": [ { "influxDB": { "destinationArn": "arn:aws:iot:us-west-2:111122223333:ruledestination/influxdb/a1b2c3d4", "roleArn": "arn:aws:iam::111122223333:role/iot-influxdb-role", "organization": "my-org", "databaseName": "sensor_data", "tableName": "device_metrics", "tags": { "device_id": "${clientid()}", "location": "building-a" }, "timestampUnit": "ms" } } ] } } ``` **有效载荷示例：** ``` { "timestamp": 1700000000000, "temperature": 23.5, "humidity": 60.1, "pressure": 1013.25, "battery_level": 87 } ``` **生成的线路协议：** ``` device_metrics,device_id=myDevice123,location=building-a temperature=23.5,humidity=60.1,pressure=1013.25,battery_level=87i 1700000000000 ``` 输出中的字段顺序可能会有所不同 — 字段不是按字母顺序排序的。 ### Client-batched 带有每个元素模板的数组有效负载 ``` { "influxDB": { "destinationArn": "arn:aws:iot:us-west-2:111122223333:ruledestination/influxdb/a1b2c3d4", "roleArn": "arn:aws:iam::111122223333:role/iot-influxdb-role", "organization": "my-org", "databaseName": "sensor_data", "tableName": "@{measurement_type}", "tags": { "sensor_id": "@{sensor_id}", "location": "${topic(2)}" }, "timestampUnit": "ns" } } ``` **示例有效载荷（发布至`devices/floor3/telemetry`）：** ``` [ {"measurement_type": "temperature", "sensor_id": "sensor-42", "timestamp": 1700000000000000000, "value": 23.5}, {"measurement_type": "humidity", "sensor_id": "sensor-42", "timestamp": 1700000001000000000, "value": 60.1}, {"measurement_type": "pressure", "sensor_id": "sensor-43", "timestamp": 1700000002000000000, "value": 1013.25} ] ``` **生成的线路协议：** ``` temperature,location=floor3,sensor_id=sensor-42 value=23.5 1700000000000000000 humidity,location=floor3,sensor_id=sensor-42 value=60.1 1700000001000000000 pressure,location=floor3,sensor_id=sensor-43 value=1013.25 1700000002000000000 ``` ### Server-side 使用 InfluxDB 进行批处理 要将服务器端批处理添加到任何 InfluxDB 操作中，请在操作配置`batchConfig`中包含： ``` "batchConfig": { "maxBatchSize": 50, "maxBatchOpenMs": 1000, "maxBatchSizeBytes": 65536, "batchAcrossTopics": false } ``` ### 规则操作角色的 IAM 政策 **信任政策：** ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "iot.amazonaws.com"}, "Action": "sts:AssumeRole" } ] } ``` **权限策略**： ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "secretsmanager:GetSecretValue", "Resource": "arn:aws:secretsmanager:us-west-2:111122223333:secret:my-influxdb-secret-a1b2c3" } ] } ``` ### 组合的客户端和服务器端批处理（点重新排序） 当您同时启用客户端批处理（JSON 数组有效负载）和服务器端批处理（`batchConfig`）时，请注意，服务器端批处理可能会对客户端批处理负载中的点进行重新排序。规则引擎将来自多条传入消息的积分累积到单个服务器端批处理中。由于消息是从不同的设备或主题异步到达的，因此在最终写入时，在原始客户端负载中排序的点可能会与其他消息的点交错。 操作配置： ``` { "influxDB": { "destinationArn": "arn:aws:iot:us-west-2:111122223333:ruledestination/influxdb/abc123", "roleArn": "arn:aws:iam::111122223333:role/iot-influxdb-role", "databaseName": "sensor_data", "tableName": "@{measurement_type}", "tags": { "device_id": "${topic(2)}", "floor": "@{floor}" }, "timestampUnit": "ms", "batchConfig": { "maxBatchSize": 100, "maxBatchOpenMs": 500, "maxBatchSizeBytes": 65536, "batchAcrossTopics": true } } } ``` 设备 A 发布到时间 `devices/deviceA/telemetry` T： ``` [ {"measurement_type": "temperature", "floor": "1", "timestamp": 1700000000000, "value": 22.1}, {"measurement_type": "temperature", "floor": "2", "timestamp": 1700000000100, "value": 23.4}, {"measurement_type": "humidity", "floor": "1", "timestamp": 1700000000200, "value": 55.0} ] ``` 设备 B 在 T\+10m `devices/deviceB/telemetry` s 时间发布到： ``` [ {"measurement_type": "temperature", "floor": "3", "timestamp": 1700000000050, "value": 21.8}, {"measurement_type": "humidity", "floor": "3", "timestamp": 1700000000150, "value": 62.3} ] ``` 这两条消息都在 500 毫秒的批处理窗口 (`maxBatchOpenMs`) 内到达，因此规则引擎将所有五个点合并为一个服务器端批处理。 生成的行协议（服务器端批量写入）： ``` temperature,device_id=deviceB,floor=3 value=21.8 1700000000050 temperature,device_id=deviceA,floor=1 value=22.1 1700000000000 temperature,device_id=deviceA,floor=2 value=23.4 1700000000100 humidity,device_id=deviceB,floor=3 value=62.3 1700000000150 humidity,device_id=deviceA,floor=1 value=55.0 1700000000200 ``` 请注意，积分不再按照它们在每个客户端有效载荷中出现的顺序排列。来自设备 A 的三个点（时间戳 1700000000000、1700000000100、1700000000200）与来自设备 B 的两个点（时间戳 170000000000050、1700000000150）交错。服务器端批处理不保证每条消息中的原始顺序。InfluxDB使用时间戳字段将每个点放在时间轴上，因此重新排序不会影响查询的正确性。但是，如果您的应用程序依赖写入顺序语义（例如，在同一毫秒内处理字段类型冲突或最后一次写入获胜的重复数据删除），请注意有效写入顺序可能与发布顺序不同。