本文目录导读：
我来给您提供一个完整的监控数据抽样脚本,包含多种抽样方法和统计检验。
基础抽样脚本
#!/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
核心功能说明
-
多种抽样方法：
- 简单随机抽样
- 分层抽样
- 系统抽样
- 整群抽样
-
质量验证：
- 统计量对比（均值、标准差等）
- 假设检验（t检验、KS检验）
- 可视化对比
-
实时采样支持：
- 蓄水池算法
- 时间窗口采样
-
输出功能：
- CSV文件保存
- 可视化报告
- 统计验证报告
这个脚本可以根据您的具体监控数据类型和需求进行调整。
