如何从00:00开始计算IoT设备15分钟时段数据平均值？

方案一：AWS IoT Rules Engine + Amazon Timestream（推荐） #

Timestream是专为时序数据设计的托管数据库，天然适配IoT设备的时间序列数据聚合需求，无需额外处理批量文件。

实施步骤 #

• 1. 创建Timestream数据库与表，将设备name设为维度属性，time字段作为时间戳主键。
• 2. 修改IoT Rules Engine规则，将设备上报数据直接写入Timestream，规则中指定时间字段为payload里的time（epoch timestamp）。
• 3. 配置Timestream连续查询，定义900秒（15分钟）的对齐滚动窗口（Tumbling Window），对齐到小时/15分钟边界（确保从00:00开始划分时段），按设备名、日期、时段分组，计算voltage和power的平均值：

  CREATE OR REPLACE CONTINUOUS QUERY device_avg_query
  ON your_timestream_table
  INTO your_aggregated_table
  SELECT
    name AS device_name,
    DATE_TRUNC('minute', time, 15) AS block_start,
    DATE_FORMAT(block_start, '%d-%b-%Y') AS event_date,
    CONCAT(DATE_FORMAT(block_start, '%H:%i'), '-', DATE_FORMAT(block_start + INTERVAL 15 MINUTE, '%H:%i')) AS time_block,
    AVG(voltage) AS avg_voltage,
    AVG(power) AS avg_power
  GROUP BY name, DATE_TRUNC('minute', time, 15);
  

• 4. 将聚合结果导出到目标S3桶：通过Timestream的导出功能，按日期、设备维度生成CSV格式文件，匹配需求的表格结构。

优势 #

• 自动处理时序数据的时段对齐与聚合，适配200+设备的上报量（800条/分钟），性能稳定；
• 无需手动处理S3文件的批量读取与分区逻辑，运维成本低。

方案二：Kinesis Data Analytics 处理现有Firehose数据流 #

如果需要保留现有Firehose到S3的存储流程，可通过Kinesis Data Analytics实时计算聚合值。

实施步骤 #

• 1. 调整数据流转：让IoT Rules Engine将数据发送到Kinesis Data Stream，再通过该流同时分发给Firehose（存S3）和Kinesis Data Analytics（计算聚合）。
• 2. 创建Kinesis Data Analytics应用，用SQL定义15分钟对齐滚动窗口，计算平均值：

  CREATE OR REPLACE STREAM aggregated_stream (
      device_name VARCHAR(20),
      event_date VARCHAR(20),
      time_block VARCHAR(20),
      avg_voltage DOUBLE,
      avg_power DOUBLE
  );
  
  CREATE OR REPLACE PUMP aggregation_pump AS
  INSERT INTO aggregated_stream
  SELECT
      name AS device_name,
      DATE_FORMAT(FROM_UNIXTIME(time), '%d-%b-%Y') AS event_date,
      CONCAT(
          DATE_FORMAT(FROM_UNIXTIME(FLOOR(time/900)*900), '%H:%i'),
          '-',
          DATE_FORMAT(FROM_UNIXTIME(FLOOR(time/900)*900 + 900), '%H:%i')
      ) AS time_block,
      AVG(voltage) AS avg_voltage,
      AVG(power) AS avg_power
  FROM source_sql_stream
  GROUP BY
      name,
      FLOOR(time/900),
      DATE_FORMAT(FROM_UNIXTIME(time), '%d-%b-%Y')
  WINDOW TUMBLING (SIZE 900 SECONDS, ALIGNMENT 'HOUR');
  

• 3. 将Kinesis Data Analytics的输出流导出到目标S3桶（通过Firehose），生成符合格式的汇总文件。

优势 #

• 兼容现有数据存储流程，实时计算聚合结果，避免Lambda处理S3文件的延迟问题。

方案三：改进S3+Lambda方案（解决原方案痛点） #

如果坚持使用原有架构，需解决跨时段数据、重复计算等核心问题。

原方案常见问题 #

• Firehose的900秒缓冲导致单S3文件包含跨15分钟时段的数据；
• Lambda触发逻辑不合理，引发重复计算或数据遗漏；
• 未按设备+时段正确分组计算。

改进步骤 #

• 1. 调整Firehose分区规则：除设备名外，新增15分钟时段分区键，格式为device=!{name}/date=!{timestamp:yyyy-MM-dd}/block=!{timestamp:HH-mm}（确保block取00:00、00:15这类对齐值），保证每个S3文件仅对应一个设备的一个15分钟时段。
• 2. 配置Lambda触发：设置仅触发特定前缀的S3文件，启用批量触发，避免单文件多次触发。
• 3. Lambda函数核心逻辑（Python示例）：

  import boto3
  import json
  from datetime import datetime
  
  s3 = boto3.client('s3')
  dynamodb = boto3.client('dynamodb')
  PROCESSED_TABLE = 'ProcessedFiles'
  TARGET_BUCKET = 'your-target-bucket'
  
  def lambda_handler(event, context):
      # 获取S3文件信息
      record = event['Records'][0]['s3']
      src_bucket = record['bucket']['name']
      src_key = record['object']['key']
  
      # 幂等校验：避免重复处理
      try:
          dynamodb.get_item(TableName=PROCESSED_TABLE, Key={'file_key': {'S': src_key}})
          return {'statusCode': 200, 'body': 'File already processed'}
      except dynamodb.exceptions.ResourceNotFoundException:
          pass
  
      # 读取并解析S3文件
      response = s3.get_object(Bucket=src_bucket, Key=src_key)
      content = response['Body'].read().decode('utf-8')
      records = [json.loads(line) for line in content.split('\n') if line.strip()]
      if not records:
          return {'statusCode': 200, 'body': 'No records to process'}
  
      # 计算时段与平均值
      first_ts = records[0]['time']
      block_start_ts = first_ts - (first_ts % 900)
      block_start = datetime.fromtimestamp(block_start_ts)
      event_date = block_start.strftime('%d-%b-%Y')
      time_block = f"{block_start.strftime('%H:%M')}-{(block_start_ts + 900):.0f}".strftime('%H:%M')
      avg_voltage = sum(r['voltage'] for r in records) / len(records)
      avg_power = sum(r['power'] for r in records) / len(records)
  
      # 写入目标S3桶
      target_key = f"aggregated/{block_start.strftime('%Y/%m/%d')}/{records[0]['name']}/{block_start.strftime('%H%M')}.csv"
      result_line = f"{event_date},{time_block},{avg_voltage:.3f},{avg_power:.3f}\n"
      s3.put_object(Bucket=TARGET_BUCKET, Key=target_key, Body=result_line)
  
      # 标记文件已处理
      dynamodb.put_item(TableName=PROCESSED_TABLE, Item={'file_key': {'S': src_key}})
      return {'statusCode': 200, 'body': 'Aggregation completed'}
  

• 4. 预创建DynamoDB表ProcessedFiles，用于记录已处理的S3文件键，实现幂等性。

内容的提问来源于stack exchange，提问作者anonymous_33008899