Python 数据处理¶

运维中的数据处理通常不是训练复杂模型，而是把 CSV、JSON、Excel、日志或接口数据清洗成可信的明细、汇总表和可导入数据。核心流程是：先确认数据口径，再读取、校验、转换、关联、汇总和输出。

原始文件 / API / 日志
  → 编码、列名、类型与必填字段检查
  → 去重、缺失值和异常值处理
  → 表关联、分组与聚合
  → 结果校验
  → CSV / Excel / JSON / 数据库

使用 uv 准备项目¶

uv init --no-package ops-data
cd ops-data
uv python pin 3.12
uv add pandas openpyxl pyyaml requests
uv add --dev pytest ruff
uv run python main.py

包	主要用途
`pandas`	表格读取、过滤、关联、分组和统计
`openpyxl`	读写 `.xlsx`，保留较基础的 Excel 结构
`requests`	从 HTTP API 获取数据
`PyYAML`	读取 YAML 配置和字段映射
Python 标准库	`csv`、`json`、`pathlib`、`logging`、`sqlite3`

CSV 清洗示例¶

from pathlib import Path
import pandas as pd

input_file = Path("input/devices.csv")
output_file = Path("output/device-summary.csv")
output_file.parent.mkdir(parents=True, exist_ok=True)

df = pd.read_csv(
    input_file,
    dtype={"device_id": "string", "site_id": "string"},
    encoding="utf-8-sig",
)
df.columns = df.columns.str.strip().str.lower()

required = {"device_id", "site_id", "vendor", "status"}
missing_columns = required - set(df.columns)
if missing_columns:
    raise ValueError(f"缺少必要字段: {sorted(missing_columns)}")

df["vendor"] = df["vendor"].str.strip().str.lower()
df["status"] = df["status"].fillna("unknown").str.strip().str.lower()
df = df.drop_duplicates(subset=["device_id"], keep="last")

summary = (
    df.groupby(["vendor", "status"], dropna=False)
      .size()
      .reset_index(name="device_count")
      .sort_values(["vendor", "status"])
)

summary.to_csv(output_file, index=False, encoding="utf-8-sig")

显式指定 ID 为字符串非常重要。设备号、手机号或工单号即使只包含数字，也通常不是数学数值；自动推断为整数可能丢失前导零。

JSON 与嵌套字段¶

import json
import pandas as pd

with open("input/alarms.json", encoding="utf-8") as file:
    payload = json.load(file)

df = pd.json_normalize(
    payload["items"],
    sep=".",
)

df[["alarm_id", "device.name", "severity", "occurred_at"]].to_csv(
    "output/alarms.csv",
    index=False,
    encoding="utf-8-sig",
)

处理 API 数据时还要保存抓取时间、接口参数、分页位置和原始响应校验信息，防止结果异常后无法重放。

表关联¶

devices = pd.read_csv("input/devices.csv", dtype={"site_id": "string"})
sites = pd.read_excel("input/sites.xlsx", dtype={"site_id": "string"})

result = devices.merge(
    sites[["site_id", "site_name", "region"]],
    on="site_id",
    how="left",
    validate="many_to_one",
    indicator=True,
)

unmatched = result[result["_merge"] != "both"]
if not unmatched.empty:
    unmatched.to_csv("output/unmatched-sites.csv", index=False)

validate="many_to_one" 可以检查站点表是否意外出现重复键；indicator=True 用于找出关联失败的记录。不要 merge 完看到行数增加后还直接发布结果。

数据质量检查¶

每次处理至少检查：

检查项	示例
输入规模	文件数、行数、时间范围是否符合预期
唯一键	`device_id`、工单号是否重复
必填字段	站点、时间、状态是否为空
类型	ID 不转数字，时间解析失败要单独输出
枚举	厂商、状态、级别是否出现未知值
关联结果	左表行数是否变化，未匹配数量是否可接受
输出守恒	汇总合计是否等于清洗后的明细数量

发现脏数据时，优先输出异常明细和原因，不要静默删除。这样才能反馈源系统，而不是让脚本每次悄悄“修正”不同问题。

大文件处理¶

数据无法一次放入内存时按块读取：

import pandas as pd

counts = {}
for chunk in pd.read_csv("input/large.csv", chunksize=100_000):
    current = chunk["status"].fillna("unknown").value_counts()
    for status, count in current.items():
        counts[status] = counts.get(status, 0) + int(count)

print(counts)

若处理长期增长、复杂查询或多次重复分析的数据，应考虑先导入 SQLite/PostgreSQL 或列式文件，而不是每次完整扫描超大 Excel。

输出与安全¶

输出写到新文件，校验成功后再替换旧结果。
文件名包含日期、数据版本或批次号，保留输入与脚本版本映射。
Excel 单表有行数限制，大结果优先 CSV、数据库或 Parquet。
CSV 若会被人工打开，注意 UTF-8 BOM、日期格式和长数字被 Excel 转换。
对姓名、电话、账号和 Token 做脱敏，日志不要打印整行敏感数据。
自动任务必须使用非零退出码表示失败，不能只打印错误后继续生成“成功”文件。