Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

InfluxDB

Choisissez cette action lorsque vous souhaitez interroger la télémétrie des appareils pour connaître les tendances opérationnelles ou surveiller les données chronologiques en temps réel. Vous pouvez utiliser le traitement par lots côté client ou côté serveur pour combiner plusieurs points dans une seule demande d'écriture. AWS IoT Core convertit chaque message en protocole de ligne InfluxDB et l'écrit dans la base de données et la table spécifiées. Pour plus d'informations sur le format, consultez le protocole de ligne InfluxDB dans la InfluxData documentation. Pour plus d'informations sur les clusters gérés, consultez Amazon Timestream pour InfluxDB.

Conditions préalables

Cette action de règle comporte les conditions préalables suivantes :

Destination d’action InfluxDB

Avant de pouvoir utiliser l'action de règle InfluxDB, vous devez créer une destination d'action InfluxDB. La destination définit les paramètres de connexion pour votre instance InfluxDB. AWS IoT Core valide ensuite la propriété du terminal.

Création d'une destination

Utilisez l'CreateTopicRuleDestinationAPI pour créer une destination 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"
    }
  }
}

Paramètres de destination

Validation de propriété des terminaux

Lorsque vous créez une destination d'action InfluxDB, vous AWS IoT Core validez la propriété du point de terminaison en vous authentifiant auprès de l'API InfluxDB à l'aide des informations d'identification que vous avez fournies :

Une réponse réussie (2xx) définit le statut de destination surENABLED. En cas d'échec, AWS IoT Core définit le statut surERROR. Pour réessayer la validation, appelez UpdateTopicRuleDestination avec le statut défini sur. IN_PROGRESS

Cartographie terminologique InfluxDB

Les termes suivants diffèrent entre InfluxDB V2 et V3.

Parameters

Lorsque vous créez une AWS IoT règle avec l'action InfluxDB, vous devez spécifier les informations suivantes :

Traitement par lot

L'action InfluxDB prend en charge deux modes de traitement par lots.

Client-side mise en lots

Votre appareil IoT regroupe les données chronologiques sous forme de tableau JSON et les publie sous forme de message MQTT unique. Avec le traitement par lots côté client, chaque élément du tableau devient un point de protocole de ligne dans une seule demande d'écriture. Vous n'avez pas besoin de configuration supplémentaire.

Server-side mise en lots

Utilisez le traitement par lots côté serveur pour regrouper les messages individuels avant de les écrire dans InfluxDB. Configurez le traitement par lots côté serveur à l'aide du paramètre. batchConfig Le lot est vidé lorsqu'une limite configurée (maxBatchSizemaxBatchOpenMs, oumaxBatchSizeBytes) est atteinte pour la première fois.

L'action InfluxDB est mesurée en fonction de la taille de la charge utile sortante par incréments de 5 KiB. Line-protocol les échecs de conversion sont également mesurés.

Per-element modèles pour les charges utiles des réseaux

Lorsque vos appareils IoT envoient des données de séries chronologiques par lots sous forme de tableau JSON, vous pouvez utiliser les « modèles de substitution par élément » pour résoudre les valeurs de chaque élément de tableau individuel. Cela permet d'acheminer chaque point de données vers une table différente ou d'appliquer des balises spécifiques à un élément. Pour plus d’informations, consultez Per-element substitution des charges utiles du réseau dans AWS IoT rules.

Syntaxe de deux modèles de substitution

Utilisez @{...} les valeurs in tableName et tag pour résoudre l'expression par rapport à chaque élément du tableau individuellement.

Exemple

Compte tenu de la charge utile de l'appareil suivante (tableau JSON) :

[
  {"measurement_type": "temperature", "room": "kitchen", "timestamp": 1700000000000, "value": 23.5},
  {"measurement_type": "humidity", "room": "bedroom", "timestamp": 1700000001000, "value": 60.1}
]

Et la configuration d'action suivante :

{
  "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"
  }
}

La sortie du protocole de ligne qui en résulte contient deux points :

temperature,room=kitchen value=23.5 1700000000000
humidity,room=bedroom value=60.1 1700000001000

Restrictions

Contenu de l'enregistrement InfluxDB

Pour chaque enregistrement du résultat de la requête post-SQL, le point de protocole de ligne InfluxDB qui en résulte contient les composants suivants :

Conversion du type de données

AWS IoT Core convertit les valeurs JSON en types de protocole de ligne InfluxDB comme suit :

Restrictions de dénomination

Clés réservées

Les clés de charge utile suivantes sont supprimées du champ défini lors de la conversion par protocole de ligne :

Actions d'erreur

Si l'action InfluxDB échoue, l'action d'erreur configurée est déclenchée.

Utilisation d'InfluxDB comme action d'erreur

Vous pouvez configurer l'action InfluxDB comme action d'erreur pour n'importe quelle règle. L'intégralité de la charge utile est écrite dans un seul tableName enregistrement. Message-scope les modèles de substitution (${...}) sont pris en charge dans les actions d'erreur tableName ettags.

