Pandas 数据分析详解

核心概念 Pandas 是 Python 核心数据分析库，提供 DataFrame 和 Series 数据结构。

1. 基础定位与核心作用

1.1. 定义

pandas 是 Python 数据分析标准库，基于 numpy 开发，专门用于结构化表格数据处理，对标 Excel、SQL 表。

两大核心数据结构：

Series：一维带标签数组（单列）
DataFrame：二维表格（多列、带行索引、列名，最常用）

1.2. 适配你的运维/AIOps 典型场景

时序监控指标清洗（Prometheus/VictoriaMetrics 导出指标、CPU/内存时间序列）
日志结构化分析：过滤报错、统计错误频次、聚合每分钟告警数量
数据库 ClickHouse/MySQL 查询结果做统计、异常阈值筛选
批量设备巡检数据聚合、磁盘使用率排序、找出高负载机器
训练时序异常检测数据集（Prophet、IsolationForest 输入预处理）
导出备份统计报表、故障复盘表格、CSV 监控报告

1.3. 安装


pip install pandas numpy openpyxl

# openpyxl 用于读写 Excel

1.4. 导入惯例（行业统一写法）


import pandas as pd

import numpy as np

2. 两大核心数据结构

2.1. Series 一维序列（单列）

结构 = 索引(index) + 一列值


# 创建Series

s = pd.Series([22, 45, 78, 12], index=["hostA","hostB","hostC","hostD"], name="cpu")

print(s)

# 取值

print(s["hostA"])

# 转列表

print(s.tolist())

# 过滤大于50的CPU

print(s[s > 50])

2.2. DataFrame 二维数据表（重点，99%场景使用）

类比数据库一张表：

index：行索引（时间戳、主机名、序号）
columns：列名（cpu、mem、disk、time、error_count）
values：表格数值

2.2.1. 快速创建DataFrame


# 字典构造（最常用）

data = {

    "host": ["node1", "node2", "node3"],

    "cpu": [23, 89, 41],

    "mem": [45, 92, 33],

    "time": ["2026-07-07 10:00","2026-07-07 10:00","2026-07-07 10:00"]

}

df = pd.DataFrame(data)

print(df)

3. 数据读取与导出（运维高频）

3.1. 读取外部数据源

3.1.1. 1）CSV 监控/日志文件


df = pd.read_csv("monitor.csv", encoding="utf-8")

# 常用参数

df = pd.read_csv(

    "metrics.csv",

    parse_dates=["time"],    # 自动把time列转为时间格式（时序必备）

    index_col="time",        # 将时间设为行索引

    dtype={"cpu": float, "host": str}, # 指定列数据类型

    sep=",",

    na_values=["-", "null"]  # 识别空值标记

)

3.1.2. 2）从 MySQL/ClickHouse 读取（搭配SQL驱动）


import pymysql

conn = pymysql.connect(host="127.0.0.1", user="root", password="xxx", database="monitor")

sql = "select host, cpu, mem, collect_time from host_metrics limit 10000;"

df = pd.read_sql(sql, conn)

3.1.3. 3）列表/字典直接生成（psutil采集数据）


rows = [

    {"host":"node1","cpu":30,"mem":44},

    {"host":"node2","cpu":92,"mem":88}

]

df = pd.DataFrame(rows)

3.2. 导出保存


# 输出csv（监控报表）

df.to_csv("report.csv", index=False, encoding="utf-8-sig")

# 输出Excel

df.to_excel("report.xlsx", index=False)

# 输出SQL插入语句、写入数据库

df.to_sql(name="daily_report", con=conn, if_exists="append", index=False)

4. 基础查看与数据筛选

4.1. 基础信息查看


df.head(5)        # 前5行

df.tail(3)         # 后3行

df.info()          # 列类型、空值统计（排查缺失指标）

df.describe()      # 数值列统计：均值、最大最小、标准差（异常检测基线依据）

df.columns         # 所有列名

df.shape           # (行数,列数)

df.dtypes          # 每列数据类型

4.2. 取列操作


# 单列 → Series

df["cpu"]

# 多列 → 新DataFrame

df[["host","cpu","mem"]]

# 通过属性取值（无空格列名可用）

df.cpu

4.3. 行筛选（布尔索引，AIOps异常过滤核心）

4.3.1. 示例1：筛选CPU > 80 的异常主机


# 条件

mask = df["cpu"] > 80

# 取出满足条件行

error_df = df[mask]

print(error_df)

4.3.2. 多条件组合 `&` 且 / `|` 或


# CPU>80 并且 内存>85

mask = (df["cpu"] > 80) & (df["mem"] > 85)

# CPU>90 或者 磁盘>95

mask = (df["cpu"] > 90) | (df["disk"] > 95)

# 排除node1机器

mask = df["host"] != "node1"

