22. MongoDB 时序集合与 IoT 数据实践

// 创建时序集合：记录 1000 台设备的温度/湿度/电量
db.createCollection("device_metrics", {
  timeseries: {
    timeField: "ts",           // 必须存在，类型为 Date
    metaField: "deviceId",     // meta 字段，用于桶内分组
    granularity: "seconds"     // 写入间隔预估：seconds | minutes | hours
  },
  expireAfterSeconds: 60 * 60 * 24 * 90  // 90 天后自动删除
})

// 查看集合的 timeseries 配置
db.runCommand({ listCollections: 1, filter: { name: "device_metrics" } })
  .cursor.firstBatch[0].options.timeseries
// {
//   "timeField": "ts",
//   "metaField": "deviceId",
//   "granularity": "seconds",
//   "bucketMaxSpanSeconds": 900,
//   "bucketRoundingSeconds": 900
// }

// 良好的字段划分示例
db.device_metrics.insertMany([
  {
    ts: ISODate("2026-09-27T08:00:00.000Z"),
    deviceId: { sn: "SN-1001", rack: "A-12" },   // meta：高基数但分组稳定
    temp: 36.5,   // measurement：同一设备内的测量值
    humidity: 58.2,
    voltage: 3.72
  },
  {
    ts: ISODate("2026-09-27T08:00:05.000Z"),
    deviceId: { sn: "SN-1001", rack: "A-12" },
    temp: 36.7,
    humidity: 58.0,
    voltage: 3.71
  }
])

// 反面示例：把时间戳放进 metaField，破坏桶化（绝不这样做）
db.createCollection("bad_metrics", {
  timeseries: { timeField: "ts", metaField: "requestId", granularity: "seconds" }
})

// 通过内部视图观察桶的数量与跨度（system.buckets 视图，生产环境慎用）
db.getSiblingDB("admin").command({ listCollections: 1, filter: { type: "timeseries" } })

// 使用 $listSearchIndexes 之外的 explain 观察查询是否命中桶裁剪
db.device_metrics.explain("executionStats").aggregate([
  { $match: { deviceId: { sn: "SN-1001" }, ts: { $gte: ISODate("2026-09-27T08:00:00Z"), $lt: ISODate("2026-09-27T09:00:00Z") } } }
])

db.createCollection("wind_speed", {
  timeseries: {
    timeField: "ts",
    metaField: "stationId",
    granularity: "seconds",
    bucketMaxSpanSeconds: 60,     // 每桶最多覆盖 60 秒
    bucketRoundingSeconds: 60     // 桶边界对齐到整分钟
  }
})

// 查询某设备最近 10 分钟的温度（自动走桶裁剪 + 时间字段索引）
db.device_metrics.find({
  "deviceId.sn": "SN-1001",
  ts: { $gte: new Date(Date.now() - 10 * 60 * 1000) }
}).sort({ ts: 1 }).limit(120)

// 按 5 分钟粒度计算平均温度与最大湿度
db.device_metrics.aggregate([
  { $match: { "deviceId.sn": "SN-1001", ts: { $gte: ISODate("2026-09-27T08:00:00Z") } } },
  { $group: {
      _id: { $dateTrunc: { date: "$ts", unit: "minute", binSize: 5 } },
      avgTemp: { $avg: "$temp" },
      maxHumidity: { $max: "$humidity" },
      count: { $sum: 1 }
  }},
  { $sort: { _id: 1 } }
])

db.device_metrics.aggregate([
  { $match: { "deviceId.sn": "SN-1001", ts: { $gte: ISODate("2026-09-27T08:00:00Z"), $lt: ISODate("2026-09-27T09:00:00Z") } } },
  { $densify: { field: "ts", range: { step: 60 * 1000, unit: "millisecond", bounds: [ISODate("2026-09-27T08:00:00Z"), ISODate("2026-09-27T09:00:00Z")] } } },
  { $fill: { output: { temp: { method: "linear" } }, sortBy: { ts: 1 } } }
])

// 每小时降采样：将原始 5 秒数据汇总为小时均值
db.device_metrics.aggregate([
  { $match: { ts: { $gte: ISODate("2026-09-01T00:00:00Z"), $lt: ISODate("2026-09-07T00:00:00Z") } } },
  { $group: {
      _id: { device: "$deviceId", hour: { $dateTrunc: { date: "$ts", unit: "hour" } } },
      avgTemp: { $avg: "$temp" },
      minTemp: { $min: "$temp" },
      maxTemp: { $max: "$temp" },
      samples: { $sum: 1 }
  }},
  { $merge: { into: "device_metrics_hourly", on: "_id", whenMatched: "replace", whenNotMatched: "insert" } }
])

db.runCommand({
  collMod: "device_metrics",
  timeseries: { expireAfterSeconds: 60 * 60 * 24 * 7 }  // 改为保留 7 天
})

// 为 meta 字段创建二级索引（复合：meta + 时间）
db.device_metrics.createIndex({ "deviceId.sn": 1, ts: -1 })

// 查看索引
db.device_metrics.getIndexes()
// [ { v: 2, key: { "deviceId.sn": 1, ts: -1 }, name: "deviceId.sn_1_ts_-1" }, ... ]

// 对 measurement 字段建部分索引（示例：仅对关键测量值字段，慎用）
db.device_metrics.createIndex(
  { "deviceId.sn": 1, temp: -1 },
  { partialFilterExpression: { temp: { $type: "number" } } }
)

# mongod.conf：时序集合为主时的引擎建议
storage:
  wiredTiger:
    engineConfig:
      cacheSizeGB: 8
    collectionConfig:
      blockCompressor: zstd   # 列式压缩之外的块级压缩兜底

// 步骤 1：创建原始时序集合
db.createCollection("sensor_raw", {
  timeseries: {
    timeField: "ts",
    metaField: "sensorId",
    granularity: "seconds"
  },
  expireAfterSeconds: 60 * 60 * 24 * 30
})

// 步骤 2：为查询热点建索引（楼层 + 时间）
db.sensor_raw.createIndex({ "sensorId.floor": 1, ts: -1 })

// 步骤 3：写入遥测
function report(sensorId, floor, temp, watt) {
  db.sensor_raw.insertOne({
    ts: new Date(),
    sensorId: { id: sensorId, floor: floor },
    temp: temp,
    power: watt
  });
}

// 步骤 4：每 5 分钟聚合一次（可放入定时任务）
db.sensor_raw.aggregate([
  { $match: { ts: { $gte: new Date(Date.now() - 5 * 60 * 1000) } } },
  { $group: {
      _id: { floor: "$sensorId.floor", slot: { $dateTrunc: { date: "$ts", unit: "minute", binSize: 5 } } },
      avgTemp: { $avg: "$temp" },
      avgPower: { $avg: "$power" }
  }},
  { $merge: { into: "sensor_5min", on: "_id", whenMatched: "replace", whenNotMatched: "insert" } }
])

// 步骤 5：能耗告警——某楼层近 5 分钟平均功率超过阈值
db.sensor_5min.find({
  "_id.floor": 3,
  "_id.slot": { $gte: new Date(Date.now() - 5 * 60 * 1000) },
  avgPower: { $gt: 8000 }
}).sort({ "_id.slot": -1 }).limit(1)