监控数据抽样结果的脚本如何写

wen 实用脚本 2026-07-05 86

本文目录导读：

监控数据抽样结果的脚本如何写

基础抽样脚本
进阶版本：实时监控抽样
使用示例
核心功能说明

我来给您提供一个完整的监控数据抽样脚本,包含多种抽样方法和统计检验。

基础抽样脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
监控数据抽样脚本
功能：从监控数据中进行随机抽样、分层抽样，并进行统计验证
"""
import pandas as pd
import numpy as np
import random
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
from scipy import stats
class MonitoringSampler:
    """监控数据抽样器"""
    def __init__(self, data):
        """
        初始化抽样器
        :param data: pandas DataFrame，包含监控数据
        """
        self.data = data
        self.sample_result = None
    def simple_random_sample(self, sample_size, seed=42):
        """
        简单随机抽样
        :param sample_size: 抽样数量或比例（小于1为比例，大于1为数量）
        :param seed: 随机种子
        """
        np.random.seed(seed)
        if sample_size < 1:
            # 按比例抽样
            n = int(len(self.data) * sample_size)
        else:
            n = int(sample_size)
        sample_indices = np.random.choice(len(self.data), n, replace=False)
        self.sample_result = self.data.iloc[sample_indices]
        return self.sample_result
    def stratified_sample(self, stratify_column, sample_size, seed=42):
        """
        分层抽样
        :param stratify_column: 分层依据的列名
        :param sample_size: 每层抽样数量或比例
        """
        np.random.seed(seed)
        if isinstance(sample_size, float) and sample_size < 1:
            # 按比例从每层抽样
            self.sample_result = self.data.groupby(stratify_column).apply(
                lambda x: x.sample(frac=sample_size, random_state=seed)
            ).reset_index(drop=True)
        else:
            # 按固定数量从每层抽样
            self.sample_result = self.data.groupby(stratify_column).apply(
                lambda x: x.sample(n=sample_size, random_state=seed)
            ).reset_index(drop=True)
        return self.sample_result
    def systematic_sample(self, sample_size):
        """
        系统抽样
        :param sample_size: 抽样数量
        """
        n = len(self.data)
        k = n // sample_size  # 抽样间隔
        start = random.randint(0, k - 1)
        indices = list(range(start, n, k))[:sample_size]
        self.sample_result = self.data.iloc[indices]
        return self.sample_result
    def cluster_sample(self, cluster_column, n_clusters, seed=42):
        """
        整群抽样
        :param cluster_column: 群标识列
        :param n_clusters: 抽取的群数量
        """
        np.random.seed(seed)
        clusters = self.data[cluster_column].unique()
        selected_clusters = np.random.choice(clusters, n_clusters, replace=False)
        self.sample_result = self.data[self.data[cluster_column].isin(selected_clusters)]
        return self.sample_result
class SampleValidator:
    """抽样结果验证器"""
    def __init__(self, original_data, sample_data):
        self.original = original_data
        self.sample = sample_data
    def compare_statistics(self, numeric_columns):
        """
        比较原始数据和样本数据的统计量
        :param numeric_columns: 数值型列名列表
        """
        comparison = {}
        for col in numeric_columns:
            stats_dict = {
                '原始均值': self.original[col].mean(),
                '样本均值': self.sample[col].mean(),
                '均值偏差(%)': abs(self.original[col].mean() - self.sample[col].mean()) / self.original[col].mean() * 100,
                '原始标准差': self.original[col].std(),
                '样本标准差': self.sample[col].std(),
                '原始中位数': self.original[col].median(),
                '样本中位数': self.sample[col].median()
            }
            comparison[col] = stats_dict
        return pd.DataFrame(comparison).T
    def hypothesis_test(self, numeric_columns, alpha=0.05):
        """
        假设检验：验证样本是否能代表总体
        """
        results = {}
        for col in numeric_columns:
            # t检验
            t_stat, p_value = stats.ttest_1samp(
                self.sample[col].dropna(), 
                self.original[col].mean()
            )
            # Kolmogorov-Smirnov检验
            ks_stat, ks_pvalue = stats.ks_2samp(
                self.original[col].dropna(),
                self.sample[col].dropna()
            )
            results[col] = {
                't统计量': t_stat,
                't检验p值': p_value,
                'KS统计量': ks_stat,
                'KS检验p值': ks_pvalue,
                '通过检验(p>0.05)': p_value > alpha and ks_pvalue > alpha
            }
        return pd.DataFrame(results).T
    def visualize_comparison(self, numeric_columns, save_path=None):
        """
        可视化比较原始数据和样本数据
        """
        n_cols = len(numeric_columns)
        fig, axes = plt.subplots(2, n_cols, figsize=(5*n_cols, 10))
        for i, col in enumerate(numeric_columns):
            # 直方图对比
            axes[0, i].hist(self.original[col], alpha=0.5, label='原始数据', bins=30)
            axes[0, i].hist(self.sample[col], alpha=0.5, label='样本数据', bins=30)
            axes[0, i].set_title(f'{col} 分布对比')
            axes[0, i].legend()
            # 箱线图对比
            data_to_plot = [self.original[col], self.sample[col]]
            axes[1, i].boxplot(data_to_plot, labels=['原始', '样本'])
            axes[1, i].set_title(f'{col} 箱线图对比')
        plt.tight_layout()
        if save_path:
            plt.savefig(save_path, dpi=100, bbox_inches='tight')
        plt.show()
def generate_monitoring_data(n_samples=10000):
    """
    生成模拟监控数据
    """
    np.random.seed(42)
    data = pd.DataFrame({
        'timestamp': pd.date_range('2024-01-01', periods=n_samples, freq='5min'),
        'cpu_usage': np.random.normal(60, 15, n_samples).clip(0, 100),
        'memory_usage': np.random.normal(70, 10, n_samples).clip(0, 100),
        'response_time': np.random.exponential(200, n_samples),
        'error_count': np.random.poisson(2, n_samples),
        'server_group': np.random.choice(['A', 'B', 'C', 'D'], n_samples),
        'region': np.random.choice(['us-east', 'us-west', 'eu-west'], n_samples)
    })
    return data
def main():
    """主函数：演示使用"""
    print("="*60)
    print("监控数据抽样脚本")
    print("="*60)
    # 1. 生成模拟数据
    print("\n[1] 生成模拟监控数据...")
    data = generate_monitoring_data(10000)
    print(f"数据维度: {data.shape}")
    print(f"数据列: {list(data.columns)}")
    print(f"时间范围: {data['timestamp'].min()} 到 {data['timestamp'].max()}")
    # 2. 执行抽样
    print("\n[2] 执行抽样...")
    sampler = MonitoringSampler(data)
    # 简单随机抽样
    print("\n2.1 简单随机抽样 (10%)")
    sample_simple = sampler.simple_random_sample(0.1)
    print(f"样本大小: {len(sample_simple)}")
    # 分层抽样
    print("\n2.2 分层抽样 (按region, 每层20%)")
    sample_stratified = sampler.stratified_sample('region', 0.2)
    print(f"样本大小: {len(sample_stratified)}")
    # 系统抽样
    print("\n2.3 系统抽样 (1000个样本)")
    sample_systematic = sampler.systematic_sample(1000)
    print(f"样本大小: {len(sample_systematic)}")
    # 3. 验证抽样结果
    print("\n[3] 验证抽样结果...")
    validator = SampleValidator(data, sample_simple)
    # 统计量对比
    numeric_cols = ['cpu_usage', 'memory_usage', 'response_time', 'error_count']
    print("\n3.1 统计量对比:")
    stats_comparison = validator.compare_statistics(numeric_cols)
    print(stats_comparison.to_string())
    # 假设检验
    print("\n3.2 假设检验:")
    hypothesis_results = validator.hypothesis_test(numeric_cols)
    print(hypothesis_results.to_string())
    # 可视化
    print("\n3.3 生成可视化对比图...")
    validator.visualize_comparison(numeric_cols, 'sampling_comparison.png')
    # 4. 输出抽样报告
    print("\n[4] 抽样报告摘要")
    print(f"原始数据大小: {len(data)}")
    print(f"样本数据大小: {len(sample_simple)}")
    print(f"抽样比例: {len(sample_simple)/len(data)*100:.2f}%")
    # 保存样本数据
    sample_simple.to_csv('monitoring_sample.csv', index=False)
    print("\n样本数据已保存到: monitoring_sample.csv")
    return sample_simple
if __name__ == "__main__":
    main()

进阶版本：实时监控抽样

import asyncio
from kafka import KafkaConsumer
import json
from collections import deque
class RealTimeMonitoringSampler:
    """实时监控数据抽样器"""
    def __init__(self, kafka_topic, bootstrap_servers=['localhost:9092']):
        self.consumer = KafkaConsumer(
            kafka_topic,
            bootstrap_servers=bootstrap_servers,
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        self.buffer = deque(maxlen=10000)  # 滑动窗口
        self.sample_buffer = deque(maxlen=1000)
    async def reservoir_sample(self, k, stream):
        """
        蓄水池抽样算法
        :param k: 抽样数量
        :param stream: 数据流
        """
        reservoir = []
        for i, item in enumerate(stream):
            if i < k:
                reservoir.append(item)
            else:
                j = random.randint(0, i)
                if j < k:
                    reservoir[j] = item
            if i % 100 == 0:  # 每100条记录报告一次
                print(f"处理了 {i} 条记录，当前样本大小: {len(reservoir)}")
        return reservoir
    async def time_interval_sample(self, interval_seconds=60):
        """
        时间间隔抽样：每interval_seconds抽取一条
        """
        last_sample_time = datetime.now()
        for message in self.consumer:
            current_time = datetime.now()
            self.buffer.append(message.value)
            if (current_time - last_sample_time).seconds >= interval_seconds:
                # 从buffer中随机抽取
                sample = random.choice(list(self.buffer))
                self.sample_buffer.append(sample)
                last_sample_time = current_time
                yield sample
# 命令行工具版本
if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description='监控数据抽样工具')
    parser.add_argument('--input', help='输入数据文件路径')
    parser.add_argument('--output', default='sample_output.csv', help='输出文件路径')
    parser.add_argument('--method', choices=['random', 'stratified', 'systematic', 'cluster'],
                       default='random', help='抽样方法')
    parser.add_argument('--size', type=float, default=0.1, help='样本大小（比例或数量）')
    parser.add_argument('--validate', action='store_true', help='是否进行验证')
    args = parser.parse_args()
    # 加载数据
    data = pd.read_csv(args.input)
    # 执行抽样
    sampler = MonitoringSampler(data)
    if args.method == 'random':
        sample = sampler.simple_random_sample(args.size)
    elif args.method == 'stratified':
        sample = sampler.stratified_sample('server_group', args.size)
    elif args.method == 'systematic':
        sample = sampler.systematic_sample(args.size)
    elif args.method == 'cluster':
        sample = sampler.cluster_sample('region', args.size)
    # 保存结果
    sample.to_csv(args.output, index=False)
    # 验证（如果指定）
    if args.validate:
        validator = SampleValidator(data, sample)
        numeric_cols = data.select_dtypes(include=[np.number]).columns.tolist()
        stats_df = validator.compare_statistics(numeric_cols)
        print(stats_df.to_string())

使用示例

# 基本使用
python monitoring_sampler.py
# 命令行使用
python monitoring_sampler.py \
    --input monitoring_data.csv \
    --output sample_result.csv \
    --method stratified \
    --size 0.2 \
    --validate