使用 AWS IoT Core Rule Action 将遥测数据直接写入 InfluxDB

摘要：IoT Core Rule Action 支持 InfluxDB 写入，通过 Basic Ingest 节约消息和写入成本

01 一、引言

02 二、使用数组元素模板将数据写入不同的 measurement

03 三、配置环境变量

04 四、数据映射与类型转换

05 五、批处理配置

06 六、成本考量

07 七、清理资源

08 八、参考

一、引言

2026 年 8 月 25 日，AWS IoT Core 发布了 InfluxDB rule action。该功能允许 AWS IoT Core 规则引擎将设备上报的 JSON 遥测数据转换为 InfluxDB line protocol，并直接写入 InfluxDB，无需通过 Lambda 函数实现数据格式转换和写入逻辑。

传统方案：设备 → AWS IoT Core → Lambda（转换数据格式并写入）→ InfluxDB
使用 InfluxDB 规则操作后：设备 → AWS IoT Core（InfluxDB rule action）→ InfluxDB

本文演示如何使用 AWS CLI 和 curl 配置该数据链路，并验证数据写入结果。

二、使用数组元素模板将数据写入不同的 measurement

设备可以将多个指标封装在同一个 JSON 数组中上报。InfluxDB 规则操作支持在 tableName 和 tag 值中使用 @{…} 模板，按数组中的每个元素解析字段值，从而将不同指标写入对应的 measurement。

以下规则配置将 metric 字段用作 measurement 名称，并从每个数组元素中提取 sensor_id 作为 tag。device_id 则通过 MQTT topic 提取。

{
  "influxDB": {
    "destinationArn": "...",
    "roleArn": "...",
    "databaseName": "telemetry",
    "organization": "demo-org",
    "tableName": "@{metric}",
    "tags": { "sensor_id": "@{sensor_id}", "device_id": "${topic(2)}" },
    "timestampUnit": "ms"
  }
}

例如，设备可以通过一条消息上报以下数组：

[
  {"metric": "temperature", "sensor_id": "sensor-42", "timestamp": 1791525500000, "value": 23.5},
  {"metric": "humidity",    "sensor_id": "sensor-42", "timestamp": 1791525501000, "value": 60.1},
  {"metric": "pressure",    "sensor_id": "sensor-43", "timestamp": 1791525502000, "value": 1013.25}
]

写入完成后，可以在 InfluxDB 中查询到对应的三个 measurement：

--- temperature
_time,_value,_field,_measurement,device_id,sensor_id
2026-10-09T05:58:20Z,23.5,value,temperature,dev-002,sensor-42

--- humidity
2026-10-09T05:58:21Z,60.1,value,humidity,dev-002,sensor-42

--- pressure
2026-10-09T05:58:22Z,1013.25,value,pressure,dev-002,sensor-43

${…} 与 @{…} 的行为不同，主要区别如下：

	${expression}	@{expression}
求值对象	整条消息	数组的单个元素
求值次数	每条消息一次	每个元素一次
可用位置	各处（含 SQL）	仅 tableName 和 tag 值

使用 @{…} 时需要注意以下限制：表达式只能引用字段，不支持函数；每个值最多包含一个模板标记；同一个值中不能混用 @{…} 和 ${…}。如果输入是单个 JSON 对象，规则引擎会将其视为只包含一个元素的数组。被 @{…} 引用的字段不会再写入 field set，因此本例每个 measurement 中仅包含 value 这个 field。

三、配置环境变量

export REGION=us-west-2
export DB_NAME=iot-influx-demo
export INFLUX_ORG=demo-org
export INFLUX_BUCKET=telemetry
export INFLUX_USER=admin
export RULE_NAME=iotinflux_demo_rule
export SECRET_NAME=iotinflux-demo/influx-token
export ROLE_NAME=iotinflux-demo-rule-role
export ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)

3.1 创建 InfluxDB 实例

