Python脚本如何操作时序数据库

本文目录导读：

1. 操作InfluxDB
2. 操作TimescaleDB
3. 操作Prometheus
4. 使用通用方法（Pandas + SQL）
5. 异步操作（高性能场景）
6. 连接池和批量处理优化
7. 选择建议

我来详细介绍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：适合数据分析和原型开发

根据你的具体需求（性能、查询复杂度、生态集成）选择合适的数据库和操作方法。