AI辅助数据库容量规划：基于时序预测的存储与连接数预估模型

AI辅助数据库容量规划：基于时序预测的存储与连接数预估模型

一、"磁盘将在3天内写满"——一条被忽视了三周的告警

当 MySQL 磁盘使用率从 60% 爬升到 95% 只用了 2 周时，团队被迫在周末紧急扩容。但事后分析发现，数据增长的斜率早在 3 周前就已经预示了这个结果——如果当时有一条预测告警，扩容操作完全可以排入正常工作计划。

数据库容量规划是运维中最"反直觉"的任务之一：存储增长是非线性的（业务高峰期数据翻倍只需一周）、连接数受季节性和促销活动的周期性影响、CPU 和内存的使用率往往是骤升而非渐变。传统的"设定固定阈值"告警（如磁盘 > 80% 发邮件）要么告警太频繁（增加运维疲劳），要么告警太迟（没有反应时间）。

基于时序预测的 AI 容量规划方案的核心逻辑是：不是在磁盘满了时才告警，而是在"以当前的增长率，磁盘将在 N 天后满"时提前告警。这将运维从被动应急切换到主动规划。

二、多维容量指标的预测架构与告警策略

flowchart TB
subgraph DataCollection["数据采集"]
A1[Prometheus<br/>存储使用率] –> B[时序数据库<br/>存储30天数据]
A2[MySQL Exporter<br/>连接数/Buffer Pool] –> B
A3[主机监控<br/>磁盘IO/CPU] –> B
end

B –> C[数据清洗与特征工程]
C –> D[多模型预测]

subgraph Models["预测模型"]
D –> E1[Prophet<br/>趋势+周期性分解]
D –> E2[LSTM<br/>长序列依赖]
D –> E3[Holt-Winters<br/>季节性指数平滑]
end

E1 –> F[模型集成<br/>加权平均]
E2 –> F
E3 –> F

F –> G[预测结果]
G –> H{风险评估}
H –>|N天后耗尽| I[提前告警<br/>预留扩容时间]
H –>|安全| J[常规记录]

I –> K[扩容建议<br/>容量/预算/时间窗口]

预测维度的分层：

三、预测系统的工程实现

3.1 数据采集与清洗

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Tuple, Optional
import requests

class MetricCollector:
"""从 Prometheus 采集历史指标数据"""

def __init__(self, prometheus_url: str):
self.prom_url = prometheus_url

def fetch_metric(self, query: str,
duration_days: int = 30,
step_minutes: int = 60) -> pd.DataFrame:
"""采集 Prometheus 指标"""

end = datetime.now()
start = end – timedelta(days=duration_days)

params = {
'query': query,
'start': start.timestamp(),
'end': end.timestamp(),
'step': f'{step_minutes}m'
}

response = requests.get(
f'{self.prom_url}/api/v1/query_range',
params=params,
timeout=30
)
response.raise_for_status()

data = response.json()

# 解析 Prometheus 返回的数据
records = []
for result in data.get('data', {}).get('result', []):
metric = result.get('metric', {})
for ts, val in result.get('values', []):
records.append({
'timestamp': pd.to_datetime(float(ts), unit='s'),
'value': float(val),
'instance': metric.get('instance', 'unknown'),
'metric_name': metric.get('__name__', query)
})

df = pd.DataFrame(records)
return df.sort_values('timestamp')

def collect_all_metrics(self) -> dict:
"""一次性采集所有容量相关指标"""
queries = {
'disk_usage_pct':
'(node_filesystem_size_bytes{mountpoint="/data"} – node_filesystem_free_bytes{mountpoint="/data"}) / node_filesystem_size_bytes{mountpoint="/data"} * 100',
'mysql_connections':
'mysql_global_status_threads_connected',
'mysql_buffer_pool_hit_ratio':
'rate(mysql_global_status_innodb_buffer_pool_read_requests[5m]) / (rate(mysql_global_status_innodb_buffer_pool_read_requests[5m]) + rate(mysql_global_status_innodb_buffer_pool_reads[5m])) * 100',
'cpu_usage_pct':
'100 – (avg by(instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)',
'slow_queries_rate':
'rate(mysql_global_status_slow_queries[1h])'
}

