批处理任务运行分析：从原始记录到可验证结论

生成或接收运行记录
        │
        ▼
字段、类型与质量检查 ──► quality_report.csv
        │
        ▼
去重、缺失处理、异常标记
        │
        ├──► worker_summary.csv
        ├──► daily_summary.csv
        ├──► analysis_summary.txt
        └──► figures/*.png

batch-run-analysis/
├── data/
├── output/
│   └── figures/
├── prepare_sample.py
├── analyze.py
└── requirements.txt

pandas
numpy
matplotlib
seaborn

# prepare_sample.py
from pathlib import Path

import pandas as pd

rows = [
    ("run-001", "2026-07-01 09:00", "worker-a", 12, 18.0, 42.0, "success", 0),
    ("run-002", "2026-07-01 10:10", "worker-b", 20, 35.0, 70.0, "success", 0),
    ("run-003", "2026-07-01 11:20", "worker-c", 8, 10.0, 31.0, "failed", 2),
    ("run-004", "2026-07-02 09:15", "worker-a", 30, 52.0, 96.0, "success", 1),
    ("run-005", "2026-07-02 10:30", "worker-b", 16, 22.0, 55.0, "success", 0),
    ("run-006", "2026-07-02 13:00", "worker-c", 25, 41.0, 88.0, "failed", 3),
    ("run-007", "2026-07-03 08:50", "worker-a", 10, 14.0, 36.0, "success", 0),
    ("run-008", "2026-07-03 10:40", "worker-b", 35, 61.0, 112.0, "success", 1),
    ("run-009", "2026-07-03 15:20", "worker-c", 18, 29.0, 67.0, "success", 0),
    ("run-010", "2026-07-04 09:05", "worker-a", 40, 74.0, 138.0, "success", 1),
    ("run-010", "2026-07-04 09:05", "worker-a", 40, 74.0, 138.0, "success", 1),
    ("run-011", "invalid-time", "worker-b", 14, 19.0, 48.0, "failed", 2),
    ("run-012", "2026-07-04 12:10", "worker-c", 22, 38.0, None, "success", 0),
    ("run-013", "2026-07-04 16:30", "worker-b", 60, 90.0, 620.0, "success", 2),
]

columns = [
    "run_id", "started_at", "worker", "file_count", "input_mb",
    "duration_sec", "status", "retry_count",
]

output = Path("data/batch_runs.csv")
output.parent.mkdir(parents=True, exist_ok=True)
pd.DataFrame(rows, columns=columns).to_csv(output, index=False)
print(f"created {output} with {len(rows)} rows")

# analyze.py
from pathlib import Path

import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns

DATA_PATH = Path("data/batch_runs.csv")
OUTPUT_DIR = Path("output")
FIGURE_DIR = OUTPUT_DIR / "figures"
REQUIRED_COLUMNS = {
    "run_id", "started_at", "worker", "file_count", "input_mb",
    "duration_sec", "status", "retry_count",
}


def load_data(path: Path) -> pd.DataFrame:
    data = pd.read_csv(path, dtype={"run_id": "string", "worker": "string"})
    missing = REQUIRED_COLUMNS - set(data.columns)
    if missing:
        raise ValueError(f"缺少必需列: {', '.join(sorted(missing))}")
    return data


def clean_data(raw: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    data = raw.copy()
    data["started_at"] = pd.to_datetime(data["started_at"], errors="coerce")
    quality = pd.DataFrame(
        {
            "metric": ["raw_rows", "duplicate_ids", "invalid_times", "missing_duration"],
            "value": [
                len(data),
                int(data.duplicated(subset="run_id").sum()),
                int(data["started_at"].isna().sum()),
                int(data["duration_sec"].isna().sum()),
            ],
        }
    )

    data = data.drop_duplicates(subset="run_id", keep="first")
    data = data.dropna(subset=["started_at", "input_mb", "duration_sec"]).copy()
    valid_status = {"success", "failed"}
    if not set(data["status"]).issubset(valid_status):
        raise ValueError("status 只能是 success 或 failed")
    if (data[["file_count", "input_mb", "duration_sec", "retry_count"]] < 0).any().any():
        raise ValueError("数值指标不能为负数")

    q1 = data["duration_sec"].quantile(0.25)
    q3 = data["duration_sec"].quantile(0.75)
    iqr = q3 - q1
    data["duration_outlier"] = ~data["duration_sec"].between(
        q1 - 1.5 * iqr, q3 + 1.5 * iqr
    )
    data["day"] = data["started_at"].dt.date
    data["is_success"] = data["status"].eq("success")
    return data, quality


def summarize(data: pd.DataFrame):
    regular = data.loc[~data["duration_outlier"]]
    worker = (
        regular.groupby("worker", as_index=False)
        .agg(
            run_count=("run_id", "count"),
            success_rate=("is_success", "mean"),
            median_duration=("duration_sec", "median"),
        )
        .sort_values(["success_rate", "median_duration"], ascending=[False, True])
    )
    daily = (
        data.groupby("day", as_index=False)
        .agg(run_count=("run_id", "count"), failed_count=("is_success", lambda x: (~x).sum()))
    )
    correlation = float(regular["input_mb"].corr(regular["duration_sec"]))
    return worker, daily, correlation


def draw_figures(data: pd.DataFrame, worker: pd.DataFrame, daily: pd.DataFrame) -> None:
    FIGURE_DIR.mkdir(parents=True, exist_ok=True)
    sns.set_theme(style="whitegrid")
    regular = data.loc[~data["duration_outlier"]]

    figure, axis = plt.subplots(figsize=(7, 4))
    sns.histplot(data=data, x="duration_sec", bins=8, ax=axis)
    axis.set(title="Duration distribution", xlabel="Seconds", ylabel="Count")
    figure.tight_layout(); figure.savefig(FIGURE_DIR / "duration_distribution.png", dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(7, 4))
    sns.scatterplot(data=regular, x="input_mb", y="duration_sec", hue="worker", ax=axis)
    axis.set(title="Input size and duration", xlabel="Input MB", ylabel="Seconds")
    figure.tight_layout(); figure.savefig(FIGURE_DIR / "size_duration.png", dpi=150)
    plt.close(figure)

    figure, axis = plt.subplots(figsize=(7, 4))
    sns.barplot(data=worker, x="worker", y="median_duration", ax=axis)
    axis.set(title="Median duration by worker", xlabel="Worker", ylabel="Seconds")
    figure.tight_layout(); figure.savefig(FIGURE_DIR / "worker_duration.png", dpi=150)
    plt.close(figure)


def main() -> None:
    OUTPUT_DIR.mkdir(exist_ok=True)
    raw = load_data(DATA_PATH)
    clean, quality = clean_data(raw)
    worker, daily, correlation = summarize(clean)
    quality.to_csv(OUTPUT_DIR / "quality_report.csv", index=False)
    worker.to_csv(OUTPUT_DIR / "worker_summary.csv", index=False)
    daily.to_csv(OUTPUT_DIR / "daily_summary.csv", index=False)
    draw_figures(clean, worker, daily)

    summary = (
        f"raw_rows={len(raw)}\nanalysis_rows={len(clean)}\n"
        f"duration_outliers={int(clean['duration_outlier'].sum())}\n"
        f"input_duration_correlation={correlation:.3f}\n"
    )
    (OUTPUT_DIR / "analysis_summary.txt").write_text(summary, encoding="utf-8")
    print(summary)
    print(worker.to_string(index=False))


if __name__ == "__main__":
    main()

import pandas as pd


def test_clean_data_reports_invalid_rows() -> None:
    raw = pd.DataFrame([
        {
            "run_id": "a", "started_at": "2026-01-01", "worker": "w1",
            "file_count": 1, "input_mb": 2.0, "duration_sec": 3.0,
            "status": "success", "retry_count": 0,
        },
        {
            "run_id": "b", "started_at": "bad", "worker": "w1",
            "file_count": 1, "input_mb": 2.0, "duration_sec": 3.0,
            "status": "failed", "retry_count": 1,
        },
    ])
    clean, quality = clean_data(raw)
    invalid = quality.loc[quality["metric"] == "invalid_times", "value"].item()
    assert invalid == 1
    assert len(clean) == 1

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python prepare_sample.py
python analyze.py
find output -maxdepth 2 -type f -print