Redis `Timeseries` (时间序列)：传感器数据与日志存储

好的，没问题。

各位朋友，大家好！今天咱们来聊聊 Redis 的一个相当酷炫的功能——TimeSeries，也就是时间序列。这玩意儿可不是什么高深莫测的黑科技，说白了就是专门用来存储和处理时间戳数据的。想象一下，你家智能家居里的温度传感器，每隔几秒钟就往外蹦出一个温度值，或者你的服务器日志，每时每刻都在记录各种事件。这些数据都有一个共同点：都带着时间戳。那怎么高效地存储和查询这些数据呢？Redis 的 TimeSeries 就派上用场了。

什么是时间序列？

简单来说，时间序列就是按照时间顺序排列的一系列数据点。每个数据点都包含一个时间戳和一个对应的值。比如：

2023-10-27 10:00:00, 25.5 (温度：25.5 摄氏度)
2023-10-27 10:00:01, 25.6
2023-10-27 10:00:02, 25.7

这些数据点按照时间先后顺序排列，就构成了一个时间序列。

为什么要用 Redis TimeSeries？

可能有人会问，我用关系型数据库，或者 NoSQL 数据库，甚至直接写文件，也能存这些数据啊，为啥非得用 Redis TimeSeries？

原因很简单：快！

Redis 本身就是个内存数据库，读写速度极快。TimeSeries 又是专门针对时间序列数据做了优化，所以在存储和查询效率上，比其他方案要高得多。另外，Redis TimeSeries 提供了许多方便的功能，比如：

自动聚合 (Aggregation): 可以按照指定的时间间隔对数据进行聚合，比如计算平均值、最大值、最小值等等。
数据保留策略 (Retention Policy): 可以自动删除旧的数据，避免数据量无限增长。
灵活的查询 (Querying): 可以按照时间范围、标签等条件进行查询。

开始使用 Redis TimeSeries

要使用 Redis TimeSeries，首先需要安装 RedisTimeSeries 模块。如果你用的是 Redis Stack，那它已经自带了 TimeSeries 模块，可以直接使用。如果你用的是普通的 Redis，需要手动安装 RedisTimeSeries 模块。具体安装方法可以参考 RedisTimeSeries 的官方文档。

安装好之后，就可以开始使用了。假设我们有一个传感器，每秒钟产生一个温度值。我们用 Python 来模拟这个传感器，并将数据存储到 Redis TimeSeries 中。

import redis
import time
import random

# 连接 Redis
r = redis.Redis(host='localhost', port=6379)

# TimeSeries 的 key
ts_key = 'temperature:sensor1'

# 创建 TimeSeries (如果不存在)
try:
    r.ts().create(ts_key)
except redis.exceptions.ResponseError as e:
    if "TSDB: key already exists" in str(e):
        print(f"TimeSeries key '{ts_key}' already exists.")
    else:
        raise e

# 模拟传感器数据
while True:
    timestamp = int(time.time() * 1000)  # 毫秒级别的时间戳
    temperature = random.uniform(20, 30)  # 随机生成温度值

    # 添加数据到 TimeSeries
    r.ts().add(ts_key, timestamp, temperature)

    print(f"Timestamp: {timestamp}, Temperature: {temperature}")

    time.sleep(1)  # 每隔 1 秒钟产生一个数据

这段代码会不断地生成温度数据，并将数据存储到 Redis TimeSeries 中。 r.ts().add(ts_key, timestamp, temperature) 就是往 TimeSeries 中添加数据的关键语句。

查询数据

存进去的数据，当然要能查出来。 Redis TimeSeries 提供了多种查询方式。

按照时间范围查询:

# 查询最近 10 秒钟的数据
end_time = int(time.time() * 1000)
start_time = end_time - 10000

data = r.ts().range(ts_key, start_time, end_time)

print(data)

这段代码会查询最近 10 秒钟的温度数据。 r.ts().range(ts_key, start_time, end_time) 就是按照时间范围查询数据的语句。返回的结果是一个列表，每个元素是一个元组，包含时间戳和对应的值。

使用 mrange 命令进行多时间序列查询:

如果想同时查询多个时间序列，可以使用 mrange 命令。比如，我们有多个传感器的温度数据，存储在不同的 TimeSeries 中：temperature:sensor1, temperature:sensor2, temperature:sensor3。

