时序数据库对比实战：从IoT到金融分析的5种生产模式
数据库
时序数据正在吞噬你的关系型数据库？
监控指标每秒写入10万条、IoT传感器每分钟上报百万数据点、金融Tick数据一天产生数十亿行——当你还在用MySQL按时间分区勉强支撑时，查询已经慢到超时，存储成本已经爆炸式增长。2026年，**时序数据库（TSDB）**已经成为处理时间序列数据的标准答案，但InfluxDB、TimescaleDB、TDengine、QuestDB到底怎么选？
本文从5种真实生产场景出发，给出每个场景的最佳数据库选型和完整代码，帮你少走弯路。
时序数据库核心架构对比
| 特性 | InfluxDB 3.0 | TimescaleDB | TDengine | QuestDB |
|---|---|---|---|---|
| 底层引擎 | Apache Arrow + Parquet | PostgreSQL扩展 | 自研存储引擎 | 自研列式引擎 |
| 查询语言 | InfluxQL / SQL | SQL | SQL | SQL |
| 写入协议 | Line Protocol | SQL INSERT | SQL INSERT / Schemaless | ILP / SQL INSERT |
| 压缩比 | 10:1~20:1 | 5:1~10:1 | 10:1~20:1 | 10:1~15:1 |
| 集群模式 | 3.0原生集群 | 依赖PG集群 | 原生集群 | 企业版集群 |
| 开源协议 | Apache 2.0（社区版） | Apache 2.0（社区版） | AGPL-3.0 | Apache 2.0 |
| 适用场景 | 监控/Observability | 金融/复杂分析 | IoT/边缘计算 | 高频写入/实时分析 |
为什么不能只用一个？
- InfluxDB：监控场景生态最强，但复杂SQL分析弱
- TimescaleDB：SQL生态完美，但写入吞吐不如专用TSDB
- TDengine：IoT场景写入极快，但SQL兼容性有限
- QuestDB：高频写入无敌，但生态和工具链不够成熟
Pattern 1：InfluxDB 3.0 监控与可观测性
场景描述
Kubernetes集群500个节点、10万个Pod，Prometheus每15秒采集一次指标，Grafana实时展示。需要支持降采样（downsampling）、数据保留策略（retention policy）、Flux/InfluxQL双查询。
安装与配置
# Docker部署InfluxDB 3.0
docker run -d --name influxdb3 \
-p 8086:8086 \
-p 8181:8181 \
-v influxdb3-data:/var/lib/influxdb3 \
influxdb:3.0-core
# 创建Bucket（数据库）
influx bucket create \
--name monitoring \
--retention 30d \
--org myorg \
--token ${INFLUX_TOKEN}
写入监控数据
from influxdb_client_3 import InfluxDBClient3, Point
from datetime import datetime, timezone
import random
import time
client = InfluxDBClient3(
host="localhost:8181",
database="monitoring",
token="my-token"
)
def write_metrics():
points = []
for i in range(10000):
point = Point("cpu_metrics") \
.tag("host", f"node-{random.randint(1, 500)}") \
.tag("region", random.choice(["us-east", "eu-west", "ap-south"])) \
.field("usage_percent", random.uniform(10, 95)) \
.field("load_1m", random.uniform(0.5, 8.0)) \
.field("memory_percent", random.uniform(20, 90)) \
.time(datetime.now(timezone.utc))
points.append(point)
client.write(record=points, write_precision="ms")
print(f"Written {len(points)} points")
for _ in range(100):
write_metrics()
time.sleep(1)
降采样与连续查询
-- InfluxDB 3.0 SQL降采样：5分钟平均CPU使用率
SELECT
time_bucket('5 minutes', time) AS bucket,
host,
avg(cpu_usage_percent) AS avg_cpu,
max(cpu_usage_percent) AS max_cpu,
min(cpu_usage_percent) AS min_cpu
FROM cpu_metrics
WHERE time >= now() - INTERVAL '1 hour'
GROUP BY bucket, host
ORDER BY bucket DESC;
-- InfluxQL降采样（兼容模式）
SELECT
MEAN("usage_percent") AS "avg_cpu",
MAX("usage_percent") AS "max_cpu"
INTO "monitoring"."downsampled_5m"."cpu_metrics"
FROM "monitoring"."autogen"."cpu_metrics"
GROUP BY time(5m), "host"
数据保留策略
from influxdb_client import InfluxDBClient, BucketsService
client = InfluxDBClient(url="http://localhost:8086", token="my-token")
buckets_api = client.buckets_api()
raw_bucket = buckets_api.find_bucket_by_name("monitoring")
raw_bucket.retention_rules[0].every_seconds = 30 * 24 * 3600 # 30天
downsampled = buckets_api.create_bucket(
bucket_name="monitoring-downsampled",
retention_rules={"every_seconds": 365 * 24 * 3600}, # 1年
org="myorg"
)
Grafana集成
# grafana/provisioning/datasources/influxdb.yaml
apiVersion: 1
datasources:
- name: InfluxDB 3.0
type: influxdb
url: http://influxdb3:8181
access: proxy
jsonData:
version: Flux
organization: myorg
defaultBucket: monitoring
secureJsonData:
token: ${INFLUX_TOKEN}
Pattern 2：TimescaleDB 金融分析
场景描述
量化交易平台，每秒接收5000条股票Tick数据，需要实时计算VWAP（成交量加权平均价）、移动平均线、布林带，并支持复杂JOIN查询（关联交易订单表和用户表）。
安装与配置
# Docker部署TimescaleDB
docker run -d --name timescaledb \
-p 5432:5432 \
-e POSTGRES_PASSWORD=postgres \
-v tsdb-data:/var/lib/postgresql/data \
timescale/timescaledb:latest-pg16
# 启用扩展
psql -h localhost -U postgres -c "CREATE EXTENSION IF NOT EXISTS timescaledb;"
创建Hypertable与Schema
CREATE TABLE stock_ticks (
time TIMESTAMPTZ NOT NULL,
symbol VARCHAR(10) NOT NULL,
price NUMERIC(12, 4) NOT NULL,
volume BIGINT NOT NULL,
bid NUMERIC(12, 4),
ask NUMERIC(12, 4),
exchange VARCHAR(10)
);
SELECT create_hypertable('stock_ticks', 'time',
chunk_time_interval => INTERVAL '1 day',
partitioning_column => 'symbol',
number_partitions => 4
);
CREATE INDEX idx_symbol_time ON stock_ticks (symbol, time DESC);
CREATE TABLE trading_orders (
id BIGSERIAL,
time TIMESTAMPTZ NOT NULL,
symbol VARCHAR(10) NOT NULL,
side VARCHAR(4) NOT NULL,
price NUMERIC(12, 4) NOT NULL,
quantity BIGINT NOT NULL,
user_id BIGINT NOT NULL,
status VARCHAR(10) DEFAULT 'pending'
);
SELECT create_hypertable('trading_orders', 'time',
chunk_time_interval => INTERVAL '1 day'
);
连续聚合（Continuous Aggregates）
-- 1分钟K线
CREATE MATERIALIZED VIEW ohlc_1min
WITH (timescaledb.continuous) AS
SELECT
time_bucket('1 minute', time) AS bucket,
symbol,
first(price, time) AS open,
last(price, time) AS close,
max(price) AS high,
min(price) AS low,
sum(volume) AS volume
FROM stock_ticks
GROUP BY bucket, symbol
WITH NO DATA;
-- 自动刷新策略
SELECT add_continuous_aggregate_policy('ohlc_1min',
start_offset => INTERVAL '3 hours',
end_offset => INTERVAL '1 minute',
schedule_interval => INTERVAL '1 minute'
);
-- 5分钟K线（基于1分钟聚合）
CREATE MATERIALIZED VIEW ohlc_5min
WITH (timescaledb.continuous) AS
SELECT
time_bucket('5 minutes', bucket) AS bucket,
symbol,
first(open, bucket) AS open,
last(close, bucket) AS close,
max(high) AS high,
min(low) AS low,
sum(volume) AS volume
FROM ohlc_1min
GROUP BY bucket, symbol
WITH NO DATA;
SELECT add_continuous_aggregate_policy('ohlc_5min',
start_offset => INTERVAL '12 hours',
end_offset => INTERVAL '5 minutes',
schedule_interval => INTERVAL '5 minutes'
);
超函数（Hyperfunctions）计算技术指标
-- VWAP（成交量加权平均价）
SELECT
time_bucket('5 minutes', time) AS bucket,
symbol,
approx_percentile(0.5, 'UddSketch', price * volume / NULLIF(sum(price * volume) OVER (PARTITION BY symbol, time_bucket('5 minutes', time)), 0)) AS vwap
FROM stock_ticks
WHERE time >= now() - INTERVAL '1 hour'
GROUP BY bucket, symbol, price, volume;
-- 移动平均线（使用time_bucket_gapfill填充缺失数据）
SELECT
time_bucket_gapfill('1 minute', time, now() - INTERVAL '1 hour', now()) AS bucket,
symbol,
locf(last(price, time)) AS last_price,
avg(last(price, time)) OVER (
PARTITION BY symbol
ORDER BY bucket
ROWS BETWEEN 19 PRECEDING AND CURRENT ROW
) AS ma_20
FROM stock_ticks
WHERE time >= now() - INTERVAL '2 hours'
GROUP BY bucket, symbol;
-- 布林带
WITH ma_20 AS (
SELECT
time_bucket('5 minutes', time) AS bucket,
symbol,
avg(price) AS ma,
stddev(price) AS std
FROM stock_ticks
WHERE time >= now() - INTERVAL '2 hours'
GROUP BY bucket, symbol
HAVING count(*) >= 10
)
SELECT
bucket,
symbol,
ma,
ma + 2 * std AS upper_band,
ma - 2 * std AS lower_band
FROM ma_20
ORDER BY bucket DESC, symbol;
Python写入与查询
import asyncpg
import asyncio
from datetime import datetime, timezone
async def write_tick(pool, symbol, price, volume, bid, ask, exchange):
async with pool.acquire() as conn:
await conn.execute(
"""INSERT INTO stock_ticks (time, symbol, price, volume, bid, ask, exchange)
VALUES ($1, $2, $3, $4, $5, $6, $7)""",
datetime.now(timezone.utc), symbol, price, volume, bid, ask, exchange
)
async def batch_write_ticks(pool, ticks):
async with pool.acquire() as conn:
await conn.executemany(
"""INSERT INTO stock_ticks (time, symbol, price, volume, bid, ask, exchange)
VALUES ($1, $2, $3, $4, $5, $6, $7)""",
[(t['time'], t['symbol'], t['price'], t['volume'],
t['bid'], t['ask'], t['exchange']) for t in ticks]
)
async def query_ohlc(pool, symbol, hours=1):
async with pool.acquire() as conn:
rows = await conn.fetch(
"""SELECT bucket, symbol, open, close, high, low, volume
FROM ohlc_1min
WHERE symbol = $1 AND bucket >= now() - INTERVAL '$2 hours'
ORDER BY bucket""",
symbol, hours
)
return [dict(row) for row in rows]
async def main():
pool = await asyncpg.create_pool(
"postgresql://postgres:postgres@localhost:5432/postgres",
min_size=5, max_size=20
)
ticks = [
{"time": datetime.now(timezone.utc), "symbol": "AAPL",
"price": 198.50, "volume": 1000, "bid": 198.48, "ask": 198.52, "exchange": "NASDAQ"},
{"time": datetime.now(timezone.utc), "symbol": "GOOGL",
"price": 175.30, "volume": 800, "bid": 175.28, "ask": 175.32, "exchange": "NASDAQ"},
]
await batch_write_ticks(pool, ticks)
result = await query_ohlc(pool, "AAPL")
print(f"Got {len(result)} candles")
await pool.close()
asyncio.run(main())
Pattern 3：TDengine IoT边缘计算
场景描述
工业IoT网关，1000个传感器每秒上报数据，边缘节点资源有限（4核8G），需要超低延迟写入、超级表（Super Table）管理同类设备、集群部署保证高可用。
安装与配置
# Docker部署TDengine
docker run -d --name tdengine \
-p 6030:6030 \
-p 6041:6041 \
-v tdengine-data:/var/lib/taos \
tdengine/tdengine:3.3.4.0
# 使用taos CLI
taos
创建超级表与子表
-- 创建数据库
CREATE DATABASE iot_edge PRECISION 'ms' KEEP 3650 BUFFER 96 WAL_LEVEL 1;
USE iot_edge;
-- 创建超级表（模板）
CREATE STABLE sensor_data (
ts TIMESTAMP,
temperature FLOAT,
humidity FLOAT,
pressure FLOAT,
voltage FLOAT
) TAGS (
device_id BINARY(32),
device_type BINARY(16),
location BINARY(64),
factory BINARY(32)
);
-- 自动建表：插入时指定TAGS即可创建子表
INSERT INTO d_sensor_001 USING sensor_data TAGS ('sensor-001', 'temperature', 'workshop-A-line-1', 'factory-east')
VALUES (NOW, 23.5, 65.2, 1013.25, 3.3);
INSERT INTO d_sensor_002 USING sensor_data TAGS ('sensor-002', 'pressure', 'workshop-A-line-2', 'factory-east')
VALUES (NOW, 25.1, 60.8, 1012.80, 3.28);
-- 批量写入（高性能）
INSERT INTO d_sensor_001 VALUES (NOW + 1s, 23.6, 65.1, 1013.20, 3.31)
d_sensor_002 VALUES (NOW + 1s, 25.0, 60.9, 1012.75, 3.29)
d_sensor_001 VALUES (NOW + 2s, 23.7, 65.0, 1013.15, 3.30);
查询与分析
-- 按设备类型聚合：最近1小时平均温度
SELECT device_type, avg(temperature) AS avg_temp, max(temperature) AS max_temp
FROM sensor_data
WHERE ts >= NOW - 1h
GROUP BY device_type;
-- 按工厂和位置聚合
SELECT factory, location, count(*) AS sample_count,
avg(temperature) AS avg_temp,
stddev(temperature) AS temp_std
FROM sensor_data
WHERE ts >= NOW - 6h AND device_type = 'temperature'
GROUP BY factory, location;
-- 异常检测：温度超过3倍标准差
SELECT ts, device_id, temperature
FROM sensor_data
WHERE ts >= NOW - 1h
AND temperature > (
SELECT avg(temperature) + 3 * stddev(temperature)
FROM sensor_data
WHERE ts >= NOW - 24h AND device_type = 'temperature'
);
-- 降采样：每5分钟聚合
SELECT _wstart AS bucket, device_id,
avg(temperature) AS avg_temp,
min(temperature) AS min_temp,
max(temperature) AS max_temp
FROM sensor_data
WHERE ts >= NOW - 1h
INTERVAL(5m) SLIDING(5m) PARTITION BY device_id;
Python SDK写入
from taosrest import RestClient
from datetime import datetime, timezone
import json
import time
client = RestClient("http://localhost:6041", user="root", password="taosdata")
def write_sensor_batch(records):
sql_lines = []
for r in records:
line = f"d_{r['device_id']} USING sensor_data TAGS ('{r['device_id']}', '{r['device_type']}', '{r['location']}', '{r['factory']}') VALUES ('{r['ts']}', {r['temperature']}, {r['humidity']}, {r['pressure']}, {r['voltage']})"
sql_lines.append(line)
sql = "INSERT INTO " + " ".join(sql_lines)
client.sql(sql, database="iot_edge")
def query_latest(device_id, minutes=10):
result = client.sql(
f"SELECT ts, temperature, humidity, pressure FROM sensor_data "
f"WHERE device_id = '{device_id}' AND ts >= NOW - {minutes}m "
f"ORDER BY ts DESC LIMIT 100",
database="iot_edge"
)
return result
import random
records = []
for i in range(1000):
records.append({
"device_id": f"sensor-{i % 100:03d}",
"device_type": random.choice(["temperature", "pressure", "humidity"]),
"location": f"workshop-A-line-{i % 10}",
"factory": "factory-east",
"ts": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f"),
"temperature": round(random.uniform(20, 30), 2),
"humidity": round(random.uniform(50, 70), 2),
"pressure": round(random.uniform(1010, 1015), 2),
"voltage": round(random.uniform(3.2, 3.4), 2),
})
write_sensor_batch(records)
集群部署
# taos.cfg - 第一个dnode
firstEp tdnode1:6030
secondEp tdnode2:6030
serverPort 6030
dataDir /var/lib/taos
logDir /var/log/taos
arbitrator tdnode-arb:6042
# 创建dnode
taos> CREATE DNODE "tdnode2:6030";
taos> CREATE DNODE "tdnode3:6030";
-- 创建mnode
taos> ALTER DNODE 2 mnodeRole 'mnode';
taos> ALTER DNODE 3 mnodeRole 'mnode';
-- 数据库副本数
ALTER DATABASE iot_edge REPLICA 3;
Pattern 4：QuestDB 高频写入
场景描述
加密货币交易所行情数据，每秒50万条Tick，需要SQL接口直接查询、SIMD优化的聚合计算、ILP（InfluxDB Line Protocol）高速写入。
安装与配置
# Docker部署QuestDB
docker run -d --name questdb \
-p 9000:9000 \
-p 9009:9009 \
-p 8812:8812 \
-v questdb-data:/root/.questdb \
questdb/questdb:8.2.1
# Web Console: http://localhost:9000
# ILP端口: 9009
# PostgreSQL兼容端口: 8812
创建表与写入
-- 通过SQL创建表
CREATE TABLE crypto_ticks (
ts TIMESTAMP,
symbol SYMBOL,
price DOUBLE,
volume DOUBLE,
side SYMBOL,
exchange SYMBOL
) TIMESTAMP(ts) PARTITION BY DAY WAL;
-- 创建指定分区（提前规划）
-- QuestDB会自动按时间分区
ILP高速写入
import socket
import time
import random
from datetime import datetime, timezone
class QuestDBILPWriter:
def __init__(self, host="localhost", port=9009):
self.host = host
self.port = port
self.sock = None
def connect(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.host, self.port))
def write_line(self, line):
self.sock.sendall((line + "\n").encode())
def close(self):
if self.sock:
self.sock.close()
def write_tick(self, symbol, price, volume, side, exchange, ts_ns):
line = f"crypto_ticks,symbol={symbol},side={side},exchange={exchange} price={price},volume={volume} {ts_ns}"
self.write_line(line)
writer = QuestDBILPWriter()
writer.connect()
symbols = ["BTC-USD", "ETH-USD", "SOL-USD", "BNB-USD", "XRP-USD"]
base_time = int(time.time() * 1_000_000_000)
batch_size = 10000
total_written = 0
start = time.time()
for i in range(100000):
symbol = random.choice(symbols)
price = round(random.uniform(20000, 70000) if symbol == "BTC-USD" else random.uniform(1000, 4000), 2)
volume = round(random.uniform(0.01, 5.0), 6)
side = random.choice(["buy", "sell"])
exchange = random.choice(["binance", "coinbase", "kraken"])
ts_ns = base_time + i * 1_000_000
writer.write_tick(symbol, price, volume, side, exchange, ts_ns)
total_written += 1
if total_written % batch_size == 0:
elapsed = time.time() - start
rate = total_written / elapsed
print(f"Written {total_written} ticks, rate: {rate:.0f} ticks/s")
writer.close()
print(f"Total: {total_written} ticks in {time.time()-start:.2f}s")
SQL查询（SIMD优化）
-- 最近1分钟每个交易对的VWAP
SELECT
symbol,
sum(price * volume) / sum(volume) AS vwap,
sum(volume) AS total_volume,
count() AS tick_count
FROM crypto_ticks
WHERE ts >= dateadd('m', -1, now())
GROUP BY symbol;
-- 最近5分钟的OHLC
SELECT
symbol,
first(price) AS open,
last(price) AS close,
max(price) AS high,
min(price) AS low,
sum(volume) AS volume
FROM crypto_ticks
WHERE ts >= dateadd('m', -5, now())
SAMPLE BY 1m ALIGN TO CALENDAR;
-- 买卖量比
SELECT
symbol,
sum(volume) FILTER (WHERE side = 'buy') AS buy_volume,
sum(volume) FILTER (WHERE side = 'sell') AS sell_volume,
sum(volume) FILTER (WHERE side = 'buy') / sum(volume) FILTER (WHERE side = 'sell') AS buy_sell_ratio
FROM crypto_ticks
WHERE ts >= dateadd('m', -5, now())
GROUP BY symbol;
-- 价格波动率（5分钟标准差）
SELECT
symbol,
stddev(price) AS price_std,
avg(price) AS avg_price,
stddev(price) / avg(price * 1.0) AS cv
FROM crypto_ticks
WHERE ts >= dateadd('m', -5, now())
GROUP BY symbol
ORDER BY cv DESC;
PostgreSQL兼容查询
import psycopg2
conn = psycopg2.connect(
host="localhost",
port=8812,
user="admin",
password="quest",
database="qdb"
)
cursor = conn.cursor()
cursor.execute("""
SELECT symbol, sum(price * volume) / sum(volume) AS vwap
FROM crypto_ticks
WHERE ts >= dateadd('m', -5, now())
GROUP BY symbol
""")
for row in cursor.fetchall():
print(f"{row[0]}: VWAP = {row[1]:.2f}")
cursor.close()
conn.close()
Pattern 5：多数据库架构
场景描述
大型平台同时存在IoT数据、监控数据、金融数据，单一TSDB无法满足所有需求。需要设计路由层，根据数据特征自动选择最优数据库，并支持跨库联邦查询。
路由层设计
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional
import time
class DataType(Enum):
IOT_SENSOR = "iot_sensor"
MONITORING = "monitoring"
FINANCIAL = "financial"
HIGH_FREQUENCY = "high_frequency"
@dataclass
class TimeSeriesPoint:
measurement: str
tags: dict[str, str]
fields: dict[str, Any]
timestamp: int
data_type: DataType
class TSDBAdapter(ABC):
@abstractmethod
def write(self, points: list[TimeSeriesPoint]) -> int:
pass
@abstractmethod
def query(self, sql: str, params: Optional[dict] = None) -> list[dict]:
pass
@abstractmethod
def health_check(self) -> bool:
pass
class InfluxDBAdapter(TSDBAdapter):
def __init__(self, host: str, token: str, org: str, bucket: str):
from influxdb_client_3 import InfluxDBClient3
self.client = InfluxDBClient3(host=host, database=bucket, token=token)
self.org = org
self.bucket = bucket
def write(self, points: list[TimeSeriesPoint]) -> int:
from influxdb_client_3 import Point
influx_points = []
for p in points:
pt = Point(p.measurement)
for k, v in p.tags.items():
pt.tag(k, v)
for k, v in p.fields.items():
pt.field(k, v)
pt.time(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(p.timestamp / 1000)))
influx_points.append(pt)
self.client.write(record=influx_points)
return len(influx_points)
def query(self, sql: str, params: Optional[dict] = None) -> list[dict]:
result = self.client.query(sql)
return [row for row in result]
def health_check(self) -> bool:
try:
self.client.query("SELECT 1")
return True
except Exception:
return False
class TimescaleDBAdapter(TSDBAdapter):
def __init__(self, dsn: str):
import asyncpg
import asyncio
self.dsn = dsn
self.pool = None
async def init_pool(self):
import asyncpg
self.pool = await asyncpg.create_pool(self.dsn, min_size=2, max_size=10)
def write(self, points: list[TimeSeriesPoint]) -> int:
import asyncio
return asyncio.get_event_loop().run_until_complete(self._async_write(points))
async def _async_write(self, points: list[TimeSeriesPoint]) -> int:
async with self.pool.acquire() as conn:
rows = []
for p in points:
rows.append((
p.timestamp, p.measurement,
p.fields.get("price", 0), p.fields.get("volume", 0)
))
await conn.executemany(
"INSERT INTO ts_data (time, symbol, price, volume) VALUES ($1, $2, $3, $4)",
rows
)
return len(rows)
def query(self, sql: str, params: Optional[dict] = None) -> list[dict]:
import asyncio
return asyncio.get_event_loop().run_until_complete(self._async_query(sql))
async def _async_query(self, sql: str) -> list[dict]:
async with self.pool.acquire() as conn:
rows = await conn.fetch(sql)
return [dict(row) for row in rows]
def health_check(self) -> bool:
try:
return asyncio.get_event_loop().run_until_complete(self._async_health())
except Exception:
return False
async def _async_health(self) -> bool:
async with self.pool.acquire() as conn:
await conn.fetchval("SELECT 1")
return True
class TSDBRouter:
ROUTING_TABLE = {
DataType.IOT_SENSOR: "tdengine",
DataType.MONITORING: "influxdb",
DataType.FINANCIAL: "timescaledb",
DataType.HIGH_FREQUENCY: "questdb",
}
def __init__(self):
self.adapters: dict[str, TSDBAdapter] = {}
def register(self, name: str, adapter: TSDBAdapter):
self.adapters[name] = adapter
def write(self, points: list[TimeSeriesPoint]) -> dict[str, int]:
grouped: dict[str, list[TimeSeriesPoint]] = {}
for p in points:
target = self.ROUTING_TABLE.get(p.data_type, "influxdb")
grouped.setdefault(target, []).append(p)
results = {}
for target, target_points in grouped.items():
adapter = self.adapters.get(target)
if adapter and adapter.health_check():
results[target] = adapter.write(target_points)
else:
fallback = "influxdb" if target != "influxdb" else "timescaledb"
fallback_adapter = self.adapters.get(fallback)
if fallback_adapter:
results[f"{target}->fallback->{fallback}"] = fallback_adapter.write(target_points)
return results
def query(self, data_type: DataType, sql: str) -> list[dict]:
target = self.ROUTING_TABLE.get(data_type, "influxdb")
adapter = self.adapters.get(target)
if adapter and adapter.health_check():
return adapter.query(sql)
raise RuntimeError(f"No available adapter for {data_type}")
联邦查询
class FederatedQuery:
def __init__(self, router: TSDBRouter):
self.router = router
def cross_db_correlation(self, device_id: str, symbol: str, hours: int = 1):
iot_data = self.router.query(
DataType.IOT_SENSOR,
f"SELECT ts, avg(temperature) AS avg_temp FROM sensor_data "
f"WHERE device_id = '{device_id}' AND ts >= NOW - {hours}h "
f"INTERVAL(5m)"
)
finance_data = self.router.query(
DataType.FINANCIAL,
f"SELECT bucket, symbol, avg(close) AS avg_price "
f"FROM ohlc_5min WHERE symbol = '{symbol}' "
f"AND bucket >= now() - INTERVAL '{hours} hours' "
f"GROUP BY bucket, symbol"
)
return {
"iot": iot_data,
"finance": finance_data,
"correlation_note": "Cross-database correlation requires application-level join"
}
数据迁移工具
class TSDBMigrator:
def __init__(self, source: TSDBAdapter, target: TSDBAdapter, batch_size: int = 10000):
self.source = source
self.target = target
self.batch_size = batch_size
def migrate(self, query: str, transform_fn=None) -> int:
total = 0
offset = 0
while True:
paginated_query = f"{query} LIMIT {self.batch_size} OFFSET {offset}"
rows = self.source.query(paginated_query)
if not rows:
break
if transform_fn:
rows = [transform_fn(row) for row in rows]
self.target.write(rows)
total += len(rows)
offset += self.batch_size
print(f"Migrated {total} rows...")
print(f"Migration complete: {total} rows")
return total
避坑指南
坑1：InfluxDB Flux查询性能差
-- ❌ Flux嵌套管道，每一步都全量扫描
from(bucket: "monitoring")
|> range(start: -30d)
|> filter(fn: (r) => r._measurement == "cpu")
|> filter(fn: (r) => r.host =~ /node-.*/ )
|> aggregateWindow(every: 5m, fn: mean)
|> yield()
-- ✅ InfluxDB 3.0 SQL查询，利用Arrow向量化
SELECT time_bucket('5 minutes', time) AS bucket,
host, avg(cpu_usage_percent) AS avg_cpu
FROM cpu_metrics
WHERE time >= now() - INTERVAL '30 days'
AND host LIKE 'node-%'
GROUP BY bucket, host
坑2：TimescaleDB Hypertable分区键选错
-- ❌ 按月分区，查询1小时数据扫描过多chunk
SELECT create_hypertable('ticks', 'time',
chunk_time_interval => INTERVAL '1 month'
);
-- ✅ 按天分区，1小时查询只扫描1个chunk
SELECT create_hypertable('ticks', 'time',
chunk_time_interval => INTERVAL '1 day'
);
坑3：TDengine子表名冲突
-- ❌ 子表名全局唯一，不同超级表的子表名不能重复
CREATE STABLE temp_sensor (...) TAGS (device_id BINARY(32));
INSERT INTO d_001 USING temp_sensor TAGS ('dev-001', ...);
CREATE STABLE pressure_sensor (...) TAGS (device_id BINARY(32));
INSERT INTO d_001 USING pressure_sensor TAGS ('dev-001', ...);
-- ERROR: Table already exists
-- ✅ 子表名加上超级表前缀
INSERT INTO d_temp_001 USING temp_sensor TAGS ('dev-001', ...);
INSERT INTO d_pres_001 USING pressure_sensor TAGS ('dev-001', ...);
坑4：QuestDB SYMBOL类型滥用
-- ❌ 高基数字段用SYMBOL导致内存爆炸
CREATE TABLE logs (ts TIMESTAMP, user_id SYMBOL, message STRING) TIMESTAMP(ts);
-- ✅ 高基数字段用STRING，低基数用SYMBOL
CREATE TABLE logs (ts TIMESTAMP, level SYMBOL, user_id STRING, message STRING) TIMESTAMP(ts);
坑5：忽略压缩策略导致存储爆炸
-- TimescaleDB：启用原生压缩
ALTER TABLE stock_ticks SET (
timescaledb.compress,
timescaledb.compress_segmentby = 'symbol',
timescaledb.compress_orderby = 'time DESC'
);
SELECT add_compression_policy('stock_ticks', INTERVAL '7 days');
-- InfluxDB：设置合理的retention policy
influx bucket update --name monitoring --retention 30d
-- TDengine：建库时指定KEEP
CREATE DATABASE iot KEEP 3650;
报错排查
| 序号 | 报错信息 | 原因 | 解决方法 |
|---|---|---|---|
| 1 | InfluxDB engine: cache max memory size exceeded |
写入过快导致WAL缓存溢出 | 增大cache-max-memory-size，或降低写入频率 |
| 2 | TimescaleDB invalid value for chunk_time_interval |
分区间隔不是合法interval | 使用INTERVAL '1 day'而非纯数字 |
| 3 | TimescaleDB cannot drop chunk because it is compressed |
删除压缩chunk需先解压 | SELECT decompress_chunk('chunk_name') 再删除 |
| 4 | TDengine Invalid table name |
子表名包含非法字符或超长 | 子表名仅允许字母数字下划线，最长192字节 |
| 5 | TDengine Group by error |
GROUP BY字段不在SELECT中 | TDengine要求GROUP BY字段必须出现在SELECT |
| 6 | QuestDB commit lag |
ILP写入未及时commit | 调整commit.lag配置或显式flush |
| 7 | QuestDB symbol count exceeded |
SYMBOL基数超限 | 增大max.symbol.count或改用STRING |
| 8 | InfluxDB database not found |
Bucket不存在或名称错误 | 检查influx bucket list确认bucket名 |
| 9 | TimescaleDB hypertable already exists |
重复创建hypertable | 先DROP TABLE再重建，或用remove_distributed_hypertable |
| 10 | TDengine Out of memory |
边缘节点内存不足 | 减小BUFFER参数，降低VGROUPS数量 |
进阶优化
1. InfluxDB写入批量化
from influxdb_client_3 import InfluxDBClient3, Point
from datetime import datetime, timezone
import random
client = InfluxDBClient3(host="localhost:8181", database="monitoring", token="my-token")
batch = []
for i in range(50000):
point = Point("cpu_metrics") \
.tag("host", f"node-{random.randint(1, 500)}") \
.field("usage_percent", random.uniform(10, 95)) \
.time(datetime.now(timezone.utc))
batch.append(point)
if len(batch) >= 5000:
client.write(record=batch, write_precision="ms")
batch = []
if batch:
client.write(record=batch, write_precision="ms")
2. TimescaleDB压缩与空间节省
-- 查看压缩效果
SELECT
hypertable_name,
total_chunks,
number_compressed_chunks,
before_compression_bytes / 1024 / 1024 AS before_mb,
after_compression_bytes / 1024 / 1024 AS after_mb,
ROUND((1 - after_compression_bytes::FLOAT / before_compression_bytes) * 100, 1) AS compression_ratio_pct
FROM timescaledb_information.compressed_hypertable_stats;
3. TDengine性能调优参数
-- 创建数据库时优化参数
CREATE DATABASE iot_edge
PRECISION 'ms'
KEEP 3650
BUFFER 256
WAL_LEVEL 1
VGROUPS 4
CACHEMODEL 'both';
-- BUFFER: 写入缓存大小（MB），越大写入越快
-- WAL_LEVEL: 0=不写WAL 1=写WAL不fsync 2=写WAL并fsync
-- VGROUPS: 虚拟节点组数，建议=CPU核数
-- CACHEMODEL: 'both'缓存最新数据和元数据
4. QuestDB写入性能优化
# questdb.conf
cairo.commit.lag=10000
cairo.commit.mode=async
line.tcp.max.uncommitted.rows=10000
line.tcp.worker.count=2
对比分析
| 维度 | InfluxDB 3.0 | TimescaleDB | TDengine | QuestDB |
|---|---|---|---|---|
| 写入吞吐 | 50万点/秒 | 10万点/秒 | 100万点/秒 | 150万点/秒 |
| 查询延迟（简单聚合） | <10ms | <20ms | <5ms | <5ms |
| SQL兼容性 | InfluxQL + 基础SQL | 完整PostgreSQL | 部分SQL | 大部分SQL |
| JOIN支持 | 有限 | 完整 | 有限 | 有限 |
| 降采样 | Flux Task / SQL | Continuous Aggregates | INTERVAL查询 | SAMPLE BY |
| 集群 | 3.0原生 | 依赖PG | 原生 | 企业版 |
| 生态 | Grafana/Prometheus最强 | PG生态完整 | IoT生态 | 金融/IoT |
| 学习曲线 | 中 | 低（SQL） | 中 | 低（SQL） |
| 边缘部署 | 一般 | 不适合 | 极佳 | 一般 |
| 运维复杂度 | 中 | 中 | 低 | 低 |
总结：时序数据库选型没有银弹，关键看场景——监控选InfluxDB（生态最强）、金融选TimescaleDB（SQL完整）、IoT选TDengine（写入最快+边缘友好）、高频选QuestDB（SIMD优化极致）。大型平台推荐多数据库路由架构，按数据特征自动分流，避免一个数据库扛所有场景。
在线工具推荐
- JSON格式化：/zh-CN/json/format
- Hash计算：/zh-CN/encode/hash
- cURL转代码：/zh-CN/dev/curl-to-code
- SQL格式化：/zh-CN/dev/sql-format
本站提供浏览器本地工具，免注册即可试用 →
#时序数据库#InfluxDB#TimescaleDB#TDengine#QuestDB#2026#数据库