本示例使用的 InfluxDB 规则操作通过公网端点连接 InfluxDB，不提供 VPC 内部连接路径，因此实例需要启用公网可访问性。支持的端口为 443、8443、8086（InfluxDB V2 默认端口）和 8181（InfluxDB V3 默认端口），不支持明文 HTTP。

VPC_ID=$(aws ec2 describe-vpcs --filters Name=isDefault,Values=true \
  --query 'Vpcs[0].VpcId' --output text)

SG_ID=$(aws ec2 create-security-group --group-name iotinflux-demo-sg \
  --description "PoC: IoT Core InfluxDB rule action" \
  --vpc-id "$VPC_ID" --query GroupId --output text)

aws ec2 authorize-security-group-ingress --group-id "$SG_ID" \
  --ip-permissions 'IpProtocol=tcp,FromPort=8086,ToPort=8086,IpRanges=[{CidrIp=0.0.0.0/0}]'

实例密码长度需为 8–64 个字符，且仅允许使用 [a-zA-Z0-9]：

SUBNET_ID=$(aws ec2 describe-subnets --filters Name=vpc-id,Values=$VPC_ID \
  Name=map-public-ip-on-launch,Values=true --query 'Subnets[0].SubnetId' --output text)

INFLUX_PASSWORD=$(LC_ALL=C tr -dc 'A-Za-z0-9' </dev/urandom | head -c 24)

aws timestream-influxdb create-db-instance \
  --name "$DB_NAME" --username "$INFLUX_USER" --password "$INFLUX_PASSWORD" \
  --organization "$INFLUX_ORG" --bucket "$INFLUX_BUCKET" \
  --db-instance-type db.influx.medium \
  --db-storage-type InfluxIOIncludedT1 --allocated-storage 20 \
  --deployment-type SINGLE_AZ \
  --vpc-subnet-ids "$SUBNET_ID" --vpc-security-group-ids "$SG_ID" \
  --publicly-accessible --port 8086 --network-type IPV4

实例创建需要一定时间。本示例通过轮询实例状态，直到状态变为 AVAILABLE：

DB_ID=$(aws timestream-influxdb list-db-instances \
  --query "items[?name=='$DB_NAME'].id | [0]" --output text)

while true; do
  S=$(aws timestream-influxdb get-db-instance --identifier "$DB_ID" --query status --output text)
  echo "$(date +%T) $S"; [ "$S" = AVAILABLE ] && break; sleep 30
done

INFLUX_URL="https://$(aws timestream-influxdb get-db-instance --identifier "$DB_ID" \
  --query endpoint --output text):8086"

3.2 创建 InfluxDB API token

实例创建过程中生成的 secret（influxAuthParametersSecretArn）包含 bucket、organization、password 和 username，但不包含 API token。对于 InfluxDB V2，需要使用实例凭证登录后创建 token：

COOKIE=$(mktemp)
curl -sS -f -c "$COOKIE" -X POST "$INFLUX_URL/api/v2/signin" \
  -u "$INFLUX_USER:$INFLUX_PASSWORD" > /dev/null

ORG_ID=$(curl -sS -f -b "$COOKIE" "$INFLUX_URL/api/v2/orgs?org=$INFLUX_ORG" | jq -r '.orgs[0].id')
BUCKET_ID=$(curl -sS -f -b "$COOKIE" "$INFLUX_URL/api/v2/buckets?name=$INFLUX_BUCKET" | jq -r '.buckets[0].id')
USER_ID=$(curl -sS -f -b "$COOKIE" "$INFLUX_URL/api/v2/me" | jq -r '.id')

TOKEN=$(curl -sS -f -b "$COOKIE" -X POST "$INFLUX_URL/api/v2/authorizations" \
  -H 'Content-Type: application/json' \
  -d "$(jq -n --arg org "$ORG_ID" --arg bucket "$BUCKET_ID" --arg uid "$USER_ID" '{
        orgID: $org,
        description: "AWS IoT Core InfluxDB rule action",
        permissions: [
          {action: "write", resource: {type: "buckets", id: $bucket, orgID: $org}},
          {action: "read",  resource: {type: "buckets", id: $bucket, orgID: $org}},
          {action: "read",  resource: {type: "users",   id: $uid}}
        ]
      }')" | jq -r '.token')