Sortie d'action en cas d'erreur

Lorsque l'action d'erreur est déclenchée, la charge utile de sortie contient : ruleNametopic,cloudwatchTraceId,clientId,sourceIp, base64OriginalPayload (message Base64-encoded d'origine) et un failures tableau où chaque entrée contientfailedAction,failedResource, eterrorMessage.

Dans le cas du traitement par lots côté serveur, la charge utile utilise payloadsWithMetadata une entrée pour chaque message MQTT entrant distinct. affectedIdsLes valeurs de chaque échec font référence aux id valeurs de ces entrées de message ; il ne s'agit pas d'indices de points. Un tableau JSON côté client est un message entrant même s'il produit plusieurs points InfluxDB. Pour le format complet de la charge utile, voirActions d'erreur liées au traitement par lots.

Les erreurs réessayables (503) sont réessayées avec un retard exponentiel. Pour une réponse HTTP 401, AWS IoT Core recharge le jeton depuis AWS Secrets Manager et réessaie la demande une fois. Non-retryable les erreurs (404, 422) déclenchent immédiatement l'action d'erreur. Pour connaître les limites de nouvelles tentatives, consultez la section Quotas AWS IoT Core de service.

Exemples

Action de règle 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"
        }
      }
    ]
  }
}

Exemple de charge utile :

{
  "timestamp": 1700000000000,
  "temperature": 23.5,
  "humidity": 60.1,
  "pressure": 1013.25,
  "battery_level": 87
}

Protocole de ligne résultant :

device_metrics,device_id=myDevice123,location=building-a temperature=23.5,humidity=60.1,pressure=1013.25,battery_level=87i 1700000000000

L'ordre des champs dans la sortie peut varier : les champs ne sont pas triés par ordre alphabétique.

Client-batched charge utile du tableau avec modèles par élément

{
  "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"
  }
}

Exemple de charge utile (publiée surdevices/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}
]

Protocole de ligne résultant :

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 traitement par lots avec InfluxDB

Pour ajouter le traitement par lots côté serveur à n'importe quelle action InfluxDB, incluez batchConfig dans la configuration de l'action :

"batchConfig": {
  "maxBatchSize": 50,
  "maxBatchOpenMs": 1000,
  "maxBatchSizeBytes": 65536,
  "batchAcrossTopics": false
}

Politique IAM pour le rôle d'action des règles

Politique de confiance :

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {"Service": "iot.amazonaws.com"},
      "Action": "sts:AssumeRole"
    }
  ]
}

Politique d'autorisation :

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "secretsmanager:GetSecretValue",
      "Resource": "arn:aws:secretsmanager:us-west-2:111122223333:secret:my-influxdb-secret-a1b2c3"
    }
  ]
}

Traitement par lots combiné côté client et côté serveur (réorganisation par points)

Lorsque vous activez à la fois le traitement par lots côté client (charges utiles du tableau JSON) et le traitement par lots côté serveur (batchConfig), sachez que le lot côté serveur peut réorganiser les points à partir d'une charge utile par lots côté client. Le moteur de règles accumule les points provenant de plusieurs messages entrants en un seul lot côté serveur. Comme les messages arrivent de manière asynchrone à partir de différents appareils ou sujets, les points qui ont été ordonnés dans la charge utile du client d'origine peuvent être entrelacés avec des points provenant d'autres messages lors de l'écriture finale.

Configuration de l'action :

{
  "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
    }
  }
}

L'appareil A publie devices/deviceA/telemetry à l'instant 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}
]

L'appareil B publie à l'devices/deviceB/telemetryheure T+10 ms :

[
  {"measurement_type": "temperature", "floor": "3", "timestamp": 1700000000050, "value": 21.8},
  {"measurement_type": "humidity", "floor": "3", "timestamp": 1700000000150, "value": 62.3}
]

Les deux messages arrivent dans le délai de 500 ms (maxBatchOpenMs), de sorte que le moteur de règles combine les cinq points en un seul lot côté serveur.

Protocole de ligne résultant (écriture par lots côté serveur) :

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

Notez que les points ne sont plus dans l'ordre dans lequel ils apparaissaient dans la charge utile de chaque client. Les trois points du périphérique A (horodatages 1700000000000, 1700000000100, 1700000000200) sont entrelacés avec les deux points du périphérique B (horodatages 1700000000050, 1700000000150). Le lot côté serveur ne garantit pas la commande originale dans chaque message. InfluxDB utilise le champ d'horodatage pour placer chaque point sur la chronologie, de sorte que la réorganisation n'affecte pas l'exactitude des requêtes. Toutefois, si votre application s'appuie sur la sémantique de l'ordre d'écriture (par exemple, pour gérer les conflits de type de champ ou la déduplication des dernières écritures gagnantes dans la même milliseconde), sachez que l'ordre d'écriture effectif peut être différent de l'ordre de publication.