# 查询多个传感器最近 10 秒钟的数据
end_time = int(time.time() * 1000)
start_time = end_time - 10000

data = r.ts().mrange(start_time, end_time, filters=['sensor=sensor*']) # 使用filter来匹配键

print(data)

r.ts().mrange(start_time, end_time, filters=['sensor=sensor*']) 这条命令会查询所有以 temperature:sensor 开头的 TimeSeries 中，最近 10 秒钟的数据。 filters 参数用于过滤 TimeSeries 的 key。

聚合 (Aggregation)

Redis TimeSeries 最强大的功能之一就是聚合。它可以按照指定的时间间隔，对数据进行聚合计算。比如，我们可以计算每分钟的平均温度：

# 计算每分钟的平均温度
end_time = int(time.time() * 1000)
start_time = end_time - 60000  # 一分钟前

data = r.ts().range(ts_key, start_time, end_time, aggregation_type='avg', time_bucket=60000)

print(data)

r.ts().range(ts_key, start_time, end_time, aggregation_type='avg', time_bucket=60000) 这条命令会计算过去一分钟内，每分钟的平均温度。 aggregation_type 参数指定了聚合的类型，这里是 avg，表示计算平均值。 time_bucket 参数指定了时间间隔，这里是 60000 毫秒，表示一分钟。

除了平均值，还可以计算其他的聚合类型，比如：

min: 最小值
max: 最大值
sum: 总和
count: 数量
stddev: 标准差
var: 方差
first: 第一个值
last: 最后一个值

数据保留策略 (Retention Policy)

时间序列数据往往会不断增长，如果不加以控制，很快就会占满存储空间。 Redis TimeSeries 提供了数据保留策略，可以自动删除旧的数据。

# 创建 TimeSeries，并设置数据保留时间为 1 天
try:
    r.ts().create(ts_key, retention_msecs=86400000) # 1天 = 86400000 毫秒
except redis.exceptions.ResponseError as e:
    if "TSDB: key already exists" in str(e):
        print(f"TimeSeries key '{ts_key}' already exists.")
    else:
        raise e

r.ts().create(ts_key, retention_msecs=86400000) 这条命令会创建一个 TimeSeries，并设置数据保留时间为 1 天。超过 1 天的数据会自动被删除。

总结

Redis TimeSeries 是一个非常强大的时间序列数据库，特别适合存储和处理传感器数据、日志数据等。它具有以下优点：

速度快： 基于内存存储，读写速度极快。
功能丰富： 提供了自动聚合、数据保留策略、灵活的查询等功能。
易于使用： 提供了简单易用的 API。

实际应用场景

物联网 (IoT): 存储和分析传感器数据，比如温度、湿度、压力等等。
监控系统： 存储和分析服务器的 CPU 使用率、内存使用率、网络流量等等。
金融分析： 存储和分析股票价格、交易量等等。
日志分析： 存储和分析应用程序的日志数据。

一些建议

合理选择数据类型： TimeSeries 支持多种数据类型，比如整数、浮点数等等。选择合适的数据类型可以节省存储空间。
合理设置数据保留策略： 根据实际需求，设置合适的数据保留时间。
使用标签 (Labels)： 使用标签可以方便地对 TimeSeries 进行分类和查询。
善用聚合功能： 聚合功能可以大大减少需要存储的数据量，提高查询效率。

高级用法

除了上面介绍的基本用法，Redis TimeSeries 还提供了许多高级功能，比如：

压缩 (Compression): 可以对数据进行压缩，节省存储空间。
多级索引 (Multi-level Indexing): 可以创建多级索引，提高查询效率。
规则引擎 (Rule Engine): 可以根据时间序列数据触发一些规则，比如发送警报等等。

这些高级功能需要更深入的了解 Redis TimeSeries 的原理和使用方法。如果你有兴趣，可以参考 RedisTimeSeries 的官方文档。

RedisTimeSeries 命令速查表

