# 时序数据库

本系统中的时序数据库基于 PostgreSQL 的 Timescale 扩 展 (opens new window),可用于存储工业控制、IOT 等设备的运行时状态数据, 或存储其他场景的海量时序数据。

# 参考说明

本文档参考自 https://docs.timescale.com/getting-started/latest/create-hypertable/ (opens new window)

# 目标读者

本文档的目标读者为:需要使用本系统存储大量时序数据的的开发人员

# 关于 Timescale 扩展

Timescale 扩展了 PostgreSQL 的时间序列和分析功能,因此您可以更快地构建应用,更好 地扩展应用场景。

# 运行 TimescaleDB

我们正在使用带有 Timescale 扩展的 PostgreSQL 镜像,更详细信息可可参考 https://docs.timescale.com/self-hosted/latest/install/installation-docker/ (opens new window)

  • Docker 镜像: https://hub.docker.com/r/timescale/timescaledb/ (opens new window) 当前版本是 14.7, 以下是示例的 docker-compose 配置片段

     database:
       image: timescale/timescaledb-ha:pg14-latest
       restart: always
       environment:
         - POSTGRES_DATABASE=platform
         - POSTGRES_USER=postgres
         - POSTGRES_PASSWORD=password
       volumes:
         - ${PWD}/runtime/database/data:/var/lib/postgresql/data
       healthcheck:
         test: ['CMD-SHELL', 'pg_isready -U postgres']
         interval: 10s
         timeout: 5s
         retries: 5
       ports: 
         - 5432:5432
       networks:
         - scm
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18

可以直接从 https://github.com/muyantech/grails-docker (opens new window) 下载完整的 docker-compose 配置文件,然后运行 docker compose up -d database redis pgadmin 即可。

注意

如果当前已经在使用 grails-docker 项目,在将原 postgresql 镜像切换为 timescale/timescaledb-ha 镜像前,请备份数据目录 ${PWD}/runtime/database/data

# 创建 TimescaleDB 扩展

在 PostgreSQL 中创建 TimescaleDB 扩展,可以使用以下命令:

-- 如果还没有创建数据,先创建数据库
CREATE DATABASE platform WITH ENCODING='UTF8' OWNER=postgres;

-- 连接到数据库
\c platform

-- 创建 timescaledb 扩展
CREATE EXTENSION IF NOT EXISTS timescaledb;

-- 显示活动扩展以确认
\dx
1
2
3
4
5
6
7
8
9
10
11

# 使用 Timescale 扩展创建表

TimescaleDB 扩展提供了一个特殊的函数 create_hypertable,用于将现有表转换为超表,


-- 创建表
CREATE TABLE stocks_real_time (
  time TIMESTAMPTZ NOT NULL,
  symbol TEXT NOT NULL,
  price DOUBLE PRECISION NULL,
  day_volume INT NULL
);

-- 创建 hypertable, 将 stocks_real_time 转换为时序表, time 列为时序列
SELECT create_hypertable('stocks_real_time','time');

-- 如果表已经存在,可以使用以下命令迁移到时序表
SELECT create_hypertable('stocks_real_time','time', migrate_data => true);

-- 基于 symbol 和 time 列创建索引
CREATE INDEX ix_symbol_time ON stocks_real_time (symbol, time DESC);

-- 创建另一个普通表
CREATE TABLE company (
  symbol TEXT NOT NULL,
  name TEXT NOT NULL
);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 查询数据

TimeScaleDB 使用标准的 PostgreSQL 语法进行查询,除此之外, TimescaleDB 还扩展提供 了一些特殊的函数,用于查询时序数据,例如 time_bucket 函数等。

# 插入测试数据

  1. 准备一些测试数据, 从以下地址下载样本数据 https://assets.timescale.com/docs/downloads/get-started/real_time_stock_data.zip (opens new window) 使用命令 unzip real_time_stock_data.zip 解压缩

  2. 将示例数据导入数据库:转到正在运行的 Timescale Docker 容器,并进入 psql 控制台

