第22课：实时监控平台

💡 实时监控平台

构建实时系统监控平台，支持指标采集、异常检测、告警通知和可视化大屏。

🐍 Python实战：核心实战

# 实时监控引擎
import time, random
from datetime import datetime, timedelta
from collections import defaultdict

class MetricCollector:
    def __init__(self): self.metrics = defaultdict(list)
    def collect(self, name, value, timestamp=None):
        self.metrics[name].append({'value':value,'ts':timestamp or datetime.now()})
    def get_series(self, name, window_sec=300):
        cutoff = datetime.now() - timedelta(seconds=window_sec)
        return [(m['ts'],m['value']) for m in self.metrics[name] if m['ts'] > cutoff]

class AnomalyDetector:
    def __init__(self, window_size=30, threshold=3):
        self.window_size = window_size; self.threshold = threshold; self.history = []
    def check(self, value):
        self.history.append(value)
        if len(self.history) < self.window_size: return False, 0
        window = self.history[-self.window_size:]
        mean = sum(window)/len(window)
        std = (sum((v-mean)**2 for v in window)/len(window))**0.5
        if std == 0: return False, 0
        zscore = abs((value-mean)/std)
        return zscore > self.threshold, round(zscore,2)

class AlertManager:
    def __init__(self): self.alerts = []; self.suppress = defaultdict(int)
    def fire(self, name, level, message):
        key = f"{name}_{level}"
        if self.suppress[key] > 0: self.suppress[key] -= 1; return
        alert = {'name':name,'level':level,'message':message,'time':datetime.now().isoformat()}
        self.alerts.append(alert)
        print(f"  🚨 [{level}] {name}: {message}")
        self.suppress[key] = 5  # suppress next 5

# 运行监控
collector = MetricCollector()
detector = AnomalyDetector(window_size=20, threshold=2.5)
alerter = AlertManager()

print("实时监控引擎:")
for i in range(100):
    cpu = 30 + random.gauss(0, 10) + (100 if i == 50 else 0)  # spike at 50
    cpu = max(0, min(100, cpu))
    collector.collect('cpu_percent', cpu)
    is_anomaly, zscore = detector.check(cpu)
    if is_anomaly:
        alerter.fire('cpu_high', 'CRITICAL' if cpu > 90 else 'WARNING', f'CPU={cpu:.1f}% (z={zscore})')

series = collector.get_series('cpu_percent', 600)
print(f"\n指标统计:")
print(f"  CPU数据点: {len(series)}")
print(f"  告警数: {len(alerter.alerts)}")
print("✅ 验证通过 - 实时监控引擎运行正常")

📖 原理深入：企业数据平台架构

🏆 最佳实践：数据平台建设检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答

🔑 关键技术选型决策

数据量级？
├── <1GB/天 → SQLite/PostgreSQL + Python脚本
├── 1-100GB/天 → 数据仓库(Snowflake/BQ) + dbt
├── 100GB-1TB/天 → Spark + 数据湖
└── >1TB/天 → Spark/Flink + 湖仓一体

延迟要求？
├── T+1 → 批处理(Spark) + Airflow
├── 小时级 → 微批(Spark Streaming) + Airflow
├── 分钟级 → 流处理(Flink) + Kafka
└── 秒级 → 纯流(Flink) + Kafka + 缓存

数据类型？
├── 纯结构化 → 数据仓库
├── 结构化+半结构化 → 湖仓一体
└── 全类型 → 数据湖 + 元数据管理