TimescaleDB 時序資料庫完全指南：從 Hypertable 到連續聚合與壓縮策略 | PostgreSQL

# 加入 TimescaleDB APT 來源
sudo apt install -y gnupg postgresql-common
sudo /usr/share/postgresql-common/pgdg/apt.postgresql.org.sh
sudo sh -c "echo 'deb https://packagecloud.io/timescale/timescaledb/ubuntu/ $(lsb_release -cs) main' > /etc/apt/sources.list.d/timescaledb.list"
wget --quiet -O - https://packagecloud.io/timescale/timescaledb/gpgkey | sudo gpg --dearmor -o /etc/apt/trusted.gpg.d/timescaledb.gpg
sudo apt update

# 安裝（PostgreSQL 16 版本）
sudo apt install -y timescaledb-2-postgresql-16

# 執行調校精靈
sudo timescaledb-tune --quiet --yes

# 重啟 PostgreSQL
sudo systemctl restart postgresql

docker run -d --name timescaledb \
  -p 5432:5432 \
  -e POSTGRES_PASSWORD=password \
  timescale/timescaledb:latest-pg16

-- 在目標資料庫中啟用
CREATE EXTENSION IF NOT EXISTS timescaledb;

-- 確認版本
SELECT extversion FROM pg_extension WHERE extname = 'timescaledb';
-- 2.17.2

-- Step 1：建立普通表
CREATE TABLE sensor_data (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT        NOT NULL,
    temperature DOUBLE PRECISION,
    humidity    DOUBLE PRECISION,
    battery     DOUBLE PRECISION
);

-- Step 2：轉換為 Hypertable
-- chunk_time_interval 決定每個 Chunk 涵蓋的時間範圍
SELECT create_hypertable(
    'sensor_data',
    by_range('time', INTERVAL '1 day')
);

SELECT create_hypertable(
    'sensor_data',
    by_range('time', INTERVAL '1 day'),
    by_hash('device_id', 4)  -- 4 個 Hash 分區
);

-- 列出所有 Chunk
SELECT chunk_name, range_start, range_end,
       pg_size_pretty(total_bytes) AS size
FROM timescaledb_information.chunks
WHERE hypertable_name = 'sensor_data'
ORDER BY range_start DESC
LIMIT 10;

-- 每 5 分鐘平均溫度
SELECT
    time_bucket('5 minutes', time) AS bucket,
    device_id,
    AVG(temperature) AS avg_temp,
    MAX(temperature) AS max_temp,
    MIN(temperature) AS min_temp,
    COUNT(*) AS readings
FROM sensor_data
WHERE time > NOW() - INTERVAL '1 hour'
GROUP BY bucket, device_id
ORDER BY bucket DESC;

-- 月度統計（使用 INTERVAL）
SELECT
    time_bucket('1 month', time) AS month,
    COUNT(*) AS total_readings,
    AVG(temperature) AS avg_temp
FROM sensor_data
GROUP BY month
ORDER BY month;

-- 帶偏移量的時間桶（工廠三班制：08:00 開始）
SELECT
    time_bucket('8 hours', time, '2026-01-01 08:00:00'::timestamptz) AS shift,
    AVG(temperature) AS avg_temp
FROM sensor_data
GROUP BY shift
ORDER BY shift;

-- 近似行數（比 COUNT(*) 快 10-100 倍）
SELECT approximate_row_count('sensor_data');

-- 第一筆與最後一筆
SELECT
    device_id,
    first(temperature, time) AS first_temp,
    last(temperature, time)  AS last_temp
FROM sensor_data
GROUP BY device_id;

-- 時間加權平均（處理不規則取樣間隔）
SELECT
    time_bucket('1 hour', time) AS bucket,
    average(time_weight('linear', time, temperature)) AS tw_avg
FROM sensor_data
WHERE time > NOW() - INTERVAL '24 hours'
GROUP BY bucket
ORDER BY bucket;

