时序数据库选型：IoT场景下QuestDB vs InfluxDB vs TDengine 2026
数据库
时序数据库选型：IoT场景下QuestDB vs InfluxDB vs TDengine
2026年的IoT世界里，你的传感器每秒吐出上万条数据点——温度、湿度、振动频率、GPS坐标……传统MySQL？写入瓶颈5分钟后就崩了。MongoDB？时序聚合查询慢到让人怀疑人生。
选错时序数据库，轻则查询超时、存储爆炸，重则整个数据管线瘫痪。QuestDB、InfluxDB、TDengine三足鼎立，各有绝活，但到底谁才是你的IoT场景最优解？本文从5个核心实战模式出发，带你彻底搞清楚。
核心概念速览
| 概念 | 说明 | 典型场景 |
|---|---|---|
| 时序数据（Time Series） | 按时间顺序排列的数据点序列 | 传感器采集、监控指标 |
| 标签（Tag） | 数据的维度/索引字段 | 设备ID、区域、类型 |
| 字段（Field） | 数据的度量值 | 温度值、电压值 |
| 降采样（Downsampling） | 将高频数据聚合为低频 | 秒级→分钟级→小时级 |
| 保留策略（Retention Policy） | 数据自动过期清理 | 热数据7天，冷数据1年 |
| 超级表（Super Table） | TDengine特有，模板化多设备表 | 1000台设备共用一张模板 |
IoT时序数据的5大痛点
- 写入吞吐瓶颈：百万级设备并发写入，传统数据库IOPS扛不住
- 查询聚合低效：时间窗口聚合、降采样查询性能差，P99延迟飙升
- 存储成本失控：时序数据量大且持续增长，冷热数据分离难
- 多设备管理复杂：千台设备各有标签，Schema管理混乱
- 生态集成割裂：与Grafana、Kafka、Telegraf等工具链对接成本高
模式一：IoT时序数据特征与建模
IoT时序数据有鲜明的特征：高基数标签、持续追加写入、时间局部性强、读多写多。理解这些特征是选型的前提。
# Python: 模拟IoT时序数据生成
# 运行环境: Python 3.12+ / 无额外依赖
import time
import json
import random
from datetime import datetime, timezone
def generate_iot_sensor_data(device_count: int = 100, interval_ms: int = 1000) -> dict:
"""生成IoT传感器时序数据点
Args:
device_count: 设备数量
interval_ms: 采集间隔(毫秒)
Returns:
单条时序数据点
"""
device_id = f"sensor-{random.randint(1, device_count):04d}"
region = random.choice(["east-cn", "west-cn", "south-cn", "north-cn"])
sensor_type = random.choice(["temperature", "humidity", "vibration", "pressure"])
# 模拟带漂移的传感器读数
base_values = {
"temperature": 25.0,
"humidity": 60.0,
"vibration": 0.5,
"pressure": 101.3,
}
value = round(base_values[sensor_type] + random.gauss(0, 2), 3)
return {
"timestamp": int(time.time() * 1_000_000), # 微秒精度
"tags": {
"device_id": device_id,
"region": region,
"sensor_type": sensor_type,
"factory": f"plant-{region.split('-')[0]}",
},
"fields": {
"value": value,
"quality": random.choice(["good", "good", "good", "warning", "error"]),
"battery": round(random.uniform(10, 100), 1),
},
}
def simulate_iot_stream(duration_seconds: int = 10, devices: int = 100) -> list[dict]:
"""模拟IoT数据流
Args:
duration_seconds: 模拟时长(秒)
devices: 设备数量
Returns:
时序数据点列表
"""
data_points = []
start = time.time()
while time.time() - start < duration_seconds:
point = generate_iot_sensor_data(device_count=devices)
data_points.append(point)
time.sleep(0.01) # 模拟100Hz采集
print(f"生成 {len(data_points)} 条数据点, "
f"时间跨度 {duration_seconds}s, "
f"设备数 {devices}")
return data_points
# 执行模拟
if __name__ == "__main__":
points = simulate_iot_stream(duration_seconds=5, devices=50)
print(json.dumps(points[0], indent=2, ensure_ascii=False))
模式二：QuestDB SQL时序查询
QuestDB以"SQL优先"著称，对标准SQL的时序扩展让查询极其直观。零依赖安装、高性能写入是它的杀手锏。
-- QuestDB: IoT时序数据SQL查询模式
-- 运行环境: QuestDB 8.x / ILP(InfluxDB Line Protocol)写入
-- 1. 创建表（QuestDB自动从ILP写入创建，也可手动定义）
CREATE TABLE IF NOT EXISTS sensor_data (
timestamp TIMESTAMP,
device_id SYMBOL,
region SYMBOL,
sensor_type SYMBOL,
factory SYMBOL,
value DOUBLE,
quality SYMBOL,
battery DOUBLE
) TIMESTAMP(timestamp) PARTITION BY DAY WAL;
-- 2. 时间窗口聚合 - 每5分钟平均温度
SELECT
timestamp,
device_id,
avg(value) AS avg_temp,
min(value) AS min_temp,
max(value) AS max_temp,
count() AS sample_count
FROM sensor_data
WHERE sensor_type = 'temperature'
AND timestamp >= dateadd('h', -1, now())
SAMPLE BY 5m ALIGN TO CALENDAR;
-- 3. 最新值查询 - LATEST ON语法
SELECT * FROM sensor_data LATEST ON timestamp PARTITION BY device_id;
-- 4. 降采样 + 多维度分组
SELECT
timestamp,
region,
sensor_type,
avg(value) AS avg_value,
stddev(value) AS std_value,
count() AS data_points
FROM sensor_data
WHERE timestamp >= dateadd('d', -7, now())
SAMPLE BY 1h ALIGN TO CALENDAR
GROUP BY region, sensor_type;
-- 5. 异常检测 - 连续3个点超阈值告警
WITH spike_detected AS (
SELECT
timestamp,
device_id,
value,
case when value > 35 OR value < 15 THEN 1 ELSE 0 END AS is_spike
FROM sensor_data
WHERE sensor_type = 'temperature'
)
SELECT
timestamp,
device_id,
value,
sum(is_spike) OVER (
ORDER BY timestamp
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
) AS consecutive_spikes
FROM spike_detected
WHERE consecutive_spikes >= 3;
# Python: QuestDB ILP写入客户端
# 运行环境: Python 3.12+ / pip install questdb
from questdb.ingress import IngressError, Sender, Buffer, Protocol
def write_iot_data_questdb(host: str = "localhost", port: int = 9009):
"""通过ILP协议写入QuestDB
Args:
host: QuestDB主机地址
port: ILP端口(默认9009)
"""
conf = f"http::addr={host}:{port};"
try:
with Sender.from_conf(conf) as sender:
# 批量写入传感器数据
for i in range(1000):
device_id = f"sensor-{i % 50:04d}"
region = ["east-cn", "west-cn", "south-cn"][i % 3]
sender.row(
"sensor_data",
symbols={
"device_id": device_id,
"region": region,
"sensor_type": "temperature",
"factory": f"plant-{region.split('-')[0]}",
},
columns={
"value": 25.0 + (i % 10) * 0.5,
"battery": 80.0 + (i % 20),
},
at=Sender.current_timestamp_with_nanos(),
)
sender.flush()
print(f"成功写入1000条数据到QuestDB")
except IngressError as e:
print(f"写入失败: {e}")
if __name__ == "__main__":
write_iot_data_questdb()
模式三：InfluxDB Flux查询
InfluxDB 3.x已转向SQL查询，但Flux在2.x生态中仍广泛使用。掌握Flux是运维存量系统的必备技能。
# Python: InfluxDB 2.x Flux查询模式
# 运行环境: Python 3.12+ / pip install influxdb-client[async]
from influxdb_client import InfluxDBClient, Point, WriteOptions
from influxdb_client.client.write_api import SYNCHRONOUS
import datetime
# 连接配置
INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = "your-api-token"
INFLUX_ORG = "iot-org"
INFLUX_BUCKET = "sensor-bucket"
def write_iot_data_influxdb():
"""写入IoT数据到InfluxDB 2.x"""
client = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
write_api = client.write_api(write_options=SYNCHRONOUS)
points = []
for i in range(500):
point = (
Point("sensor_data")
.tag("device_id", f"sensor-{i % 50:04d}")
.tag("region", ["east-cn", "west-cn", "south-cn"][i % 3])
.tag("sensor_type", "temperature")
.field("value", 25.0 + (i % 10) * 0.5)
.field("battery", 80.0 + (i % 20))
.time(datetime.datetime.utcnow(), write_precision="ms")
)
points.append(point)
write_api.write(bucket=INFLUX_BUCKET, org=INFLUX_ORG, record=points)
print(f"成功写入{len(points)}条数据到InfluxDB")
client.close()
# Flux查询示例（在InfluxDB UI或API中执行）
FLUX_QUERIES = """
// 1. 时间窗口聚合 - 每5分钟平均温度
from(bucket: "sensor-bucket")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "sensor_data")
|> filter(fn: (r) => r.sensor_type == "temperature")
|> filter(fn: (r) => r._field == "value")
|> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
|> yield(name: "avg_temp_5m")
// 2. 多维度分组统计
from(bucket: "sensor-bucket")
|> range(start: -7d)
|> filter(fn: (r) => r._measurement == "sensor_data")
|> filter(fn: (r) => r._field == "value")
|> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
|> group(columns: ["region", "sensor_type"])
|> yield(name: "hourly_by_region_type")
// 3. 异常检测 - 连续超阈值
from(bucket: "sensor-bucket")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "sensor_data")
|> filter(fn: (r) => r._field == "value")
|> map(fn: (r) => ({r with is_spike: if r._value > 35.0 or r._value < 15.0 then 1 else 0}))
|> window(every: 5m)
|> sum(column: "is_spike")
|> filter(fn: (r) => r.is_spike >= 3)
|> yield(name: "spike_alert")
// 4. 降采样写入连续查询
from(bucket: "sensor-bucket")
|> range(start: -30d)
|> filter(fn: (r) => r._measurement == "sensor_data")
|> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
|> to(bucket: "sensor-downsampled", org: "iot-org")
"""
if __name__ == "__main__":
write_iot_data_influxdb()
print("\nFlux查询示例:")
print(FLUX_QUERIES)
模式四：TDengine超级表
TDengine的超级表是IoT场景的杀手级特性：一张超级表模板管理千台设备，子表自动继承标签，查询性能极高。
-- TDengine 3.x: IoT超级表模式
-- 运行环境: TDengine 3.3+ / taos CLI
-- 1. 创建超级表（模板）
CREATE STABLE IF NOT EXISTS sensor_data (
ts TIMESTAMP,
value FLOAT,
battery FLOAT,
quality NCHAR(10)
) TAGS (
device_id NCHAR(20),
region NCHAR(20),
sensor_type NCHAR(20),
factory NCHAR(20)
);
-- 2. 自动建子表并写入（TDengine 3.x自动建表语法）
INSERT INTO d_sensor_0001 USING sensor_data TAGS ('sensor-0001', 'east-cn', 'temperature', 'plant-east')
VALUES (NOW + 0a, 25.3, 85.2, 'good');
INSERT INTO d_sensor_0002 USING sensor_data TAGS ('sensor-0002', 'west-cn', 'humidity', 'plant-west')
VALUES (NOW + 0a, 62.1, 91.5, 'good');
-- 3. 批量写入多设备（关键性能优化）
INSERT INTO
d_sensor_0001 VALUES (NOW + 1a, 25.5, 85.0, 'good') (NOW + 2a, 25.7, 84.8, 'good')
d_sensor_0002 VALUES (NOW + 1a, 62.3, 91.2, 'good') (NOW + 2a, 62.0, 91.0, 'warning');
-- 4. 超级表聚合查询 - 按区域统计
SELECT
region,
sensor_type,
AVG(value) AS avg_value,
STDDEV(value) AS std_value,
COUNT(*) AS data_points
FROM sensor_data
WHERE ts >= NOW - 1h
INTERVAL(5m)
GROUP BY region, sensor_type;
-- 5. 最新值查询 - LAST_ROW
SELECT
device_id,
LAST_ROW(value) AS latest_value,
LAST_ROW(ts) AS last_update
FROM sensor_data
GROUP BY device_id;
-- 6. 降采样 + 窗口函数
SELECT
_wstart AS window_start,
_wend AS window_end,
device_id,
AVG(value) AS avg_value,
MIN(value) AS min_value,
MAX(value) AS max_value
FROM sensor_data
WHERE ts >= NOW - 7d AND sensor_type = 'temperature'
INTERVAL(1h) SLIDING(30m)
GROUP BY device_id;
# Python: TDengine写入客户端
# 运行环境: Python 3.12+ / pip install taos-ws-py
import taosws
def write_iot_data_tdengine(host: str = "localhost", port: int = 6041):
"""写入IoT数据到TDengine 3.x
Args:
host: TDengine主机地址
port: WebSocket端口(默认6041)
"""
conn = taosws.connect(f"ws://{host}:{port}/rest/sql", user="root", password="taosdata")
# 创建数据库和超级表
conn.execute("CREATE DATABASE IF NOT EXISTS iot_db KEEP 3650")
conn.execute("USE iot_db")
conn.execute("""
CREATE STABLE IF NOT EXISTS sensor_data (
ts TIMESTAMP,
value FLOAT,
battery FLOAT,
quality NCHAR(10)
) TAGS (
device_id NCHAR(20),
region NCHAR(20),
sensor_type NCHAR(20),
factory NCHAR(20)
)
""")
# 批量写入
for i in range(100):
device_id = f"sensor-{i:04d}"
region = ["east-cn", "west-cn", "south-cn"][i % 3]
table_name = f"d_sensor_{i:04d}"
sql = f"""
INSERT INTO {table_name} USING sensor_data
TAGS ('{device_id}', '{region}', 'temperature', 'plant-{region.split('-')[0]}')
VALUES (NOW + {i}a, {25.0 + i * 0.1}, {80.0 + i * 0.2}, 'good')
"""
conn.execute(sql)
print("成功写入100条数据到TDengine")
# 查询验证
result = conn.execute("SELECT COUNT(*) FROM sensor_data")
for row in result:
print(f"总数据量: {row[0]}")
conn.close()
if __name__ == "__main__":
write_iot_data_tdengine()
模式五：生产级选型对比
# Python: 时序数据库选型评估框架
# 运行环境: Python 3.12+ / 无额外依赖
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class TSDBEvaluation:
"""时序数据库选型评估模型"""
name: str
write_throughput_per_sec: int # 单节点写入吞吐(点/秒)
query_latency_p99_ms: float # 聚合查询P99延迟(ms)
compression_ratio: float # 压缩率(原始/压缩后)
sql_compatibility: int # SQL兼容性(1-10)
ecosystem_maturity: int # 生态成熟度(1-10)
cluster_scalability: int # 集群扩展性(1-10)
learning_curve: int # 学习曲线(1-10, 10=最易)
license_type: str # 开源协议
best_for: str # 最佳场景
caution: str # 注意事项
def evaluate_tsdb_choices() -> list[TSDBEvaluation]:
"""生成三大时序数据库评估结果"""
return [
TSDBEvaluation(
name="QuestDB",
write_throughput_per_sec=1_500_000,
query_latency_p99_ms=15.0,
compression_ratio=8.5,
sql_compatibility=9,
ecosystem_maturity=6,
cluster_scalability=5,
learning_curve=9,
license_type="Apache 2.0",
best_for="SQL团队、高频写入、金融/IoT实时分析",
caution="集群版商业授权、生态插件较少",
),
TSDBEvaluation(
name="InfluxDB",
write_throughput_per_sec=500_000,
query_latency_p99_ms=45.0,
compression_ratio=5.2,
sql_compatibility=5,
ecosystem_maturity=9,
cluster_scalability=7,
learning_curve=6,
license_type="MIT / 商业",
best_for="DevOps监控、Telegraf生态、中小规模IoT",
caution="Flux学习曲线陡峭、3.x迁移成本高、集群版收费",
),
TSDBEvaluation(
name="TDengine",
write_throughput_per_sec=2_000_000,
query_latency_p99_ms=8.0,
compression_ratio=12.0,
sql_compatibility=7,
ecosystem_maturity=7,
cluster_scalability=8,
learning_curve=7,
license_type="AGPL-3.0 / 商业",
best_for="超大规模IoT、车联网、工业互联网、国产化",
caution="AGPL协议限制、社区版功能受限、SQL方言差异",
),
]
def print_comparison_table(evaluations: list[TSDBEvaluation]):
"""打印选型对比表"""
print(f"{'指标':<25} {'QuestDB':<20} {'InfluxDB':<20} {'TDengine':<20}")
print("-" * 85)
fields = [
("写入吞吐(点/秒)", "write_throughput_per_sec", True),
("查询P99延迟(ms)", "query_latency_p99_ms", False),
("压缩率", "compression_ratio", True),
("SQL兼容性(1-10)", "sql_compatibility", True),
("生态成熟度(1-10)", "ecosystem_maturity", True),
("集群扩展性(1-10)", "cluster_scalability", True),
("学习曲线(1-10)", "learning_curve", True),
("开源协议", "license_type", None),
]
for label, attr, _ in fields:
values = [str(getattr(e, attr)) for e in evaluations]
print(f"{label:<25} {values[0]:<20} {values[1]:<20} {values[2]:<20}")
print("\n最佳场景:")
for e in evaluations:
print(f" {e.name}: {e.best_for}")
print("\n注意事项:")
for e in evaluations:
print(f" {e.name}: {e.caution}")
if __name__ == "__main__":
evaluations = evaluate_tsdb_choices()
print_comparison_table(evaluations)
避坑指南：5个常见陷阱
坑1：高基数标签导致OOM
❌ 错误做法：将device_id作为InfluxDB的Tag
✅ 正确做法：高基数字段放Field，低基数维度放Tag
InfluxDB的Tag会建立倒排索引，10万个device_id = 10万个Series，内存直接爆炸。QuestDB的SYMBOL类型和TDengine的超级表天然解决了这个问题。
坑2：忽略时区导致聚合错位
-- ❌ 错误：未指定时区，跨时区聚合错位
SAMPLE BY 1h
-- ✅ 正确：QuestDB指定时区对齐
SAMPLE BY 1h ALIGN TO CALENDAR WITH TIME ZONE 'Asia/Shanghai'
坑3：保留策略未配置导致磁盘爆满
-- InfluxDB: 必须设置RP
CREATE RETENTION POLICY "7days" ON "sensor-bucket" DURATION 7d REPLICATION 1 DEFAULT
-- TDengine: 建库时指定KEEP
CREATE DATABASE iot_db KEEP 3650 -- 保留3650天
坑4：批量写入大小不当
# ❌ 错误：逐条写入，网络开销巨大
for point in data_points:
write_api.write(bucket=BUCKET, record=point)
# ✅ 正确：批量写入，推荐5000-10000条/批
BATCH_SIZE = 5000
for i in range(0, len(data_points), BATCH_SIZE):
batch = data_points[i:i + BATCH_SIZE]
write_api.write(bucket=BUCKET, record=batch)
坑5：TDengine超级表与子表混淆
-- ❌ 错误：对超级表直接INSERT数据（无子表）
INSERT INTO sensor_data VALUES (NOW, 25.0, 80.0, 'good');
-- ✅ 正确：通过子表写入，自动继承超级表标签
INSERT INTO d_sensor_0001 USING sensor_data TAGS (...)
VALUES (NOW, 25.0, 80.0, 'good');
报错排查表
| 报错信息 | 数据库 | 原因 | 解决方案 |
|---|---|---|---|
too many series |
InfluxDB | 高基数Tag导致Series超限 | 将高基数字段移至Field，或升级InfluxDB 3.x |
memory limit exceeded |
QuestDB | 查询结果集过大 | 添加LIMIT、使用SAMPLE BY降采样 |
Invalid column type |
QuestDB | SYMBOL列写入超长字符串 | 控制SYMBOL长度<64字节，或改用STRING |
table does not exist |
TDengine | 未USE数据库 | 先执行USE iot_db |
out of memory |
TDengine | 单查询内存超限 | 调整queryMemory参数，缩小时间范围 |
connection refused :9009 |
QuestDB | ILP端口未启用 | 配置tcp.enabled=true |
authorization failed |
InfluxDB | Token权限不足 | 检查Token的read/write权限 |
database not found |
InfluxDB | Bucket不存在 | 先创建Bucket或检查拼写 |
invalid timestamp |
QuestDB | 时间戳精度不匹配 | ILP默认纳秒，确认时间戳单位 |
duplicate table name |
TDengine | 子表名冲突 | 确保子表名全局唯一 |
进阶优化：5个生产级技巧
技巧1：冷热数据分层存储
-- QuestDB: 分区 + O3引擎冷热分离
ALTER TABLE sensor_data ALTER PARTITION LIST
SET ATTRIBUTE 'cold' WHERE timestamp < dateadd('d', -30, now());
-- TDengine: 多级存储
ALTER DATABASE iot_db KEEP 30,365,3650;
-- 30天热数据 / 365天温数据 / 3650天冷数据
技巧2：预聚合物化视图
-- QuestDB: 物化视图自动降采样
CREATE MATERIALIZED VIEW sensor_hourly AS (
SELECT
timestamp,
device_id,
avg(value) AS avg_value,
min(value) AS min_value,
max(value) AS max_value,
count() AS sample_count
FROM sensor_data
SAMPLE BY 1h ALIGN TO CALENDAR
) PARTITION BY DAY;
技巧3：写入批处理与背压控制
# Python: 带背压控制的QuestDB写入
# 运行环境: Python 3.12+ / pip install questdb
import time
import threading
from queue import Queue, Full
from questdb.ingress import Sender, IngressError
class BatchingWriter:
"""带背压控制的批量写入器"""
def __init__(self, host: str = "localhost", port: int = 9009,
batch_size: int = 5000, max_queue: int = 100_000):
self.batch_size = batch_size
self.queue = Queue(maxsize=max_queue)
self.sender = Sender.from_conf(f"http::addr={host}:{port};")
self._running = True
self._worker = threading.Thread(target=self._flush_loop, daemon=True)
self._worker.start()
def write(self, table: str, symbols: dict, columns: dict):
"""非阻塞写入，队列满时丢弃"""
try:
self.queue.put_nowait((table, symbols, columns))
except Full:
print("警告: 写入队列已满，丢弃数据点")
def _flush_loop(self):
"""后台批量刷盘"""
batch = []
while self._running or not self.queue.empty():
try:
item = self.queue.get(timeout=1.0)
batch.append(item)
if len(batch) >= self.batch_size:
self._flush(batch)
batch = []
except Exception:
if batch:
self._flush(batch)
batch = []
def _flush(self, batch: list):
"""执行批量写入"""
try:
for table, symbols, columns in batch:
self.sender.row(table, symbols=symbols, columns=columns,
at=Sender.current_timestamp_with_nanos())
self.sender.flush()
print(f"刷盘 {len(batch)} 条数据")
except IngressError as e:
print(f"写入失败: {e}")
def close(self):
self._running = False
self._worker.join(timeout=10)
self.sender.close()
技巧4：Grafana可视化集成
# docker-compose.yml: QuestDB + Grafana一体化部署
# 运行环境: Docker Compose v2.30+
version: "3.8"
services:
questdb:
image: questdb/questdb:8.2
ports:
- "9000:9000" # Web Console
- "9009:9009" # ILP
- "9003:9003" # Min health server
volumes:
- questdb_data:/root/.questdb
environment:
QDB_MALLOC_SIZE: 4G
grafana:
image: grafana/grafana:11.4
ports:
- "3000:3000"
environment:
GF_INSTALL_PLUGINS: grafana-postgresql-datasource
volumes:
- grafana_data:/var/lib/grafana
depends_on:
- questdb
volumes:
questdb_data:
grafana_data:
技巧5：多活写入与高可用
# TDengine 3.x 集群部署
# 运行环境: TDengine 3.3+ / 3节点集群
# taos.cfg 关键配置
cluster: 1
numOfMnodes: 3
mnodeEqualVnodeNum: 4
statusInterval: 1
maxTablesPerVnode: 1000
minRowsPerBlock: 100
maxRowsPerBlock: 4096
arbitrator: taos://arbiter:6180
三大时序数据库对比分析
| 维度 | QuestDB | InfluxDB | TDengine |
|---|---|---|---|
| 查询语言 | SQL（扩展时序函数） | Flux / SQL(3.x) | SQL方言 |
| 写入协议 | ILP / REST / PostgreSQL wire | ILP / REST | REST / WebSocket / JNI |
| 单节点写入 | 150万点/秒 | 50万点/秒 | 200万点/秒 |
| 压缩率 | 8-10x | 4-6x | 10-15x |
| 集群方案 | 企业版 | 开源集群(3.x) | 开源集群 |
| Grafana集成 | ✅ PostgreSQL数据源 | ✅ 原生插件 | ✅ 原生插件 |
| Kafka集成 | ✅ 社区 | ✅ 原生 | ✅ 原生 |
| 学习成本 | 低（SQL即学即用） | 中高（Flux语法独特） | 中（超级表概念） |
| 国产化适配 | ❌ | ❌ | ✅ 信创兼容 |
| 适用规模 | 中小→大型 | 中小型 | 中大→超大规模 |
总结
IoT时序数据库选型没有银弹，关键看你的场景匹配度：
- QuestDB：SQL团队首选，高频写入+实时分析场景，学习成本最低，但集群版需商业授权
- InfluxDB：DevOps监控生态最成熟，Telegraf+Grafana一条龙，但Flux学习曲线陡、3.x迁移成本高
- TDengine：超大规模IoT杀手，超级表+压缩率碾压，国产化友好，但AGPL协议需注意
选型决策树：SQL团队→QuestDB / DevOps生态→InfluxDB / 超大规模IoT+国产化→TDengine。
在线工具推荐
- /zh-CN/json/format - JSON格式化，处理时序数据API返回
- /zh-CN/dev/curl-to-code - cURL转代码，快速生成时序数据库API调用
- /zh-CN/encode/hash - 哈希计算，设备ID去重与校验
- /zh-CN/text/diff - 文本对比，SQL查询版本diff
本站提供浏览器本地工具，免注册即可试用 →
#时序数据库#IoT数据#QuestDB#InfluxDB#TDengine#2026#数据库