本文目录导读：
我来详细介绍Python操作时序数据库的几种主流方法：
操作InfluxDB
安装驱动
pip install influxdb-client
连接和写入数据
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import datetime
# 连接数据库
client = InfluxDBClient(
url="http://localhost:8086",
token="your-token",
org="your-org"
)
# 写入数据
write_api = client.write_api(write_options=SYNCHRONOUS)
# 创建数据点
point = Point("temperature") \
.tag("location", "room1") \
.field("value", 25.5) \
.time(datetime.datetime.utcnow())
# 批量写入
points = [
Point("sensor_data")
.tag("sensor_id", "sensor_001")
.field("temperature", 23.5)
.field("humidity", 65.2)
.time("2023-01-01T00:00:00Z"),
Point("sensor_data")
.tag("sensor_id", "sensor_002")
.field("temperature", 24.1)
.field("humidity", 62.8)
.time("2023-01-01T00:01:00Z")
]
write_api.write(bucket="my-bucket", record=points)
查询数据
from influxdb_client.client.query_api import QueryApi
query_api = client.query_api()
# 使用Flux语言查询
query = '''
from(bucket: "my-bucket")
|> range(start: -1h)
|> filter(fn: (r) => r["_measurement"] == "temperature")
|> filter(fn: (r) => r["location"] == "room1")
|> aggregateWindow(every: 5m, fn: mean)
'''
result = query_api.query(query)
# 处理结果
for table in result:
for record in table.records:
print(f"Time: {record.get_time()}, Value: {record.get_value()}")
操作TimescaleDB
安装驱动
pip install psycopg2-binary
连接和创建超表
import psycopg2
from datetime import datetime
# 连接数据库
conn = psycopg2.connect(
host="localhost",
port=5432,
database="timescale_db",
user="postgres",
password="password"
)
cursor = conn.cursor()
# 创建超表
cursor.execute("""
CREATE TABLE sensor_data (
time TIMESTAMPTZ NOT NULL,
sensor_id TEXT NOT NULL,
temperature DOUBLE PRECISION,
humidity DOUBLE PRECISION
);
""")
# 转换为超表
cursor.execute("SELECT create_hypertable('sensor_data', 'time');")
conn.commit()
批量插入数据
import pandas as pd
from sqlalchemy import create_engine
# 创建连接引擎
engine = create_engine('postgresql://postgres:password@localhost/timescale_db')
# 生成测试数据
data = []
base_time = datetime.now()
for i in range(1000):
data.append({
'time': base_time + pd.Timedelta(seconds=i),
'sensor_id': f'sensor_{i % 10}',
'temperature': 20 + 10 * (i % 100) / 100,
'humidity': 50 + 20 * (i % 100) / 100
})
df = pd.DataFrame(data)
# 批量插入
df.to_sql('sensor_data', engine, if_exists='append', index=False, method='multi')
时序查询
# 时间范围查询
cursor.execute("""
SELECT time, sensor_id, temperature
FROM sensor_data
WHERE time > NOW() - INTERVAL '1 hour'
AND sensor_id = 'sensor_001'
ORDER BY time DESC;
""")
# 时间桶聚合
cursor.execute("""
SELECT
time_bucket('5 minutes', time) AS five_min,
sensor_id,
AVG(temperature) AS avg_temp,
MAX(temperature) AS max_temp
FROM sensor_data
WHERE time > NOW() - INTERVAL '1 day'
GROUP BY five_min, sensor_id
ORDER BY five_min DESC;
""")
results = cursor.fetchall()
操作Prometheus
安装客户端
pip install prometheus-client
创建自定义指标
from prometheus_client import Counter, Gauge, Histogram, Summary, start_http_server
import random
import time
# 定义指标
request_count = Counter('app_requests_total', 'Total app requests', ['endpoint'])
in_progress = Gauge('app_in_progress', 'Number of requests in progress')
request_duration = Histogram('app_request_duration_seconds', 'Request duration')
request_summary = Summary('app_request_latency_seconds', 'Request latency')
# 模拟请求处理
@request_duration.time()
@request_summary.time()
def process_request():
in_progress.inc()
time.sleep(random.uniform(0.1, 0.3))
request_count.labels(endpoint='/api/data').inc()
in_progress.dec()
# 启动HTTP服务暴露指标
start_http_server(8000)
while True:
process_request()
time.sleep(1)
查询Prometheus数据
import requests
import json
# 查询Prometheus
def query_prometheus(query):
response = requests.get(
'http://localhost:9090/api/v1/query',
params={'query': query}
)
return response.json()
# 查询CPU使用率
result = query_prometheus('rate(node_cpu_seconds_total[1m])')
print(json.dumps(result, indent=2))
# 范围查询
response = requests.get(
'http://localhost:9090/api/v1/query_range',
params={
'query': 'node_memory_MemAvailable_bytes',
'start': '2023-01-01T00:00:00Z',
'end': '2023-01-01T01:00:00Z',
'step': '5m'
}
)
使用通用方法（Pandas + SQL）
import pandas as pd
from datetime import datetime
import numpy as np
# 生成时序数据
def generate_time_series(n_points=1000):
base = datetime.now()
timestamps = [base + pd.Timedelta(seconds=i) for i in range(n_points)]
df = pd.DataFrame({
'timestamp': timestamps,
'value': np.random.randn(n_points),
'sensor': np.random.choice(['A', 'B', 'C'], n_points)
})
return df
# 使用Pandas进行时序分析
df = generate_time_series(1000)
df.set_index('timestamp', inplace=True)
# 重采样（降采样）
hourly_mean = df['value'].resample('1H').mean()
# 滚动窗口计算
rolling_mean = df['value'].rolling(window=10).mean()
# 时间序列分析
from statsmodels.tsa.seasonal import seasonal_decompose
# 提取季节性成分
decomposition = seasonal_decompose(df['value'], model='additive', period=24)
trend = decomposition.trend
seasonal = decomposition.seasonal
residual = decomposition.resid
异步操作（高性能场景）
import asyncio
from influxdb_client.client.write_api import ASYNCHRONOUS
import aioinflux
async def async_write_influx():
client = InfluxDBClient(
url="http://localhost:8086",
token="your-token",
org="your-org",
enable_gzip=True
)
write_api = client.write_api(write_options=ASYNCHRONOUS)
# 异步批量写入
points = []
for i in range(1000):
point = Point("async_metrics") \
.tag("batch_id", f"batch_{i // 100}") \
.field("value", i) \
.time(datetime.utcnow())
points.append(point)
success = await write_api.write(bucket="my-bucket", record=points)
return success
# 运行异步任务
result = asyncio.run(async_write_influx())
连接池和批量处理优化
from contextlib import contextmanager
from influxdb_client import InfluxDBClient
import threading
class TimeSeriesDBPool:
def __init__(self, url, token, org, max_connections=5):
self.url = url
self.token = token
self.org = org
self.max_connections = max_connections
self._connections = []
self._lock = threading.Lock()
@contextmanager
def get_client(self):
with self._lock:
if len(self._connections) < self.max_connections:
client = InfluxDBClient(
url=self.url,
token=self.token,
org=self.org
)
self._connections.append(client)
else:
client = self._connections[-1]
try:
yield client
finally:
pass
def batch_write(self, bucket, points, batch_size=100):
with self.get_client() as client:
write_api = client.write_api()
# 分批写入
for i in range(0, len(points), batch_size):
batch = points[i:i + batch_size]
write_api.write(bucket=bucket, record=batch)
# 使用连接池
pool = TimeSeriesDBPool(
url="http://localhost:8086",
token="your-token",
org="your-org"
)
# 批量写入数据
pool.batch_write("my-bucket", points_list, batch_size=50)
选择建议
- InfluxDB：专注时序列，写入性能好，适合IoT场景
- TimescaleDB：基于PostgreSQL，支持完整SQL，适合复杂查询
- Prometheus：监控告警生态完善，适合系统监控
- 基于Pandas：适合数据分析和原型开发
根据你的具体需求（性能、查询复杂度、生态集成）选择合适的数据库和操作方法。
