Python工具箱系列(三十三)

Timescaledb

在物聯網時代，出現了大量以時間為中心海量產生的感測器數據，稱為時序數據。這類數據的特點是：

• 數據記錄總有一個時間戳。

• 數據幾乎總是追加，不更新也不刪除。

• 大量使用近期的數據。很少更新或者回填時間間隔的缺失數據。

• 與時間間隔頻率關係不大。但累積的數據量大，可能會有峰值。

• 對這類數據有多種聚合查詢的需求，並且越快越好。例如，截止到目前為止，最大值/最小值/平均值是多少，數據流速是多少等。

為此，IT界興起了時序資料庫。TimeScaleDB是其中的佼佼者，截止到2022年7月，它的排名在第5名，值得使用。由於TimeScaleDB是postgresql的一個插件，因此非常便於安裝與使用。同時，它也是一個開源的時間序列資料庫，為快速獲取和複雜查詢進行了優化。此外，它也是多模型設計，在體現與時序數據相關的特性外，它執行的是“完整的SQL”，程式員很容易使用與管理它。

它的安裝不複雜。使用以下命令在ubuntu bionic下安裝單機版本。

apt install -y gnupg postgresql-common apt-transport-https lsb-release wget
/usr/share/postgresql-common/pgdg/apt.postgresql.org.sh
echo "deb https://packagecloud.io/timescale/timescaledb/ubuntu/ $(lsb_release -c -s) main" > /etc/apt/sources.list.d/timescaledb.list
wget --quiet -O - https://packagecloud.io/timescale/timescaledb/gpgkey | apt-key add -
apt update
apt install -y timescaledb-2-postgresql-14
# 做一個調整
timescaledb-tune --quiet --yes
# 重新啟動資料庫服務
systemctl restart postgresql

# 以postgres用戶啟動命令行
su postgres -c psql

# 在psql命令行環境中。輸入以下命令，從而關聯到timescaledb這個擴展上。
CREATE database sensordb;
\c sensordb
CREATE EXTENSION IF NOT EXISTS timescaledb;
\q

# 再次連接
su postgres -c 'psql -d sensordb'
# 顯示擴展列表(extensions)
\dx

一、創建時序相關得表

在sensordb下創建測試用的表，這個創建的過程有些特殊。相關命令如下：

# 感測器表，傳統的表
CREATE TABLE sensors(
  id SERIAL PRIMARY KEY,
  type VARCHAR(50),
  location VARCHAR(50)
);

# 感測器資料庫，這個將轉換成為超表
CREATE TABLE sensor_data (
  time TIMESTAMP NOT NULL,
  sensor_id INTEGER,
  pm25 DOUBLE PRECISION,
  temperature DOUBLE PRECISION,
  FOREIGN KEY (sensor_id) REFERENCES sensors (id)
);

# CREATE EXTENSION IF NOT EXISTS timescaledb;

# 轉換為超表
SELECT create_hypertable('sensor_data', 'time');

# 生成4個感測器
INSERT INTO sensors (type, location) VALUES
('a','地板'),
('a', '天花板'),
('b','地板'),
('b', '天花板');

# 測試一下。
select * from sensors;

接下來，使用python連接時序資料庫，並且模擬相關的數據插入到表中。

二、使用Python模擬數據

import psycopg2
import random
import datetime

# 事先創建後資料庫demodb
demodb = psycopg2.connect(database="sensordb", user="postgres",
                          password="88488848", host="172.17.2.151", port="5432")

democur = demodb.cursor()

currenttime = datetime.datetime.now()

# 插入模擬出來的數據。
for _ in range(100000):
    currenttime = currenttime+datetime.timedelta(seconds=1)
    for id in range(1,5,1):
        pm25 = random.uniform(0, 300)
        temp = random.uniform(0, 40)
        insertsql = f'''insert into sensor_data(sensor_id,pm25,temperature,time) values({id},{pm25},{temp},'{currenttime}')'''
        democur.execute(insertsql)
    demodb.commit()

democur.close()
demodb.close()

這裡插入10萬秒的數據，相當於100000/86400=1.15(天)的數據。在插入數據的同時，就可以同時在資料庫中進行按30分鐘的分桶查詢，這是時序資料庫的一個特殊功能。

# su postgres -c 'psql -d sensordb'

SELECT
  time_bucket('30 minutes', time) AS period,
  AVG(temperature) AS avg_temp,
  last(temperature, time) AS last_temp,
  AVG(pm25) AS avg_pm25
FROM sensor_data
GROUP BY period;

SELECT
  time_bucket('60 minutes', time) AS period,
  AVG(temperature) AS avg_temp,
  last(temperature, time) AS last_temp,
  AVG(pm25) AS avg_pm25
FROM sensor_data
GROUP BY period;

此時，按30分鐘時間視窗聚會的數據查詢效果如下圖所示：

可以看出，TimeScaleDB已經將數據按30分鐘來聚合分析。當然，改成任意時間也是可以的，例如，可以改成5秒分析一次也可以，生成結果的時間也非常快。