基于Node.js批量处理IoT数据并存储至MongoDB的技术问询

问题背景 #

现有300余台IoT设备每15秒上报数据，需基于Node.js完成以下流程：

• 提取上报JSON负载中的所有变量
• 校验变量是否存在于数据库，不存在则插入variables集合
• 将变量数值、上报时间戳插入values集合

当前数据库Schema设计如下：

Device集合 #

_id : ObjectId()
name: String
description : String

Variables集合 #

_id : ObjectId()
name : String // 例如：humidity, temp, windspeed
device : ref(device_collection)

Values集合 #

variable: ref(variables collection)
value : number/float
timestamp : date

Node.js实现方案 #

基于Mongoose实现核心逻辑，针对批量上报场景做效率优化：

1. 定义Mongoose Schema #

const mongoose = require('mongoose');

// Device Schema
const deviceSchema = new mongoose.Schema({
  name: String,
  description: String
});
const Device = mongoose.model('Device', deviceSchema);

// Variable Schema
const variableSchema = new mongoose.Schema({
  name: String,
  device: { type: mongoose.Schema.Types.ObjectId, ref: 'Device' }
});
// 联合唯一索引：避免同一设备重复创建同名变量
variableSchema.index({ name: 1, device: 1 }, { unique: true });
const Variable = mongoose.model('Variable', variableSchema);

// Value Schema
const valueSchema = new mongoose.Schema({
  variable: { type: mongoose.Schema.Types.ObjectId, ref: 'Variable' },
  value: Number,
  timestamp: Date
});
// 索引：优化按变量+时间范围的查询效率
valueSchema.index({ variable: 1, timestamp: -1 });
const Value = mongoose.model('Value', valueSchema);

2. 批量处理上报数据逻辑 #

async function processDeviceData(deviceId, payload, timestamp) {
  try {
    // 提取payload中的变量名与对应值
    const variableEntries = Object.entries(payload);

    // 批量查询或创建变量（upsert自动处理存在/不存在场景）
    const variableIds = await Promise.all(
      variableEntries.map(async ([varName, _]) => {
        const variable = await Variable.findOneAndUpdate(
          { name: varName, device: deviceId },
          { name: varName, device: deviceId },
          { upsert: true, new: true, lean: true }
        );
        return variable._id;
      })
    );

    // 批量构造Value文档
    const valueDocs = variableEntries.map(([_, value], index) => ({
      variable: variableIds[index],
      value: parseFloat(value),
      timestamp: timestamp || new Date()
    }));

    // 批量插入Value集合
    await Value.insertMany(valueDocs);
  } catch (err) {
    console.error('数据处理失败:', err);
    throw err;
  }
}

Schema优化建议 #

1. 强化唯一性约束 #

在variables集合添加设备ID+变量名的联合唯一索引，既避免重复数据，又能提升变量查询的效率。

2. 改用时间序列集合应对高频数据 #

300台设备每15秒上报，日均产生约1728万条数据，传统单条存储模式长期会导致查询性能下降。建议使用MongoDB时间序列集合：

• 自动按时间分片存储，压缩率更高
• 针对时间范围查询做了专属优化
  示例时间序列Schema：

const valueTimeSeriesSchema = new mongoose.Schema({
  deviceId: mongoose.Schema.Types.ObjectId,
  variableName: String,
  timestamp: Date,
  value: Number
}, { timeseries: { timeField: 'timestamp', metaField: 'deviceId' } });

3. 补充必要索引 #

• 在values集合添加{variable: 1, timestamp: -1}索引，优化按变量查询历史数据的速度
• 若需频繁按设备查询所有变量，给variables集合添加{device: 1}索引

4. 冗余字段优化查询效率 #

如果经常需要获取某设备某变量的最新数据，可在variables集合新增lastValue和lastTimestamp字段，每次插入values时同步更新，避免每次查询都遍历海量历史数据。

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