上述 token 包含以下权限：

write buckets/<bucket> — 规则引擎写入
read buckets/<bucket> — 本文最后用同一个 token 查数据
read users/<self> — IoT Core 校验端点所有权时调 GET /api/v2/me，该接口需要此权限

将 token 以纯字符串形式存入 AWS Secrets Manager。默认情况下，destination 会将整个 secret 值作为 token 使用；如果 secret 存储为 JSON，则需要额外配置 secretKey：

SECRET_ARN=$(aws secretsmanager create-secret --name "$SECRET_NAME" \
  --secret-string="$TOKEN" --query ARN --output text)

InfluxDB token 使用 base64url 编码，生成的值可能以 – 开头。使用 AWS CLI 时，采用 –secret-string=”$TOKEN” 的参数形式，避免该值被解析为选项。

3.3 创建 IAM 角色和策略

InfluxDB API token 用于向数据库进行身份验证。AWS IoT Core 规则引擎承担的 IAM 角色只需获得读取对应 Secrets Manager secret 的权限：

aws iam create-role --role-name "$ROLE_NAME" --assume-role-policy-document "$(jq -n \
  --arg acct "$ACCOUNT_ID" --arg arn "arn:aws:iot:$REGION:${ACCOUNT_ID}:rule/*" '{
    Version: "2012-10-17",
    Statement: [{
      Effect: "Allow",
      Principal: {Service: "iot.amazonaws.com"},
      Action: "sts:AssumeRole",
      Condition: {
        StringEquals: {"aws:SourceAccount": $acct},
        ArnLike: {"aws:SourceArn": $arn}
      }
    }]
  }')"

aws iam put-role-policy --role-name "$ROLE_NAME" --policy-name read-influx-token \
  --policy-document "$(jq -n --arg s "arn:aws:secretsmanager:$REGION:${ACCOUNT_ID}:secret:$SECRET_NAME-*" '{
    Version: "2012-10-17",
    Statement: [{Effect: "Allow", Action: "secretsmanager:GetSecretValue", Resource: $s}]
  }')"

ROLE_ARN=$(aws iam get-role --role-name "$ROLE_NAME" --query Role.Arn --output text)

3.4 创建规则操作 destination

InfluxDB 连接端点和 secret 引用配置在 topic rule destination 中，规则操作通过 destination ARN 引用该配置：

ENDPOINT=$(aws timestream-influxdb get-db-instance --identifier "$DB_ID" \
  --query endpoint --output text)

DEST_ARN=$(aws iot create-topic-rule-destination \
  --destination-configuration "$(jq -n --arg ep "https://$ENDPOINT:8086" --arg sid "$SECRET_ARN" '{
      influxDBConfiguration: {
        endpoint: $ep, influxDBVersion: "V2", secretId: $sid, secretType: "SecretString"
      }
    }')" \
  --query 'topicRuleDestination.arn' --output text)

创建 destination 后，AWS IoT Core 会验证端点连接。对于 V2，验证请求为 GET /api/v2/me；对于 V3，则为 GET /api/v3/configure/database。端点返回 2xx 状态码后，destination 状态会变为 ENABLED：

until [ "$(aws iot get-topic-rule-destination --arn "$DEST_ARN" \
  --query topicRuleDestination.status --output text)" = ENABLED ]; do sleep 5; done

如果 destination 状态变为 ERROR，且 statusReason 为 Failed to validate destination connection，请检查安全组、端点和 token 配置。修复问题后，可以通过以下命令重新触发验证：

aws iot update-topic-rule-destination --arn "$DEST_ARN" --status IN_PROGRESS

3.5 创建 IoT Core 规则

aws iot create-topic-rule --rule-name "$RULE_NAME" --topic-rule-payload "$(jq -n \
  --arg sql "SELECT * FROM 'devices/+/batch'" \
  --arg dest "$DEST_ARN" --arg role "$ROLE_ARN" \
  --arg bucket "$INFLUX_BUCKET" --arg org "$INFLUX_ORG" '{
    sql: $sql,
    awsIotSqlVersion: "2016-03-23",
    ruleDisabled: false,
    actions: [{
      influxDB: {
        destinationArn: $dest,
        roleArn: $role,
        databaseName: $bucket,
        organization: $org,
        tableName: "@{metric}",
        tags: {sensor_id: "@{sensor_id}", device_id: "${topic(2)}"},
        timestampUnit: "ms"
      }
    }],
    errorAction: {
      cloudwatchLogs: {roleArn: $role, logGroupName: "/aws/iot/iotinflux-demo-errors", batchMode: false}
    }
  }')"

主要参数说明：

参数	说明
databaseName	V2 的 bucket，V3 的 database。不支持替换模板
tableName	V2 的 measurement，V3 的 table。支持替换模板
organization	V2 必填，V3 忽略
tags	line protocol 的 tag，会被索引
timestampUnit	s/ms/us/ns，默认 ms

配置 errorAction 后，规则操作执行失败时，错误消息会包含 ruleName、topic、sourceIp、base64OriginalPayload，以及 failures 数组。数组中的字段包括 failedAction、failedResource 和 errorMessage，可用于定位失败原因。

3.6 发布测试消息

本示例使用 aws iot-data publish 通过 HTTPS 数据平面发布消息，并由 AWS SigV4 进行身份验证，因此测试时无需配置设备证书或使用设备 SDK。topic 采用 Basic Ingest 格式 $aws/rules/<ruleName>/<后缀>，消息会直接进入指定规则，而不经过 MQTT 发布/订阅 broker。

IOT_EP="https://$(aws iot describe-endpoint --endpoint-type iot:Data-ATS \
  --query endpointAddress --output text)"
NOW_MS=$(( $(date +%s) * 1000 ))

aws iot-data publish --endpoint-url "$IOT_EP" \
  --topic "\$aws/rules/$RULE_NAME/devices/dev-002/batch" \
  --qos 1 --cli-binary-format raw-in-base64-out \
  --payload "$(jq -n --argjson ts "$NOW_MS" '[
    {metric: "temperature", sensor_id: "sensor-42", timestamp: $ts,          value: 23.5},
    {metric: "humidity",    sensor_id: "sensor-42", timestamp: ($ts + 1000), value: 60.1},
    {metric: "pressure",    sensor_id: "sensor-43", timestamp: ($ts + 2000), value: 1013.25}
  ]')"

payload 必须含 timestamp，整数 Unix epoch，key 名区分大小写，ts/time 不可替代。IoT Core 不生成时间戳。
{topic(2)} 取到 dev-002。这三级也不计入 topic 的 8 段和 256 字符限制。
纯 Basic Ingest 的规则，SQL 的 FROM 子句可省略。
Basic Ingest topic 不能被订阅。
QoS 1 的 PUBACK 只表示消息到达规则引擎，不代表 action 成功。
payload 只支持 JSON；二进制或 Protobuf 要先在 SQL 里 decode()。

3.7 查询写入结果

query() {
  curl -sS -f -X POST "$INFLUX_URL/api/v2/query?org=$INFLUX_ORG" \
    -H "Authorization: Token $TOKEN" \
    -H 'Content-Type: application/vnd.flux' \
    -H 'Accept: application/csv' --data-binary "$1"
}

query "
import \"influxdata/influxdb/schema\"
schema.measurements(bucket: \"$INFLUX_BUCKET\")
" | cut -d, -f4 | tail -n +2

humidity
pressure
temperature

for m in temperature humidity pressure; do
  echo "--- $m"
  query "
from(bucket: \"$INFLUX_BUCKET\")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == \"$m\")
  |> keep(columns: [\"_time\",\"_measurement\",\"_field\",\"_value\",\"device_id\",\"sensor_id\"])
" | cut -d, -f4-
done

查询结果应包含本文前面展示的三个 measurement 及对应的遥测数据。

四、数据映射与类型转换

line protocol 组成	来源
measurement	tableName
tags	tags 参数
fields	payload 里剩下的属性（排掉 tag、tableName 引用的 key、timestamp）
timestamp	payload 的 timestamp

五、批处理配置

客户端批处理：设备将多个数据点封装为一个 JSON 数组并一次性发送。数组中的每个元素对应一个 point，规则操作会将其合并到一次写入请求中，无需额外配置。本文示例采用这种方式。

服务端批处理：规则引擎可以先累积多条消息，再批量写入 InfluxDB。通过 batchConfig 配置批处理参数：

"batchConfig": {
  "maxBatchSize": 100,
  "maxBatchOpenMs": 500,
  "maxBatchSizeBytes": 65536,
  "batchAcrossTopics": false
}

三个参数的取值范围依次为 1–500、5–1000 和 100–131072。当任一批处理条件先达到时，规则引擎就会提交当前批次。客户端批处理和服务端批处理可以同时启用。

六、成本考量

按照当前 us-west-2 的真实费率（2026-10）：

计费项	单价
IoT Core messaging	$1.00 / 百万条（前 10 亿条/月）
IoT Core rules triggered	$0.15 / 百万次
IoT Core actions executed	$0.15 / 百万次
Lambda 请求	$0.20 / 百万次
Lambda 时长	$0.0000133334 / GB-秒

1 亿请求四种实现方式的月度成本

方案	messaging	rules	actions	Lambda	合计
Topic -> 规则 -> Lambda 写入	$100	$15	$15	$23.3	$153.3
Topic -> 规则 -> influxDB action	$100	$15	$15	–	$130
Basic Ingest -> 规则 -> influxDB action	–	$15	$15	–	$30
Basic Ingest -> 规则 -> influxDB action （服务端批处理 100/批）	–	$15	$0.3	–	$15.3

Lambda 一列按 128 MB 内存、单次执行 20 ms 估算：0.125 GB × 0.02 s × 10 亿 = 25 万 GB-秒 = $3.3，加上 1 亿次请求的 $20。

influxDB action 批处理 100/批合计 100 万次 action 写入，每批 10 KB（2 个 5 KB 增量），200 万个计费单位 $0.3。

七、清理资源

aws iot delete-topic-rule --rule-name "$RULE_NAME"
aws iot delete-topic-rule-destination --arn "$DEST_ARN"
aws timestream-influxdb delete-db-instance --identifier "$DB_ID"
aws secretsmanager delete-secret --secret-id "$SECRET_NAME" --force-delete-without-recovery
aws iam delete-role-policy --role-name "$ROLE_NAME" --policy-name read-influx-token
aws iam delete-role --role-name "$ROLE_NAME"
aws logs delete-log-group --log-group-name /aws/iot/iotinflux-demo-errors
until aws ec2 delete-security-group --group-id "$SG_ID" 2>/dev/null; do sleep 30; done

八、参考

➡️ 下一步行动：

相关产品：

Amazon IoT Core — 设备连接到云
AWS Lambda — 无需服务器即可运行代码
Amazon VPC — 隔离云网络
Amazon Secrets Manager — 密钥管理
Amazon Batch — 完全托管式批处理

相关文章：

立即咨询 → 从 AI 规划到落地实施，我们的专家团队为你全程护航。

*前述特定亚马逊云科技生成式人工智能相关的服务目前在亚马逊云科技海外区域可用。亚马逊云科技中国区域相关云服务由西云数据和光环新网运营，具体信息以中国区域官网为准。

本篇作者

AWS 架构师中心：云端创新的引领者

探索 AWS 架构师中心，获取经实战验证的最佳实践与架构指南，助您高效构建安全、可靠的云上应用

亚马逊AWS官方博客