時序資料庫選型：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-tw", "west-tw", "south-tw", "north-tw"])
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)
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. 建立表
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"""
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-tw", "west-tw", "south-tw"][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-tw", "west-tw", "south-tw"][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_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. 自動建子表並寫入
INSERT INTO d_sensor_0001 USING sensor_data TAGS ('sensor-0001', 'east-tw', 'temperature', 'plant-east')
VALUES (NOW + 0a, 25.3, 85.2, 'good');
INSERT INTO d_sensor_0002 USING sensor_data TAGS ('sensor-0002', 'west-tw', '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"""
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-tw", "west-tw", "south-tw"][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
@dataclass
class TSDBEvaluation:
"""時序資料庫選型評估模型"""
name: str
write_throughput_per_sec: int
query_latency_p99_ms: float
compression_ratio: float
sql_compatibility: int
ecosystem_maturity: int
cluster_scalability: int
learning_curve: int
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/Taipei'
坑3：保留策略未配置導致磁碟爆滿
-- InfluxDB: 必須設定RP
CREATE RETENTION POLICY "7days" ON "sensor-bucket" DURATION 7d REPLICATION 1 DEFAULT
-- TDengine: 建庫時指定KEEP
CREATE DATABASE iot_db KEEP 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;
技巧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"
- "9009:9009"
- "9003:9003"
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節點叢集
cluster: 1
numOfMnodes: 3
mnodeEqualVnodeNum: 4
statusInterval: 1
maxTablesPerVnode: 1000
minRowsPerBlock: 100
maxRowsPerBlock: 4096
三大時序資料庫對比分析
| 維度 | 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-TW/json/format - JSON格式化，處理時序資料API回傳
- /zh-TW/dev/curl-to-code - cURL轉程式碼，快速生成時序資料庫API呼叫
- /zh-TW/encode/hash - 雜湊計算，裝置ID去重與校驗
- /zh-TW/text/diff - 文字對比，SQL查詢版本diff
本站提供瀏覽器本地工具，免註冊即可試用 →
#时序数据库#IoT数据#QuestDB#InfluxDB#TDengine#2026#数据库