-- 建立連續聚合：每小時設備統計
CREATE MATERIALIZED VIEW sensor_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS bucket,
    device_id,
    AVG(temperature)   AS avg_temp,
    MAX(temperature)   AS max_temp,
    MIN(temperature)   AS min_temp,
    AVG(humidity)      AS avg_humidity,
    COUNT(*)           AS readings
FROM sensor_data
GROUP BY bucket, device_id
WITH NO DATA;  -- 先不回填歷史資料

-- 自動刷新：每 30 分鐘刷新，範圍為 3 小時前到 1 小時前
SELECT add_continuous_aggregate_policy(
    'sensor_hourly',
    start_offset    => INTERVAL '3 hours',
    end_offset      => INTERVAL '1 hour',
    schedule_interval => INTERVAL '30 minutes'
);

-- 回填過去 30 天
CALL refresh_continuous_aggregate(
    'sensor_hourly',
    NOW() - INTERVAL '30 days',
    NOW() - INTERVAL '1 hour'
);

-- 日粒度（基於小時粒度的 sensor_hourly）
CREATE MATERIALIZED VIEW sensor_daily
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 day', bucket) AS day,
    device_id,
    AVG(avg_temp)    AS avg_temp,
    MAX(max_temp)    AS max_temp,
    MIN(min_temp)    AS min_temp,
    SUM(readings)    AS total_readings
FROM sensor_hourly
GROUP BY day, device_id
WITH NO DATA;

SELECT add_continuous_aggregate_policy(
    'sensor_daily',
    start_offset    => INTERVAL '3 days',
    end_offset      => INTERVAL '1 day',
    schedule_interval => INTERVAL '1 hour'
);

-- 啟用壓縮
ALTER TABLE sensor_data SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id',
    timescaledb.compress_orderby = 'time DESC'
);

-- 超過 7 天的 Chunk 自動壓縮
SELECT add_compression_policy(
    'sensor_data',
    compress_after => INTERVAL '7 days'
);

-- 壓縮統計
SELECT
    pg_size_pretty(before_compression_total_bytes) AS before,
    pg_size_pretty(after_compression_total_bytes)  AS after,
    ROUND(
        (1 - after_compression_total_bytes::numeric /
             before_compression_total_bytes::numeric) * 100, 1
    ) AS compression_ratio_pct
FROM hypertable_compression_stats('sensor_data');

-- 壓縮特定時間範圍
SELECT compress_chunk(c.chunk_name)
FROM timescaledb_information.chunks c
WHERE c.hypertable_name = 'sensor_data'
  AND c.range_end < NOW() - INTERVAL '7 days'
  AND NOT c.is_compressed;

-- 自動刪除 90 天前的原始資料
SELECT add_retention_policy(
    'sensor_data',
    drop_after => INTERVAL '90 days'
);

-- 手動刪除（需要時）
SELECT drop_chunks(
    'sensor_data',
    older_than => INTERVAL '90 days'
);

-- 為聚合也設定保留策略
SELECT add_retention_policy('sensor_hourly', drop_after => INTERVAL '180 days');
-- sensor_daily 永久保留，不設保留策略

-- 完整的 IoT 資料模型
CREATE TABLE devices (
    device_id   TEXT PRIMARY KEY,
    location    TEXT,
    device_type TEXT,
    installed   TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE sensor_readings (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT        NOT NULL REFERENCES devices(device_id),
    temperature DOUBLE PRECISION,
    humidity    DOUBLE PRECISION,
    pressure    DOUBLE PRECISION,
    battery     DOUBLE PRECISION
);

SELECT create_hypertable('sensor_readings', by_range('time', INTERVAL '1 day'));

-- 建立索引（Chunk 內的局部索引）
CREATE INDEX idx_readings_device_time
    ON sensor_readings (device_id, time DESC);

-- 啟用壓縮
ALTER TABLE sensor_readings SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id',
    timescaledb.compress_orderby = 'time DESC'
);

SELECT add_compression_policy('sensor_readings', compress_after => INTERVAL '7 days');
SELECT add_retention_policy('sensor_readings', drop_after => INTERVAL '90 days');

