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

InfluxDB

当您想要查询设备遥测以了解运营趋势或实时监控时间序列数据时，请选择此操作。您可以使用客户端或服务器端批处理将多个点合并到一个写入请求中。 AWS IoT Core 将每条消息转换为 InfluxDB 线路协议并将其写入指定的数据库和表。有关格式的更多信息，请参阅文档中的 InfluxDB 线路协议。 InfluxData 有关托管集群的信息，请参阅适用于 InfluxDB 的 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操作目标时，使用您提供的凭据对InfluxDB API进行身份验证，从而 AWS IoT Core 验证端点所有权：

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

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 替换中的数组有效负载 AWS IoT 规则。

二的语法 替换模板

使用 @{...} 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 记录内容

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

数据类型转换

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

命名限制

保留的密钥

线路协议转换期间，以下有效载荷键会从字段设置中删除：

错误操作

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

使用 InfluxDB 作为错误操作

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

错误操作输出

触发错误操作时，输出负载包含:ruleName、topic、cloudwatchTraceId、clientIdsourceIp、base64OriginalPayload（Base64-encoded 原始消息），以及每个条目都有failedAction、和failedResource、的failures数组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 发布到时间 devices/deviceB/telemetry t+10ms：

[
  {"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 的三个点（时间戳 170000000000000、1700000000100、1700000000200）与设备 B 中的两个点（时间戳 1700000000050、1700000000150）交错排列。服务器端批处理不能保证每条消息中的原始顺序。InfluxDB 使用时间戳字段将每个点放在时间轴上，因此重新排序不会影响查询的正确性。但是，如果您的应用程序依赖写入顺序语义（例如，在同一毫秒内处理字段类型冲突或最后一次写入获胜的重复数据删除），请注意有效的写入顺序可能与发布顺序不同。