4.3.3. 模糊匹配日志文本（日志分析）


# 匹配包含timeout的错误日志

mask = df["log_msg"].str.contains("timeout", na=False)

timeout_log = df[mask]

4.4. loc / iloc 精准切片

loc：按行索引名、列名取值（时序时间索引最常用）
iloc：按数字下标 0,1,2 取值


# 取单行

df.loc[0]

# 按时间索引筛选时间段（时序核心）

df.loc["2026-07-07 09:00":"2026-07-07 12:00"]

# 指定行+指定列

df.loc[df["cpu"]>90, ["host","time","cpu"]]

# iloc 下标

df.iloc[0:10, 0:3] # 前10行，前3列

5. 清洗数据（运维日志/监控必做步骤）

5.1. 空值处理（采集缺失指标）


# 判断空值

df.isna().sum() # 统计每列空值数量

# 删除存在空值的行

df.dropna(subset=["cpu","mem"])

# 空值填充：cpu空值填0

df["cpu"] = df["cpu"].fillna(0)

# 填充固定字符串

df["log_msg"].fillna("empty")

5.2. 去重（重复采集指标、重复告警）


# 根据主机+时间去重，保留第一条

df.drop_duplicates(subset=["host","time"], keep="first")

5.3. 删除无用列、行


# 删除指定列

df = df.drop(columns=["useless_col"])

# 删除行，按索引

df = df.drop(index=[0,1])

5.4. 新增计算列（衍生指标）


# 新增负载总分 = cpu + mem

df["load_score"] = df["cpu"] + df["mem"]

# 标记是否异常 1=异常 0=正常

df["is_abnormal"] = np.where((df["cpu"]>80) | (df["mem"]>85), 1, 0)

5.5. 类型转换


# 字符串转浮点

df["cpu"] = pd.to_numeric(df["cpu"], errors="coerce")

# 字符串转时间（时序核心）

df["time"] = pd.to_datetime(df["time"])

6. 分组聚合（统计分析核心，AIOps高频）

6.1. groupby 分组统计：按主机、按分钟、按业务线聚合指标

6.1.1. 基础聚合函数

mean()均值 max()最大值 min()最小值 sum()求和 count()计数 std()标准差（动态基线）


# 按主机分组，求每台机器CPU平均值

host_avg = df.groupby("host")["cpu"].mean()

# 多指标聚合

stat = df.groupby("host").agg(

    cpu_max=("cpu","max"),

    mem_avg=("mem","mean"),

    error_count=("is_abnormal","sum")

).reset_index() # reset_index() 将分组索引变回普通列

print(stat)

6.1.2. 日志场景：统计每种报错出现次数


error_cnt = df.groupby("log_msg")["host"].count().sort_values(ascending=False)

7. 时序时间序列专属功能（你的AIOps核心重点）

7.1. 时间索引重采样（分钟→小时聚合）

监控原始数据10秒一条，重采样按小时取均值/最大值，降低数据量


# 将time设为索引

df = df.set_index("time")

# 每小时重采样，取CPU最大值

hour_df = df.resample("1H")["cpu"].max()

# 10分钟均值

df.resample("10min")["mem"].mean()

重采样频率标识：1s秒 1T分 1H小时 1D天

7.2. 时间范围筛选


# 只看今天数据

today = df.loc["2026-07-07"]

# 9点到11点

df.loc["2026-07-07 09:00":"2026-07-07 11:00"]

7.3. 滑动窗口计算（动态基线、时序异常检测必备）

滚动窗口计算近5分钟均值/标准差，判断当前值是否偏离基线


# 窗口5个采集点，滚动求均值

df["cpu_roll_mean"] = df["cpu"].rolling(window=5).mean()

# 滚动标准差

df["cpu_roll_std"] = df["cpu"].rolling(window=5).std()

# 动态阈值：超出均值±3倍标准差标记异常

df["anomaly"] = np.where(abs(df["cpu"] - df["cpu_roll_mean"]) > 3 * df["cpu_roll_std"], 1, 0)

8. 表关联 merge（多源监控数据合并）

类似SQL join，合并主机基础信息与监控指标


# 主机资产表

asset_df = pd.DataFrame({"host":["node1","node2"], "biz":"tax", "idc":"sh"})

# 监控指标表

metric_df = pd.DataFrame({"host":["node1","node2"], "cpu":[22,88]})

# 内连接

result = pd.merge(metric_df, asset_df, on="host", how="left")

how 参数：

left 左连接（常用）
inner 内连接
outer 全连接

9. 排序、TopN 高负载主机


# 按CPU降序

df_sorted = df.sort_values("cpu", ascending=False)

# 取负载最高前10台机器

top10 = df.sort_values("cpu", ascending=False).head(10)

10. 运维/AIOps完整实战示例：时序异常标记