-- 偵測溫度異常：超過該設備近 24 小時平均值 3 個標準差
WITH device_stats AS (
    SELECT
        device_id,
        AVG(temperature) AS avg_temp,
        STDDEV(temperature) AS std_temp
    FROM sensor_readings
    WHERE time > NOW() - INTERVAL '24 hours'
    GROUP BY device_id
)
SELECT
    r.time,
    r.device_id,
    r.temperature,
    s.avg_temp,
    ROUND((r.temperature - s.avg_temp) / NULLIF(s.std_temp, 0), 2) AS z_score
FROM sensor_readings r
JOIN device_stats s ON r.device_id = s.device_id
WHERE r.time > NOW() - INTERVAL '1 hour'
  AND ABS(r.temperature - s.avg_temp) > 3 * s.std_temp
ORDER BY ABS(r.temperature - s.avg_temp) DESC;

-- 電量低於 20% 且持續下降的設備
SELECT
    device_id,
    last(battery, time) AS current_battery,
    first(battery, time) AS battery_1h_ago,
    last(battery, time) - first(battery, time) AS battery_change
FROM sensor_readings
WHERE time > NOW() - INTERVAL '1 hour'
GROUP BY device_id
HAVING last(battery, time) < 20
   AND last(battery, time) < first(battery, time)
ORDER BY current_battery ASC;

-- APM 指標表
CREATE TABLE app_metrics (
    time         TIMESTAMPTZ NOT NULL,
    service      TEXT        NOT NULL,
    endpoint     TEXT        NOT NULL,
    status_code  INT,
    response_ms  DOUBLE PRECISION,
    error        BOOLEAN DEFAULT FALSE
);

SELECT create_hypertable('app_metrics', by_range('time', INTERVAL '1 day'));

-- 連續聚合：每分鐘的服務健康摘要
CREATE MATERIALIZED VIEW service_health_1m
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 minute', time)  AS bucket,
    service,
    COUNT(*)                       AS total_requests,
    COUNT(*) FILTER (WHERE error)  AS error_count,
    ROUND(
        100.0 * COUNT(*) FILTER (WHERE error) / COUNT(*), 2
    )                              AS error_rate,
    PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY response_ms) AS p50,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY response_ms) AS p95,
    PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY response_ms) AS p99
FROM app_metrics
GROUP BY bucket, service
WITH NO DATA;

SELECT add_continuous_aggregate_policy(
    'service_health_1m',
    start_offset    => INTERVAL '10 minutes',
    end_offset      => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute'
);

-- 過去 1 小時的服務 SLI 摘要
SELECT
    service,
    SUM(total_requests) AS requests,
    ROUND(AVG(error_rate), 2) AS avg_error_rate,
    ROUND(AVG(p50), 1) AS avg_p50_ms,
    ROUND(AVG(p95), 1) AS avg_p95_ms,
    ROUND(AVG(p99), 1) AS avg_p99_ms,
    CASE
        WHEN AVG(error_rate) < 0.1 AND AVG(p99) < 500 THEN 'HEALTHY'
        WHEN AVG(error_rate) < 1.0 AND AVG(p99) < 1000 THEN 'DEGRADED'
        ELSE 'CRITICAL'
    END AS status
FROM service_health_1m
WHERE bucket > NOW() - INTERVAL '1 hour'
GROUP BY service
ORDER BY avg_error_rate DESC;

-- 逐筆交易表
CREATE TABLE trades (
    time    TIMESTAMPTZ      NOT NULL,
    symbol  TEXT             NOT NULL,
    price   DOUBLE PRECISION NOT NULL,
    volume  DOUBLE PRECISION NOT NULL
);

SELECT create_hypertable('trades', by_range('time', INTERVAL '1 day'));

-- 1 分鐘 K 線連續聚合
CREATE MATERIALIZED VIEW ohlcv_1m
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 minute', time) AS bucket,
    symbol,
    first(price, time)  AS open,
    MAX(price)          AS high,
    MIN(price)          AS low,
    last(price, time)   AS close,
    SUM(volume)         AS volume
FROM trades
GROUP BY bucket, symbol
WITH NO DATA;

