Redis 时间序列

前言

REmote DIctionary Server(Redis) 是一个使用 ANSI C 编写的开源、支持网络、基于内存、分布式、可选持久性的键值对存储数据库。

RedisTimeSeries 是 Redis 的一个扩展模块。它专门面向时间序列数据提供了数据类型和访问接口，并且支持在 Redis 实例上直接对数据进行按时间范围的聚合计算。

由于 RedisTimeSeries 不属于 Redis 的内置数据结构，在使用时，需要先把它的源码单独编译成动态链接库 redistimeseries.so。

在本文中，直接使用官方说明文档中的 docker 实例:

1	docker run -p 6379:6379 -it --rm redislabs/redistimeseries

RedisTimeSeries 基本操作

RedisTimeSeries 的操作主要有 5 个：

TS.CREATE 命令创建时间序列数据集合
TS.ADD 命令插入数据
TS.GET 命令读取最新数据
TS.MGET 命令按标签过滤查询数据集合
TS.RANGE 支持聚合计算的范围查询

TS.CREATE 命令创建时间序列数据集合

我们可以使用 TS.CREATE 命令来创建一个时间序列数据集合，同时可以指定一些参数。

1	TS.CREATE key [RETENTION retentionPeriod] [ENCODING [UNCOMPRESSED\|COMPRESSED]] [CHUNK_SIZE size] [DUPLICATE_POLICY policy] [LABELS {label value}...]

通过设定 RETENTION 可以指定数据的过期时间（以毫秒为单位）。例如，我们执行下面的命令，创建一个 key 为 device:temperature、数据有效期为 600s 的时间序列数据集合。也就是说，这个集合中的数据创建了 600s 后，就会被自动删除。

此外，我们还可以为数据集合设置 LABELS 标签，来表示数据集合的属性。例如下面的命令表明这个数据集合中记录的是属于 device_id 为 1 的数据。

1 2	> TS.CREATE device:temperature RETENTION 600000 LABELS device_id 1 "OK"

TS.ADD 命令插入数据

使用 TS.ADD 命令插入数据, 包括时间戳和具体的数值。

1	TS.CREATE key [RETENTION retentionPeriod] [ENCODING [UNCOMPRESSED\|COMPRESSED]] [CHUNK_SIZE size] [DUPLICATE_POLICY policy] [LABELS {label value}...]

例如，执行下列 TS.ADD 命令时，向之前创建的 device:temperature 集合中插入了一条数据，时间戳为 1652321057（即 2022-05-12 10:04:17），数值为 26.5。

1 2	> TS.ADD device:temperature 1652321057 26.5 (integer) 1652321057

TS.GET 命令读取数据

使用 TS.GET 命令读取数据。

1	TS.GET key

TS.GET 命令会读取指定 key 的一条最新数据，并返回一个包含时间戳和数值的数组。

1
2
3

> TS.GET device:temperature
1) "1652321057"
2) "26.5"

TS.MGET 命令按标签过滤查询数据集合

使用 TS.MGET 命令按标签过滤查询数据集合。

1	TS.MGET [WITHLABELS \| SELECTED_LABELS label...] FILTER filter...

在使用 TS.CREATE 创建数据集合时，我们可以给集合设置标签属性。当我们进行查询时，就可以在查询条件中对集合标签属性进行匹配，最后的查询结果里只返回匹配上的集合中的最新数据。

> TS.MGET FILTER device_id!=2
1) 1) "device:temperature:1"
   2) (empty list or set)
   3) 1) (integer) 1596417000
      2) "25.3"
2) 1) "device:temperature:3"
   2) (empty list or set)
   3) 1) (integer) 1596417000
      2) "29.5"
3) 1) "device:temperature:4"
   2) (empty list or set)
   3) 1) (integer) 1596417000
      2) "30.1"

TS.RANGE 支持聚合计算的范围查询

使用 TS.RANGE 命令支持聚合计算的范围查询。

TS.RANGE key fromTimestamp toTimestamp
         [FILTER_BY_TS ts...]
         [FILTER_BY_VALUE min max]
         [COUNT count]
         [[ALIGN value] AGGREGATION aggregator bucketDuration [BUCKETTIMESTAMP bt] [EMPTY]]

例如，在执行下列命令时，我们就可以按照每 180s 的时间窗口，对 2020-08-03 09:05:00 和 2020-08-03 09-12:00 这段时间内的数据进行均值计算了。

> TS.RANGE device:temperature 1596416700 1596417120 AGGREGATION avg 180000
1) 1) (integer) 1596416700
   2) "25.6"
2) 1) (integer) 1596416880
   2) "25.8"
3) 1) (integer) 1596417060
   2) "26.1"

Python redis

在 python 中，我们可以使用 redis 库来操作数据库，示例如下：

import random
import time

import redis

# connect to redis
client = redis.Redis(host='localhost', port=6379, db=0)
device = 'G_Data0000000177'

# add time series
for i in range(10):
    timestamp = int(time.time())
    value = random.randint(0, 100)
    client.ts().add(key=f'{device}:temperature', timestamp=timestamp, value=value)
    time.sleep(1)

# get range data
print(client.ts().range(key=f'{device}:temperature',
                        from_time=0,
                        to_time=int(time.time())))
# > [(1652322321, 70.0), (1652322322, 2.0), (1652322323, 76.0), (1652322324, 29.0), (1652322325, 53.0), (1652322326, 76.0), (1652322327, 55.0), (1652322328, 45.0), (1652322329, 55.0), (1652322330, 77.0)]