Siemens Logo设备影子数据写入AWS IoT Timestream分列问题咨询

方案1：直接修改IoT Core规则SQL（适用测点固定场景） #

AWS IoT Core的SQL语法支持用括号索引法访问包含特殊字符的JSON属性，不需要修改设备上报的字段名。如果你的Siemens Logo测点数量固定，直接用UNION ALL把每个测点拆成独立行，适配Timestream的写入模型即可。

修改后的规则查询语句 #

SELECT 
  clientId() AS device_id,
  'AM..4:1-1' AS measure_name,
  state.reported["AM..4:1-1"] AS measure_value::varchar,
  metadata.reported["AM..4:1-1"].timestamp * 1000 AS time
FROM '$aws/things/HneviceIoT1/shadow/update'

UNION ALL
SELECT 
  clientId() AS device_id,
  'I..1:1-1' AS measure_name,
  state.reported["I..1:1-1"] AS measure_value::varchar,
  metadata.reported["I..1:1-1"].timestamp * 1000 AS time
FROM '$aws/things/HneviceIoT1/shadow/update'

UNION ALL
-- 其余所有需要同步的测点按照上述格式依次追加UNION ALL段即可
SELECT 
  clientId() AS device_id,
  'M..1:1-1' AS measure_name,
  state.reported["M..1:1-1"] AS measure_value::varchar,
  metadata.reported["M..1:1-1"].timestamp * 1000 AS time
FROM '$aws/things/HneviceIoT1/shadow/update'

配套配置 #

不需要额外修改维度配置，规则会自动提取device_id作为维度，measure_name、measure_value、time会自动匹配Timestream的必填字段。

方案2：通过Lambda中转处理（适用测点动态变化场景） #

如果测点数量不固定、后续会新增测点，推荐用Lambda做中间处理，自动遍历所有上报字段批量写入Timestream。

配置步骤 #

1. 修改IoT Core规则，将触发动作设置为调用Lambda函数，把完整的影子更新事件传给Lambda
2. 给Lambda的执行角色添加Timestream的写入权限
3. 编写Lambda处理逻辑，示例代码（Python）如下：

import boto3

ts_client = boto3.client('timestream-write')
# 替换为你自己的Timestream库、表名
TS_DB = "siemens_iot_db"
TS_TABLE = "device_data"

def lambda_handler(event, context):
    device_id = event["clientId"]
    reported_data = event["state"]["reported"]
    ts_meta = event["metadata"]["reported"]
    records = []

    for measure_name, value in reported_data.items():
        # 取对应测点的上报时间戳，转成Timestream要求的毫秒级
        point_ts = str(ts_meta[measure_name]["timestamp"] * 1000)
        records.append({
            "Dimensions": [{"Name": "device_id", "Value": device_id}],
            "MeasureName": measure_name,
            "MeasureValue": str(value),
            "MeasureValueType": "VARCHAR",
            "Time": point_ts
        })
    
    if records:
        ts_client.write_records(
            DatabaseName=TS_DB,
            TableName=TS_TABLE,
            Records=records
        )
    return {"code": 0}

注意事项 #

• 如果上报的值是数值类型，可将MeasureValueType改为DOUBLE或BIGINT，同时对应转换value的类型
• 单批写入记录不要超过100条，超过的话可以分批调用写入接口

内容的提问来源于stack exchange，提问作者Michal Žák