# 将示例数据复制到正在运行的 Timescale Docker 容器中
docker cp tutorial_sample_tick.csv <running_container_id>:/tutorial_sample_tick.csv
docker cp tutorial_sample_company.csv <running_container_id>:/tutorial_sample_company.csv
# 进入 psql 控制台
docker exec -it <running_container_id> bash
psql -U postgres
1
2
3
4
5
6
-- 将 CSV 文件导入 stocks_real_time 和 company 表
\COPY stocks_real_time from './tutorial_sample_tick.csv' CSV HEADER;
\COPY company from './tutorial_sample_company.csv' CSV HEADER;
1
2
3

# 使用原始 SQL 查询数据

-- 选择过去四天的所有股票数据
SELECT * FROM stocks_real_time srt
 WHERE time > now() - INTERVAL '4 days';

-- 按顺序选择 Amazon 最近的 10 笔交易
SELECT * FROM stocks_real_time srt
 WHERE symbol='AMZN'
 ORDER BY time DESC, day_volume desc
 LIMIT 10;

-- 计算过去四天内 Apple 的平均交易价格
SELECT
  avg(price)
  FROM stocks_real_time srt
       JOIN company c ON c.symbol = srt.symbol
 WHERE c.name = 'Apple' AND time > now() - INTERVAL '4 days';
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 高级 Timescale SQL 函数

参考 https://docs.timescale.com/getting-started/latest/query-data/ (opens new window)

  • first():在聚合组内基于时间找到最早的值
  • last():在聚合组内基于时间找到最新的值
  • time_bucket():按任意时间间隔分桶数据并计算这些间隔内的聚合值
-- 以下是在股票交易数据中使用 time_bucket() 的示例,可以计算过去一周内每个交易代码的每日平均价格。

SELECT
  time_bucket('1 day', time) AS bucket,
  symbol,
  avg(price)
  FROM stocks_real_time srt
 WHERE time > now() - INTERVAL '1 week'
 GROUP BY bucket, symbol
 ORDER BY bucket, symbol;
1
2
3
4
5
6
7
8
9
10

# 持续聚合

TimescaleDB 提供了一种称为持续聚合的功能,可以在后台自动计算聚合数据,而无需用户 干预。这使得用户可以在查询聚合数据时获得更快的响应时间,而不会影响原始数据的写入 速度。

在创建物化视图时,可以使用 WITH (timescaledb.continuous) 选项来指示创建持续聚合。

# 创建持续聚合


--- 查询每天的股票交易数据中的最高价、最低价、开盘价和收盘价
SELECT
  time_bucket('1 day', "time") AS day,
  symbol,
  max(price) AS high,
  first(price, time) AS open,
  last(price, time) AS close,
  min(price) AS low
FROM stocks_real_time srt
GROUP BY day, symbol
ORDER BY day DESC, symbol;

-- 使用上述 SQL 创建持续聚合,将结果存储在 stock_candlestick_daily 视图中
-- 请注意,我们使用 WITH (timescaledb.continuous) 选项来指示创建持续聚合
CREATE MATERIALIZED VIEW stock_candlestick_daily
WITH (timescaledb.continuous) AS
SELECT
  time_bucket('1 day', "time") AS day,
  symbol,
  max(price) AS high,
  first(price, time) AS open,
  last(price, time) AS close,
  min(price) AS low
FROM stocks_real_time srt
GROUP BY day, symbol;
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

# Raw SQL 查询

如果在开发过程中,需要使用原始 SQL 进行数据查询,可以使用以下方法:

  1. 在 Service 或者 Controller 中,注入 javax.sql.DataSource, 或者直接使用 BeanHelper.getBean(application, "dataSource") 获取
  2. 使用 groovy.sql.Sql.rows 或其他方法进行查询

Sql 类的详细 API 文档请参考:https://docs.groovy-lang.org/latest/html/api/groovy/sql/Sql.html (opens new window)

示例如下:

import javax.sql.DataSource
import groovy.sql.Sql

Class StockService {

    # 注入 DataSource
    DataSource dataSource

    def getStockCandlestickDaily() {
        Sql sql = new Sql(dataSource)
        String sqlString = "SELECT * FROM stock_candlestick_daily"
        List<GroovyRowResult> list = sql.rows(sqlString)        
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 进一步阅读

以上只是 TimescaleDB 的一个简单入门介绍,更多详细功能请参考以下官方文档链接:

最后更新: 2024/2/23 07:44:20