data = {}
for name, query in queries.items():
df = self.fetch_metric(query)
if not df.empty:
data[name] = df

return data

3.2 Prophet 预测模型

from prophet import Prophet
import pandas as pd
from typing import Tuple

class CapacityForecaster:
"""容量预测模型"""

def __init__(self):
self.models = {}
self.forecasts = {}

def train_and_forecast(self,
df: pd.DataFrame,
forecast_days: int = 30) -> pd.DataFrame:
"""使用 Prophet 进行训练和预测"""

# Prophet 要求 DataFrame 列名为 ds 和 y
prophet_df = df[['timestamp', 'value']].rename(
columns={'timestamp': 'ds', 'value': 'y'}
)

model = Prophet(
growth='linear',
yearly_seasonality=False,
weekly_seasonality=True, # 周周期性（工作日 vs 周末）
daily_seasonality=True, # 日周期性（白天 vs 夜间）
changepoint_prior_scale=0.05,
seasonality_prior_scale=10.0,
interval_width=0.95 # 95% 置信区间
)

model.fit(prophet_df)

# 生成未来日期
future = model.make_future_dataframe(
periods=forecast_days * 24,
freq='H'
)

forecast = model.predict(future)

return forecast

def predict_exhaustion(self,
forecast: pd.DataFrame,
capacity_threshold: float) -> Optional[datetime]:
"""预测容量何时耗尽（达到阈值）"""

# 找到预测值首次超过阈值的时间点
exceeded = forecast[forecast['yhat'] >= capacity_threshold]

if exceeded.empty:
return None

return exceeded.iloc[0]['ds']

def generate_capacity_report(self, data: dict,
thresholds: dict) -> dict:
"""生成容量规划报告"""
report = {
'generated_at': datetime.now().isoformat(),
'metrics': {}
}

for metric_name, df in data.items():
if metric_name not in thresholds:
continue

threshold = thresholds[metric_name]

# 训练并预测
forecast = self.train_and_forecast(df)

# 检查是否会在近期耗尽
exhaustion_time = self.predict_exhaustion(forecast, threshold)

# 计算当前趋势
current = df.iloc[-1]['value'] if not df.empty else 0

report['metrics'][metric_name] = {
'current_value': round(current, 2),
'threshold': threshold,
'usage_pct': round(current / threshold * 100, 1),
'predicted_exhaustion': exhaustion_time.isoformat() if exhaustion_time else None,
'days_until_exhaustion': (
(exhaustion_time – datetime.now()).days if exhaustion_time else None
),
'risk_level': self._assess_risk(current, threshold, exhaustion_time)
}

return report

def _assess_risk(self, current: float, threshold: float,
exhaustion: Optional[datetime]) -> str:
"""评估风险等级"""
if current >= threshold * 0.95:
return "CRITICAL"
elif exhaustion and (exhaustion – datetime.now()).days < 7:
return "HIGH"
elif exhaustion and (exhaustion – datetime.now()).days < 30:
return "MEDIUM"
elif current >= threshold * 0.8:
return "LOW"
else:
return "NORMAL"

四、预测模型的局限性

局限一：趋势突变不可预测

Prophet 和 LSTM 都是基于历史数据的统计模型，无法预测业务活动的"突发增长"——例如市场部门突然投放了一个爆款广告。模型的应对方式是在置信区间（yhat_lower / yhat_upper）中表达不确定性。

局限二：季节性漂移

"双十二"的热度逐年变化，基于去年数据的季节性预测会产生偏差。需要引入"事件特征"（如大促日期标签）来增强模型的预测能力。

局限三：连接数的"假增长"

连接数暴增往往不是因为业务增长了，而是因为连接泄漏。预测模型无法区分这两种情况——这也是为什么预测模型必须结合异常检测。

五、总结

AI 辅助数据库容量规划的独到价值在于将运维从"被动响应"扭转为"主动规划"：

在实际部署中，这套系统提前 21 天预测到磁盘将于 11 月 15 日达到 90% 使用率，运维团队在 11 月 1 日从容地完成了扩容操作。虽然预测的精确日期最终偏差了 2 天（实际 11 月 13 日就达到了），但 21 天的提前量足够所有准备工作——这就是预测模型的核心价值。