MongoDB时序集合

MongoDB时序集合

时序数据

时序数据就是一系列随着时间变化的数据。时序数据由3个部分组件

• 时间。数据记录的时间
• 元数据。有时也叫做数据的来源。由一些列标签或者标记（label or tag）标识唯一的时序数据。很少发生改变。
• 测量值。有时也叫做度量或者值。随着时间变化的数据点，通常以键值对来表示。

时序集合

时序集合可以高效地存储同一来源的数据，并按相近时间存储。

优点

同普通集合相比，时序集合查询效率更好，占用磁盘空间更低。从Mongodb6.3开始，会自动创建一个时间和元数据的组合索引。

时序集合使用列存储并按时间序存储时序数据。使用列存储有以下好处。

• 减少处理时序数据的复杂度。
• 提高查询效率
• 减少磁盘的存储空间
• 减少读操作的IO请求
• 提高WiredTiger缓存的使用率

行为

时序集合和普通集合非常相似，可以像普通集合一样进行插入和查询。
mongodb使用内部集合，把时序集合当作可写但不会持久化的视图。当插入数据，内部集合会自动把时序数据转成优化后的存储格式。

查询时序数据时，时序集合会充分利用内部优化的存储格式来快速返回。

时序集合的简单操作

创建时序集合

1. 创建时序集合

db.createCollection(
"weather",
{
  timeseries: {
  timeField: "timestamp",
  metaField: "metadata"
}})
1
2
3
4
5
6
7

2. 设置时间字段和元数据字段

timeseries: {
   timeField: "timestamp",
   metaField: "metadata"
}
1
2
3
4

3. 定义数据的时间间隔，分别用以下2种方式。
   定义granularity字段

timeseries: {
   timeField: "timestamp",
   metaField: "metadata",
   granularity: "seconds"
}
1
2
3
4
5

或者 在Mongodb6.3以上版本定义bucketMaxSpanSeconds和bucketRoundingSeconds字段。这2个字段的值必须一样。

timeseries: {
   timeField: "timestamp",
   metaField: "metadata",
   bucketMaxSpanSeconds: "300",
   bucketRoundingSeconds: "300"
}
1
2
3
4
5
6

4. 可选择设置expireAfterSeconds字段表示当timeField字段的值太久了导致文档过期。

timeseries: {
   timeField: "timestamp",
   metaField: "metadata",
   granularity: "seconds",
   expireAfterSeconds: "86400"
}
1
2
3
4
5
6

插入测量值到时序集合

这里的例子，每个文档只有一个数据点。使用的是批量插入。

db.weather.insertMany( [
   {
      "metadata": { "sensorId": 5578, "type": "temperature" },
      "timestamp": ISODate("2021-05-18T00:00:00.000Z"),
      "temp": 12
   },
   {
      "metadata": { "sensorId": 5578, "type": "temperature" },
      "timestamp": ISODate("2021-05-18T04:00:00.000Z"),
      "temp": 11
   },
   {
      "metadata": { "sensorId": 5578, "type": "temperature" },
      "timestamp": ISODate("2021-05-18T08:00:00.000Z"),
      "temp": 11
   },
   {
      "metadata": { "sensorId": 5578, "type": "temperature" },
      "timestamp": ISODate("2021-05-18T12:00:00.000Z"),
      "temp": 12
   }
] )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

查询时序集合

查询时序集合和普通集合的方式一样。
这里的例子只返回一个文档

db.weather.findOne({
   "timestamp": ISODate("2021-05-18T00:00:00.000Z")
})
1
2
3

输出结果

{
   timestamp: ISODate("2021-05-18T00:00:00.000Z"),
   metadata: { sensorId: 5578, type: 'temperature' },
   temp: 12,
   _id: ObjectId("62f11bbf1e52f124b84479ad")
}
1
2
3
4
5
6

向时序集合执行聚合操作

