翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

InfluxDB

デバイステレメトリに運用傾向をクエリしたり、時系列データをリアルタイムでモニタリングしたりする場合は、このアクションを選択します。クライアント側またはサーバー側のバッチ処理を使用して、1 つの書き込みリクエストに複数のポイントを組み合わせることができます。 は各メッセージを InfluxDB ラインプロトコル AWS IoT Core に変換し、指定されたデータベースとテーブルに書き込みます。形式の詳細については、InfluxDB ラインプロトコル」を参照してください。 InfluxData マネージドクラスターの詳細については、「Amazon Timestream for InfluxDB」を参照してください。

前提条件

このルールアクションには、次の前提条件があります。

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 では、以下の用語が異なります。 V2 V3

パラメータ

InfluxDB アクションを使用して AWS IoT ルールを作成するときは、次の情報を指定する必要があります。

バッチ処理

InfluxDB アクションは、2 つのバッチ処理モードをサポートしています。

クライアント側のバッチ処理

IoT デバイスは時系列データを JSON 配列としてバッチ処理し、単一の MQTT メッセージとして発行します。クライアント側のバッチ処理では、各配列要素は 1 回の書き込みリクエストで 1 つのラインプロトコルポイントになります。追加の設定は必要ありません。

サーバー側のバッチ処理

InfluxDB に書き込む前に、サーバー側のバッチ処理を使用して個々のメッセージをグループ化します。batchConfig パラメータを使用してサーバー側のバッチ処理を設定します。バッチは、設定された制限 (maxBatchSize、maxBatchOpenMs、または maxBatchSizeBytes) が最初に達するとフラッシュされます。

InfluxDB アクションは、アウトバウンドペイロードサイズに基づいて 5 KiB 単位で計測されます。ラインプロトコル変換の失敗も計測されます。

配列ペイロードの要素ごとのテンプレート

IoT デバイスがバッチ処理された時系列データを JSON 配列として送信する場合、「要素ごとの置換テンプレート」を使用して、個々の配列要素から値を解決できます。これにより、各データポイントが異なるテーブルにルーティングされるか、要素固有のタグが適用されます。詳細については、「要素ごとのテンプレート」を参照してください。

2 つの置換テンプレートの構文

tableName および タグ値@{...}で を使用して、式を各配列要素に対して個別に解決します。

例

次のデバイスペイロード (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"
  }
}

結果のラインプロトコル出力には、次の 2 つのポイントが含まれます。

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 アクションは、任意のルールのエラーアクションとして設定できます。ペイロード全体が 1 つのレコードtableNameとして 1 つの に書き込まれます。メッセージスコープ置換テンプレート (${...}) は、エラーアクションの tableNameおよび でサポートされていますtags。

エラーアクションの出力

エラーアクションがトリガーされると、出力ペイロードには、ruleName、、topic、cloudwatchTraceIdclientIdsourceIp、 base64OriginalPayload (Base64 でエンコードされた元のメッセージ）、および各エントリに failedAction、failedResource、および があるfailures配列が含まれますerrorMessage。

サーバー側のバッチ処理では、ペイロードは を使用しpayloadsWithMetadata、個別のインバウンド MQTT メッセージごとに 1 つのエントリを使用します。各失敗のaffectedIds値は、それらのメッセージエントリidの値を参照します。これらはポイントインデックスではありません。クライアント側の JSON 配列は、複数の InfluxDB ポイントを生成しても、1 つのインバウンドメッセージです。完全なペイロード形式については、「」を参照してくださいバッチ処理のエラーアクション。

再試行可能なエラー (503) はエクスポネンシャルバックオフで再試行されます。HTTP 401 レスポンスの場合、 は からトークン AWS IoT Core を再ロード AWS Secrets Manager し、リクエストを 1 回再試行します。再試行不可能なエラー (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

出力のフィールドの順序は異なる場合があります。フィールドはアルファベット順にソートされません。

エレメントごとのテンプレートを使用したクライアントバッチ配列ペイロード

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

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) の両方を有効にする場合、サーバー側のバッチはクライアント側のバッチ処理されたペイロードからポイントの順序を変更する場合があることに注意してください。ルールエンジンは、複数の受信メッセージからのポイントを 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": {
      "device_id": "${topic(2)}",
      "floor": "@{floor}"
    },
    "timestampUnit": "ms",
    "batchConfig": {
      "maxBatchSize": 100,
      "maxBatchOpenMs": 500,
      "maxBatchSizeBytes": 65536,
      "batchAcrossTopics": true
    }
  }
}

デバイス A は、時刻 T devices/deviceA/telemetryに に発行します。

[
  {"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+10ms devices/deviceB/telemetry の時点で に発行します。

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

どちらのメッセージも 500 ミリ秒のバッチウィンドウ (maxBatchOpenMs) 内に届くため、ルールエンジンは 5 つのポイントすべてを 1 つのサーバー側のバッチに結合します。

結果のラインプロトコル (サーバー側のバッチ書き込み):

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 の 3 つのポイント (タイムスタンプ 1700000000000、1700000000100、1700000000200) は、デバイス B の 2 つのポイント (タイムスタンプ 1700000000050、1700000000150) とインターリーブされます。サーバー側のバッチは、各メッセージ内の元の順序を保証するものではありません。InfluxDB はタイムスタンプフィールドを使用して各ポイントをタイムラインに配置するため、順序を変更してもクエリの正確性には影響しません。ただし、アプリケーションが書き込み順序セマンティクスに依存している場合 (たとえば、フィールドタイプの競合の処理や、同じミリ秒内のlast-write-winsの重複排除など）、有効な書き込み順序は発行順序とは異なる場合があることに注意してください。