命令	描述	示例
`TS.CREATE key [RETENTION time]`	创建一个新的时间序列。`RETENTION` 指定保留时间（毫秒）。	`TS.CREATE temperature:sensor1 RETENTION 86400000` (保留一天)
`TS.ADD key timestamp value`	向时间序列添加一个数据点。时间戳可以是 `*` (表示当前时间)。	`TS.ADD temperature:sensor1 * 25.5` (添加当前时间戳的温度值)
`TS.MADD key timestamp value [key timestamp value ...]`	一次添加多个时间序列的数据点。	`TS.MADD temperature:sensor1 * 25.5 temperature:sensor2 * 26.0`
`TS.RANGE key start end [COUNT count] [AGGREGATION type timeBucket]`	查询指定时间范围内的数据。`start` 和 `end` 可以是时间戳或 `-` (表示最小值) 和 `+` (表示最大值)。`COUNT` 限制返回结果的数量。`AGGREGATION` 指定聚合类型和时间桶。	`TS.RANGE temperature:sensor1 - + COUNT 10` (查询所有数据，最多返回 10 个点) `TS.RANGE temperature:sensor1 - + AGGREGATION avg 60000` (查询所有数据，按分钟平均)
`TS.REVRANGE key start end [COUNT count] [AGGREGATION type timeBucket]`	与 `TS.RANGE` 类似，但按时间倒序返回结果。	`TS.REVRANGE temperature:sensor1 - + COUNT 10` (查询所有数据，最多返回 10 个点，倒序)
`TS.MRANGE start end [COUNT count] [AGGREGATION type timeBucket] WITHLABELS FILTER filter...`	查询多个时间序列，通过 `FILTER` 指定过滤条件。`WITHLABELS` 返回时间序列的标签。	`TS.MRANGE - + FILTER sensor=sensor* WITHLABELS` (查询所有传感器的数据)
`TS.MREVRANGE start end [COUNT count] [AGGREGATION type timeBucket] WITHLABELS FILTER filter...`	与 `TS.MRANGE` 类似，但按时间倒序返回结果。	`TS.MREVRANGE - + FILTER sensor=sensor* WITHLABELS` (查询所有传感器的数据，倒序)
`TS.GET key`	获取时间序列的最新（最后）数据点。	`TS.GET temperature:sensor1` (获取传感器 1 的最新温度)
`TS.MGET WITHLABELS FILTER filter...`	获取多个时间序列的最新数据点，通过 `FILTER` 指定过滤条件。`WITHLABELS` 返回时间序列的标签。	`TS.MGET WITHLABELS FILTER sensor=sensor*` (获取所有传感器的最新温度)
`TS.INFO key`	获取时间序列的元数据信息（例如保留时间，数据块大小）。	`TS.INFO temperature:sensor1` (获取传感器 1 的信息)
`TS.ALTER key RETENTION time`	修改时间序列的元数据（例如修改保留时间）。	`TS.ALTER temperature:sensor1 RETENTION 172800000` (修改传感器 1 的保留时间为两天)
`TS.DEL key start end`	删除指定时间范围内的数据点。	`TS.DEL temperature:sensor1 1666886400000 1666890000000` (删除指定时间范围内的温度数据)
`TS.CREATERULE sourceKey destKey AGGREGATION type timeBucket`	创建一个聚合规则，将 `sourceKey` 的数据按照指定的聚合类型和时间桶聚合后存储到 `destKey` 中。	`TS.CREATERULE temperature:sensor1 temperature:sensor1:avg-minute AGGREGATION avg 60000` (创建聚合规则)
`TS.DELETERULE sourceKey destKey`	删除一个聚合规则。	`TS.DELETERULE temperature:sensor1 temperature:sensor1:avg-minute` (删除聚合规则)
`TS.ADD key * value DUPLICATE_POLICY policy`	添加数据点，并处理重复数据。 `policy` 可以是 `BLOCK`, `FIRST`, `LAST`, `MIN`, `MAX`。 `BLOCK` 会阻止添加重复数据，其他policy会选择性替代。	`TS.ADD temperature:sensor1 * 25.5 DUPLICATE_POLICY MAX` (添加当前时间戳的温度值，重复值取最大)

希望今天的分享能帮助大家更好地理解和使用 Redis TimeSeries。记住，技术是为了解决问题的，只要掌握了正确的方法，就能让我们的工作事半功倍。谢谢大家！

发表回复 取消回复

发表回复取消回复