db.weather.aggregate( [
   {
      $project: {
         date: {
            $dateToParts: { date: "$timestamp" }
         },
         temp: 1
      }
   },
   {
      $group: {
         _id: {
            date: {
               year: "$date.year",
               month: "$date.month",
               day: "$date.day"
            }
         },
         avgTmp: { $avg: "$temp" }
      }
   }
] )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

这个聚合操作使用时间来聚合数据并返回温度的平均值

{
  "_id" : {
    "date" : {
      "year" : 2021,
      "month" : 5,
      "day" : 18
    }
  },
  "avgTmp" : 12.714285714285714
}
{
  "_id" : {
    "date" : {
      "year" : 2021,
      "month" : 5,
      "day" : 19
    }
  },
  "avgTmp" : 13
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

时序集合的自动删除

手动激活时序集合自定过期删除

db.runCommand({
   collMod: "weather24h",
   expireAfterSeconds: 604801
})
1
2
3
4

改变时序集合的过期时间

db.runCommand({
   collMod: "weather24h",
   expireAfterSeconds: 604801
})
1
2
3
4

取消自定过期删除

db.runCommand({
    collMod: "weather24h",
    expireAfterSeconds: "off"
})
1
2
3
4

过期自动删除行为

MongoDB不保证过期数据会立马删除。一旦一个桶的所有文档的都过期，删除过期桶的后台任务要到下一次运行才会删除。一个桶存储的时序数据的时间跨度，是根据时序集合的granularity字段来决定的。

**后台删除任务每60s执行一次。因此，过期的文档在这周期内还会存在在集合中。这个周期涉及文档的过期时间，桶里面其他文档的过期时间以及后台任务的运行情况。
**

因为删除数据的周期涉及到mongodb实例的工作负载，过期数据的存在可能会超过后台运行周期的60s。

设置时序集合的时间细粒度

当你创建时序集合，mongodb会自动创建一个system.buckets的系统集合。把所有的时序数据合并到bukects（桶）里面。通过设置时间细粒度，控制数据装到桶里的频率。这个频率一般根据数据的采集频率。

从Mongodb6.3开始，可以设置bucketMaxSpanSeconds和bucketRoundingSeconds参数来自定义桶的边界。更精确地控制时序数据多长时间装桶。

使用granularity字段

granularity的值，决定桶的最大时间间隔

granularity的默认值为秒。修改granularity的值为接近实际数据采集的频率值，可以提高时序集合的性能。如果使用1000个传感器记录天气数据，但每个传感器每5分钟采集一次。应该设置granularity值为“minutes”

db.createCollection(
    "weather24h",
    {
       timeseries: {
          timeField: "timestamp",
          metaField: "metadata",
          granularity: "minutes"
       },
       expireAfterSeconds: 86400
    }
)
1
2
3
4
5
6
7
8
9
10
11

上面的例子，如果granularity值设置为hours，那么一个月的天气数据都会进入到一个桶里面。这样会造成更长的遍历时间和更慢的查询。如果granularity值设置为seconds，会导致一个周期内（5分钟）有更多的桶。大部分的桶可能只包含一个文档。

使用自定义的装桶参数

在Mongodb6.3以上版本。装桶参数，除了granularity参数外，还可以设置2个参数来手动设置桶的边界。通常在需要更加精确地优化大量查询和插入数据的性能时使用。

使用自定义装桶参数，设置这2个参数为相同值，而且不要设置granularity。

• bucketMaxSpanSeconds。设置同一个桶，数据的时间差的最大值。即最大时间跨度。值为1-31536000。
• bucketRoundingSeconds 这个值确定桶的开始时间。当一个文档路由到一个新桶，Mongdb会使用bucketRoundingSeconds来取整这个文档的时间戳并设置这个桶的最小时间（开始时间）。

对于5分钟采集一次的天气数据的例子，可以设置自定义装桶参数为300秒（5分钟），而不是设置granularity值为minutes。

db. createCollection(
   "weather24h",
   {
      timeseries: {
         timeField: "timestamp",
         metaField: "metadata",
         bucketMaxSpanSeconds: 300,
         bucketRoundingSeconds: 300
      }
   }
)
1
2
3
4
5
6
7
8
9
10
11

如果一个文档的时间戳为2023-03-27T18:24:35Z且没有一个现成的桶符合条件。Mongdb会创建一个新桶，并设置桶的最小时间为2023-03-27T18:20:00Z，最大时间为2023-03-27T18:24:59Z。

改变时序集合细粒度

提高时序集合的granularity。

db.runCommand({
   collMod: "weather24h",
   timeseries: { granularity: "seconds" || "minutes" || "hours" }
})
1
2
3
4

或者 提高bucketMaxSpanSeconds 和 bucketMaxSpanSeconds值

db.runCommand({
   collMod: "weather24h",
   timeseries: {
      bucketRoundingSeconds: "86400",
      bucketMaxSpanSeconds: "86400"
   }
})
1
2
3
4
5
6
7

不能降低granularity、bucketMaxSpanSeconds、bucketMaxSpanSeconds值

为时序集合添加索引

注意：Mongodb默认创建_id唯一索引。文档说的辅助的第二索引。这里统一叫做索引。

为了提高查询效率，可以为时序集合创建更多的索引来支持通用的查询。从MongoDB6.3开始，MongoDb会自动创建一个时间和元数据的复合索引。

以下天气数据的例子，你可能会考虑创建一个新的索引。

db.createCollection(
"weather",
{
   timeseries: {
      timeField: "timestamp",
      metaField: "metadata"
}})
1
2
3
4
5
6
7

metadata字段是一个子文档。包含传感器ID和类型。

{
   "timestamp": ISODate("2021-05-18T00:00:00.000Z"),
   "metadata": {
     "sensorId": 5578,
     "type": "temperature"
   },
   "temp": 12
}
1
2
3
4
5
6
7
8

默认的复合索引只会索引整个metadata的子文档。所以这个索引只能使用$eq操作符来查询。通过对metadata的子文档字段建立索引，来提高metadata查询性能。

例如， 这个$in查询会被metadata.type的索引提高性能。

{ metadata.type:{ $in: ["temperature", "pressure"] }}
1

使用索引来提高排序性能

对时序集合的排序，会使用timeField的索引。在某些条件下，排序操作可能会使用metaField和timeField的复合索引。

聚合操作的$match and $sort决定了时序集合使用哪个索引。下面的例子可能会使用以下索引。

• 对{ : ±1 } 排序，使用索引
• 对{ : ±1, timeField: ±1 }排序，使用默认的{ : ±1, timeField: ±1 }复合索引。
• 对{ : ±1 }排序，并且使用匹配。使用{ metaField: ±1, timeField: ±1 }复合索引。

迁移数据到时序集合

1. 如果原来的集合没有metadata元数据字段。使用 $addFields 聚合操作添加。
   这个是原来的集合

{
    "_id" : ObjectId("5553a998e4b02cf7151190b8"),
    "st" : "x+47600-047900",
    "ts" : ISODate("1984-03-05T13:00:00Z"),
    "position" : {
      "type" : "Point",
      "coordinates" : [ -47.9, 47.6 ]
    },
    "elevation" : 9999,
    "callLetters" : "VCSZ",
    "qualityControlProcess" : "V020",
    "dataSource" : "4",
    "type" : "FM-13",
    "airTemperature" : { "value" : -3.1, "quality" : "1" },
    "dewPoint" : { "value" : 999.9, "quality" : "9" },
    "pressure" : { "value" : 1015.3, "quality" : "1" },
    "wind" : {
      "direction" : { "angle" : 999, "quality" : "9" },
      "type" : "9",
      "speed" : { "rate" : 999.9, "quality" : "9" }
    },
    "visibility" : {
      "distance" : { "value" : 999999, "quality" : "9" },
      "variability" : { "value" : "N", "quality" : "9" }
    },
    "skyCondition" : {
      "ceilingHeight" : { "value" : 99999, "quality" : "9", "determination" : "9" },
      "cavok" : "N"
    },
    "sections" : [ "AG1" ],
    "precipitationEstimatedObservation" : { "discrepancy" : "2",
    "estimatedWaterDepth" : 999 }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

2. 使用 $project来包含或者移除字段。

{ $addFields: {
    metaData: {
      "st": "$st",
      "position": "$position",
      "elevation": "$elevation",
      "callLetters": "$callLetters",
      "qualityControlProcess": "$qualityControlProcess",
      "type": "$type"
    }
  },
},
{ $project: {
    _id: 1,
    ts: 1,
    metaData: 1,
    dataSource: 1,
    airTemperature: 1,
    dewPoint: 1,
    pressure: 1,
    wind: 1,
    visibility: 1,
    skyCondition: 1,
    sections: 1,
    precipitationEstimatedObservation: 1
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

3. 使用聚合操作符$out把集合的数据迁移到时序集合

db.weather_data.aggregate([
  {
     $addFields: {
       metaData: {
         "st": "$st",
         "position": "$position",
         "elevation": "$elevation",
         "callLetters": "$callLetters",
         "qualityControlProcess": "$qualityControlProcess",
         "type": "$type"
       }
     },
  }, {
     $project: {
        _id: 1,
        ts: 1,
        metaData: 1,
        dataSource: 1,
        airTemperature: 1,
        dewPoint: 1,
        pressure: 1,
        wind: 1,
        visibility: 1,
        skyCondition: 1,
        sections: 1,
        precipitationEstimatedObservation: 1
     }
  }, {
     $out: {
       db: "mydatabase",
       coll: "weathernew",
       timeseries: {
         timeField: "ts",
         metaField: "metaData"
       }
     }
  }
])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

对时序集合分片

不能重分片以及分片的时序集合。但是可以重新定义分片键。

建立分片时序集合

1. 连接分片集群
   使用mongsh连接 mongos的分片集合

mongosh --host <hostname> --port <port>
1

2. 确定数据库启动分片

sh.status()
1

--- Sharding Status ---
   sharding version: {
      "_id" : 1,
      "minCompatibleVersion" : 5,
      "currentVersion" : 6,
...
1
2
3
4
5
6

3. 创建时序集合

sh.shardCollection(
   "test.weather",
   { "metadata.sensorId": 1 },
   {
      timeseries: {
         timeField: "timestamp",
         metaField: "metadata",
         granularity: "hours"
      }
   }
)
1
2
3
4
5
6
7
8
9
10
11

分片键为 metadata.sensorId

对已经存在的时序集合进行分片

1. 使用mongsh连接 mongos的分片集合

mongosh --host <hostname> --port <port>
1

2. 确定数据库启动分片

sh.status()
1

--- Sharding Status ---
   sharding version: {
      "_id" : 1,
      "minCompatibleVersion" : 5,
      "currentVersion" : 6,
1
2
3
4
5

3. 使用shardCollection()分片时序集合

sh.shardCollection( "test.deliverySensor", { "metadata.location": 1 } )
1

分片键为 metadata.sensorId

时序数据库的最佳实践

优化插入

按元数据批量处理文档

• 避免网络往返，使用单个insertMany()比多个insertOne()好。
• 尽可能按元数据的排序多个测量值（文档）。
  例如，有2个传感器A、B，一个传感器的多个测量值只会花费一个插入，而不是每个测量值多个插入。（这里的测量值是文档）

下面的批量插入6个文档的操作，实际只会产生2次插入。因为这些文档都按传感器排序

db.temperatures.insertMany( [
   {
      "metadata": {
         "sensor": "sensorA"
      },
      "timestamp": ISODate("2021-05-18T00:00:00.000Z"),
      "temperature": 10
   },
   {
      "metadata": {
         "sensor": "sensorA"
      },
      "timestamp": ISODate("2021-05-19T00:00:00.000Z"),
      "temperature": 12
   },
   {
      "metadata": {
         "sensor": "sensorA"
      },
      "timestamp": ISODate("2021-05-20T00:00:00.000Z"),
      "temperature": 13
   },
   {
      "metadata": {
         "sensor": "sensorB"
      },
      "timestamp": ISODate("2021-05-18T00:00:00.000Z"),
      "temperature": 20
   },
   {
      "metadata": {
         "sensor": "sensorB"
      },
      "timestamp": ISODate("2021-05-19T00:00:00.000Z"),
      "temperature": 25
   },
   {
      "metadata": {
         "sensor": "sensorB"
      },
      "timestamp": ISODate("2021-05-20T00:00:00.000Z"),
      "temperature": 26
   }
] )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

使用一致的文档顺序

批量插入的文档，字段都保持一致的顺序会提高插入效率。

{
   "_id": ObjectId("6250a0ef02a1877734a9df57"),
   "timestamp": ISODate("2020-01-23T00:00:00.441Z"),
   "name": "sensor1",
   "range": 1
},
{
   "_id": ObjectId("6560a0ef02a1877734a9df66"),
   "timestamp": ISODate("2020-01-23T01:00:00.441Z"),
   "name": "sensor1",
   "range": 5
}
1
2
3
4
5
6
7
8
9
10
11
12

相反，字段顺序不一致，就没有得到优化。

{
   "range": 1,
   "_id": ObjectId("6250a0ef02a1877734a9df57"),
   "name": "sensor1",
   "timestamp": ISODate("2020-01-23T00:00:00.441Z")
},
{
   "_id": ObjectId("6560a0ef02a1877734a9df66"),
   "name": "sensor1",
   "timestamp": ISODate("2020-01-23T01:00:00.441Z"),
   "range": 5
}
1
2
3
4
5
6
7
8
9
10
11
12

增加客户端的个数

提高写入时序集合的客户端个数，会提高插入性能。

注意：必须禁用重试写入，否则时序集合不会合并多个客户端的写入。

优化压缩率

省略文档中的空对象和空数组。

{
 "timestamp": ISODate("2020-01-23T00:00:00.441Z"),
 "coordinates": [1.0, 2.0]
},
{
   "timestamp": ISODate("2020-01-23T00:00:10.441Z"),
   "coordinates": []
},
{
   "timestamp": ISODate("2020-01-23T00:00:20.441Z"),
   "coordinates": [3.0, 5.0]
}
1
2
3
4
5
6
7
8
9
10
11
12

上面这个例子，coordinates字段存在有值数组和空数组，会造成压缩器的schema发生改变。schema改变造成第2和第3个文档没有压缩。

相反，下面的例子省略了空数组，会有利于压缩器的压缩

{
   "timestamp": ISODate("2020-01-23T00:00:00.441Z"),
   "coordinates": [1.0, 2.0]
},
{
   "timestamp": ISODate("2020-01-23T00:00:10.441Z")
},
{
   "timestamp": ISODate("2020-01-23T00:00:20.441Z"),
   "coordinates": [3.0, 5.0]
}
1
2
3
4
5
6
7
8
9
10
11

减少数据的小数点位数

根据应用的情况确定保留的小数点位。更少的小数点位可以提高压缩率。

优化查询

设置适当的桶细粒度

创建时序集合，Mongodb会合并时序数据到桶里。精确地设置细粒度，能够控制数据的装桶频率，通常基于数据的采集频率。
从Mongdb6.3开始，可以设置自定义装桶参数bucketMaxSpanSeconds和bucketRoundingSeconds来指定桶的边界，更精确地控制时序数据的装桶。

创建索引

为了提高查询效率，为timeField和metaField建立索引可以支持更通用的查询。从MongoDb6.3起，默认创建timeField和metaField的复合索引。