读取监控CSV，滑动窗口3σ算法标记异常样本，输出异常报表


import pandas as pd

import numpy as np

# 1. 读取监控数据

df = pd.read_csv("metrics.csv", parse_dates=["time"])

df = df.set_index("time")

# 2. 滚动窗口计算基线

window = 12 # 近12个点（假设10s一点=2分钟窗口）

df["cpu_mean"] = df["cpu"].rolling(window=window).mean()

df["cpu_std"] = df["cpu"].rolling(window=window).std()

# 3. 3σ异常判断

df["upper"] = df["cpu_mean"] + 3 * df["cpu_std"]

df["lower"] = df["cpu_mean"] - 3 * df["cpu_std"]

df["is_abnormal"] = np.where((df["cpu"] > df["upper"]) | (df["cpu"] < df["lower"]), 1, 0)

# 4. 筛选异常数据

abnormal_df = df[df["is_abnormal"] == 1]

# 导出异常报表

abnormal_df.to_csv("abnormal_report.csv", encoding="utf-8-sig")

print(f"检测到异常条数：{len(abnormal_df)}")

11. 生产避坑与性能优化

大文件内存溢出超大监控CSV使用分块读取 pd.read_csv(chunksize=10000)，分批处理；
时间列务必parse_dates
字符串时间无法切片、重采样，必须转为datetime64；
布尔条件多括号
& | 优先级高于比较符，每个条件必须加 ()；
空值影响聚合
mean()会自动跳过NaN，但训练数据集建议fillna填充；
高频循环不要用for遍历行
pandas向量化运算远快于for循环，尽量使用布尔筛选、apply、rolling；
内存优化：数值列转float32、int32，字符串分类 astype("category")；
多进程海量数据可搭配 dask.dataframe 替代pandas，突破内存限制。

12. pandas 在自研AIOps产品中的定位

数据预处理层核心工具：清洗原始采集指标、补齐缺失值、统一格式；
时序算法预处理：滚动窗口、3σ基线、重采样生成训练数据集；
批量统计分析：每日巡检报表、故障时段聚合、高频错误统计；
对接机器学习库：sklearn、Prophet 输入统一要求DataFrame结构化数据；
输出标准化CSV/Excel复盘文件，对接告警通知。

13. 对比你已学工具

psutil：采集原始主机指标；pandas：指标清洗、统计、时序计算；二者搭配做监控采集分析；
configparser：读取配置阈值；pandas：根据阈值批量筛选异常数据；
paramiko/netmiko：远程拉取日志文本；pandas：结构化解析日志做聚类统计。

关联文档

Python 面向对象编程详解

Python 基础

Pandas 数据分析详解

1. 基础定位与核心作用

1.1. 定义

1.2. 适配你的运维/AIOps 典型场景

1.3. 安装

1.4. 导入惯例（行业统一写法）

2. 两大核心数据结构

2.1. Series 一维序列（单列）

2.2. DataFrame 二维数据表（重点，99%场景使用）

2.2.1. 快速创建DataFrame

3. 数据读取与导出（运维高频）

3.1. 读取外部数据源

3.1.1. 1）CSV 监控/日志文件

3.1.2. 2）从 MySQL/ClickHouse 读取（搭配SQL驱动）

3.1.3. 3）列表/字典直接生成（psutil采集数据）

3.2. 导出保存

4. 基础查看与数据筛选

4.1. 基础信息查看

4.2. 取列操作

4.3. 行筛选（布尔索引，AIOps异常过滤核心）

4.3.1. 示例1：筛选CPU > 80 的异常主机

4.3.2. 多条件组合 & 且 / | 或

4.3.3. 模糊匹配日志文本（日志分析）

4.4. loc / iloc 精准切片

5. 清洗数据（运维日志/监控必做步骤）

5.1. 空值处理（采集缺失指标）

5.2. 去重（重复采集指标、重复告警）

5.3. 删除无用列、行

5.4. 新增计算列（衍生指标）

5.5. 类型转换

6. 分组聚合（统计分析核心，AIOps高频）

6.1. groupby 分组统计：按主机、按分钟、按业务线聚合指标

6.1.1. 基础聚合函数

6.1.2. 日志场景：统计每种报错出现次数

7. 时序时间序列专属功能（你的AIOps核心重点）

7.1. 时间索引重采样（分钟→小时聚合）

7.2. 时间范围筛选

7.3. 滑动窗口计算（动态基线、时序异常检测必备）

8. 表关联 merge（多源监控数据合并）

9. 排序、TopN 高负载主机

10. 运维/AIOps完整实战示例：时序异常标记

11. 生产避坑与性能优化

12. pandas 在自研AIOps产品中的定位

13. 对比你已学工具

目录

4.3.2. 多条件组合 `&` 且 / `|` 或