SELECT add_continuous_aggregate_policy(
    'ohlcv_1m',
    start_offset    => INTERVAL '5 minutes',
    end_offset      => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute'
);

-- 在 1 分鐘 K 線上建構 1 小時 K 線
CREATE MATERIALIZED VIEW ohlcv_1h
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', bucket)  AS bucket,
    symbol,
    first(open, bucket)  AS open,
    MAX(high)            AS high,
    MIN(low)             AS low,
    last(close, bucket)  AS close,
    SUM(volume)          AS volume
FROM ohlcv_1m
GROUP BY time_bucket('1 hour', bucket), symbol
WITH NO DATA;

SELECT add_continuous_aggregate_policy(
    'ohlcv_1h',
    start_offset    => INTERVAL '3 hours',
    end_offset      => INTERVAL '1 hour',
    schedule_interval => INTERVAL '30 minutes'
);

-- 5 日與 20 日移動平均線
SELECT
    bucket AS date,
    symbol,
    close,
    AVG(close) OVER (
        PARTITION BY symbol ORDER BY bucket
        ROWS BETWEEN 4 PRECEDING AND CURRENT ROW
    ) AS ma5,
    AVG(close) OVER (
        PARTITION BY symbol ORDER BY bucket
        ROWS BETWEEN 19 PRECEDING AND CURRENT ROW
    ) AS ma20
FROM ohlcv_1h
WHERE symbol = 'AAPL'
  AND bucket > NOW() - INTERVAL '30 days'
ORDER BY bucket;

-- 記憶體配置（以 64GB RAM 為例）
ALTER SYSTEM SET shared_buffers = '16GB';
ALTER SYSTEM SET effective_cache_size = '48GB';
ALTER SYSTEM SET work_mem = '256MB';
ALTER SYSTEM SET maintenance_work_mem = '2GB';

-- WAL 調整（高寫入場景）
ALTER SYSTEM SET wal_buffers = '64MB';
ALTER SYSTEM SET max_wal_size = '8GB';
ALTER SYSTEM SET checkpoint_completion_target = 0.9;

-- TimescaleDB 專用
ALTER SYSTEM SET timescaledb.max_background_workers = 8;

SELECT pg_reload_conf();

import psycopg2
from psycopg2.extras import execute_values

conn = psycopg2.connect("dbname=iot user=admin")
cur = conn.cursor()

# 使用 execute_values 批次插入（比逐筆 INSERT 快 10-50 倍）
data = [
    ('2026-07-21 10:00:00+08', 'sensor-001', 25.3, 60.1, 1013.2, 95.0),
    ('2026-07-21 10:00:01+08', 'sensor-002', 24.8, 62.5, 1013.1, 87.0),
    # ... 數千筆
]

execute_values(
    cur,
    """INSERT INTO sensor_readings
       (time, device_id, temperature, humidity, pressure, battery)
       VALUES %s""",
    data,
    page_size=1000  # 每批 1000 筆
)
conn.commit()

# 從 CSV 高速載入（最快的方式）
timescaledb-parallel-copy \
    --connection "host=localhost dbname=iot user=admin" \
    --table sensor_readings \
    --file sensor_data.csv \
    --workers 4 \
    --batch-size 10000

-- 各 Hypertable 的 Chunk 統計
SELECT
    hypertable_name,
    COUNT(*) AS num_chunks,
    COUNT(*) FILTER (WHERE is_compressed) AS compressed_chunks,
    pg_size_pretty(SUM(total_bytes)) AS total_size
FROM timescaledb_information.chunks
GROUP BY hypertable_name;

-- 背景排程任務狀態
SELECT *
FROM timescaledb_information.jobs
WHERE application_name LIKE 'User-Defined%'
   OR proc_name IN ('policy_compression', 'policy_retention', 'policy_refresh_continuous_aggregate')
ORDER BY next_start;

-- 連續聚合刷新狀態
SELECT
    view_name,
    completed_threshold,
    next_start
FROM timescaledb_information.continuous_aggregate_stats;