本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

InfluxDB

当您要查询设备遥测以了解运行趋势或实时监控时间序列数据时，请选择此操作。您可以使用客户端或服务器端批处理将多个点合并到一个写入请求中。 AWS IoT Core 将每条消息转换为InfluxDB行协议并将其写入指定的数据库和表。有关该格式的更多信息，请参阅文档中的 InfluxDB行协议。 InfluxData 有关托管集群的信息，请参阅 Inf luxDB 的 Amazon Timestream。

先决条件

此规则操作具有以下先决条件：

InfluxDB 操作目的地

在使用InfluxDB规则操作之前，必须先创建InfluxDB操作目标。目标定义了您的InfluxDB实例的连接参数。 AWS 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"
    }
  }
}

目的地参数

端点所有权验证

当您创建InfluxDB操作目标时，使用您提供的 AWS IoT Core 凭据对InfluxDB API进行身份验证来验证端点所有权：

成功响应 (2xx) 将目标状态设置为。ENABLED失败时， AWS IoT Core 将状态设置为ERROR。要重试验证，请在状态设置为IN_PROGRESS的情况下调用UpdateTopicRuleDestination。

InfluxDB 术语映射

以下术语在 InfluxDB V2 和 V3 之间有所不同。

参数

使用 InfluxDB 操作创建 AWS IoT 规则时，必须指定以下信息：

批处理

InfluxDB 操作支持两种批处理模式。

Client-side 批处理

您的物联网设备将时间序列数据批处理为 JSON 数组，并将其作为单个 MQTT 消息发布。通过客户端批处理，每个数组元素在单个写入请求中成为一个线路协议点。您不需要额外的配置。

Server-side 批处理

在将单个消息写入InfluxDB之前，使用服务器端批处理对它们进行分组。使用参数配置服务器端批处理。batchConfig当首先达到任何配置的限制（maxBatchSize、或maxBatchSizeBytes）时maxBatchOpenMs，将刷新该批次。

InfluxDB 操作是根据出站负载大小进行计量的，以 5 KiB 为增量。 Line-protocol 还会计量转换失败次数。

Per-element 阵列有效载荷的模板

当您的物联网设备以 JSON 数组的形式发送批处理的时间序列数据时，您可以使用 “每元素替换模板” 来解析每个单独数组元素的值。这会将每个数据点路由到不同的表或应用特定元素的标签。有关更多信息，请参阅Per-element模板。

二的语法 替代模板

使用 @{...} 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

限制

InfluxDB 记录内容

对于后SQL查询结果中的每条记录，生成的InfluxDB行协议点包含以下组件：

数据类型转换

AWS IoT Core 将 JSON 值转换为 InfluxDB 线路协议类型，如下所示：

命名限制

保留的密钥

在线路协议转换期间，将从设置的字段中删除以下有效负载密钥：

错误操作

如果InfluxDB操作失败，则会触发配置的错误操作。

使用 InfluxDB 作为错误操作

您可以将InfluxDB操作配置为任何规则的错误操作。整个有效载荷被写入到单tableName一记录中。 Message-scope 错误操作的tableName和支持替换模板 (${...}) tags。

错误操作输出

触发错误操作时，输出负载包含:ruleName、、topic、cloudwatchTraceId、clientIdsourceIp、base64OriginalPayload（Base64-encoded 原始消息）和一个failures数组，其中每个条目都有failedActionfailedResource、和errorMessage。

在服务器端批处理中，负载使用payloadsWithMetadata，每条不同的入站 MQTT 消息都有一个条目。每个失败的affectedIds值都指的是这些消息条目的id值；它们不是积分索引。即使客户端的JSON数组生成多个InfluxDB点，它也是一条入站消息。有关完整的有效负载格式，请参阅批处理的错误操作。

重试可重试错误 (503) 时采用指数回退。对于 HTTP 401 响应，请从中 AWS IoT Core 重新加载令牌 AWS Secrets Manager 并重试请求一次。 Non-retryable 错误（404、422）会立即触发错误操作。有关重试限制，请参阅AWS IoT Core 服务配额。

示例

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使用时间戳字段将每个点放在时间轴上，因此重新排序不会影响查询的正确性。但是，如果您的应用程序依赖写入顺序语义（例如，在同一毫秒内处理字段类型冲突或最后一次写入获胜的重复数据删除），请注意有效写入顺序可能与发布顺序不同。