Changes between Version 7 and Version 8 of Workshops/JobCheckpointing/Examples

Timestamp:

04/02/2026 09:04:44 PM (4 months ago)

Author:

Carl Baribault

Comment:

Added Python scripts for post-workflow analysis

Workshops/JobCheckpointing/Examples

@@ -472 +472 @@
 }}}
 
+=== Post-workflow analysis in Python ===
+
+Here are two Python scripts used by the checkpoint submitter to analyze and plot usage and annotate anomalies in execution time and memory usage by the checkpoint runner jobs.
+
+==== Aggregate usage ====
+
+Here is the listing for the file '''aggregate_usage.py''' used in [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh].
+
+===== aggregate_usage.py =====
+
+{{{
+#!/usr/bin/env python3
+"""
+aggregate_usage.py  (SLURM 14.x–compatible)
+-------------------------------------------------------------
+Combines parent and .batch accounting rows so MaxRSS is preserved.
+
+SLURM 14.x specifics:
+  - Only .batch steps contain correct MaxRSS
+  - Parent jobs often have empty MaxRSS
+  - ElapsedRaw is not available
+  - Elapsed may be MM, MM:SS, HH:MM:SS, D-HH, D-HH:MM, D-HH:MM:SS
+
+Outputs worker_usage.csv:
+  JobID, ElapsedSeconds, TotalCPU, MaxRSS_bytes, MaxRSS_MiB,
+  ReqMem, NodeList, State
+"""
+
+import subprocess
+import pandas as pd
+import sys
+import re
+from io import StringIO
+
+IDFILE = "worker_ids.txt"
+OUTFILE = "worker_usage.csv"
+
+
+# ---------------------------------------------------------
+# Utilities
+# ---------------------------------------------------------
+def run_cmd(cmd):
+    return subprocess.check_output(cmd, shell=True, text=True)
+
+
+def parse_elapsed_to_seconds(elapsed):
+    if not isinstance(elapsed, str):
+        return None
+
+    s = elapsed.strip()
+    days = 0
+
+    # Day prefix
+    if "-" in s:
+        d, rest = s.split("-", 1)
+        try:
+            days = int(d)
+        except:
+            return None
+    else:
+        rest = s
+
+    parts = rest.split(":")
+
+    if len(parts) == 1:      # MM or D-HH
+        try:
+            x = int(parts[0])
+        except:
+            return None
+        if days > 0:
+            hours, minutes, seconds = x, 0, 0
+        else:
+            hours, minutes, seconds = 0, x, 0
+
+    elif len(parts) == 2:    # MM:SS or D-HH:MM
+        try:
+            a = int(parts[0])
+            b = int(parts[1])
+        except:
+            return None
+        if days > 0:
+            hours, minutes, seconds = a, b, 0
+        else:
+            hours, minutes, seconds = 0, a, b
+
+    elif len(parts) == 3:    # HH:MM:SS or D-HH:MM:SS
+        try:
+            hours = int(parts[0])
+            minutes = int(parts[1])
+            seconds = int(parts[2])
+        except:
+            return None
+    else:
+        return None
+
+    return days*86400 + hours*3600 + minutes*60 + seconds
+
+
+def parse_rss_to_bytes(val):
+    if val is None or pd.isna(val):
+        return None
+    s = str(val).strip().rstrip("nN")
+    try:
+        if s.endswith(("K","k")):
+            return float(s[:-1]) * 1e3
+        if s.endswith(("M","m")):
+            return float(s[:-1]) * 1e6
+        if s.endswith(("G","g")):
+            return float(s[:-1]) * 1e9
+        if re.fullmatch(r"\d+(\.\d+)?", s):
+            return float(s)
+    except:
+        return None
+    return None
+
+
+# ---------------------------------------------------------
+# Main
+# ---------------------------------------------------------
+def main():
+
+    try:
+        with open(IDFILE) as f:
+            jobids = [x.strip() for x in f if x.strip()]
+    except:
+        print(f"ERROR: {IDFILE} not found.")
+        sys.exit(1)
+
+    if not jobids:
+        print("No job IDs to process.")
+        sys.exit(0)
+
+    fields = [
+        "JobID", "Elapsed", "TotalCPU", "MaxRSS",
+        "ReqMem", "NodeList", "State"
+    ]
+
+    cmd = f"sacct -j {','.join(jobids)} --format={','.join(fields)} -P"
+    raw = run_cmd(cmd)
+
+    df = pd.read_csv(StringIO(raw), sep="|")
+    df.dropna(how="all", inplace=True)
+
+    # Split BaseID with explicit keywords (older pandas compatibility)
+    df["BaseID"] = (
+        df["JobID"].astype(str)
+        .str.split(pat=".", n=1, regex=False)
+        .str[0]
+    )
+
+    df["IsBatch"] = df["JobID"].astype(str).str.endswith(".batch")
+
+    parent = df[~df["IsBatch"]].copy()
+    batch  = df[df["IsBatch"]].copy()
+
+    merged = pd.merge(
+        parent,
+        batch[["BaseID","MaxRSS"]],
+        on="BaseID",
+        how="left",
+        suffixes=("", "_batch")
+    )
+
+    def choose(x):
+        if pd.notna(x["MaxRSS"]):
+            return x["MaxRSS"]
+        return x["MaxRSS_batch"]
+
+    merged["MaxRSS"] = merged.apply(choose, axis=1)
+
+    merged["MaxRSS_bytes"] = merged["MaxRSS"].apply(parse_rss_to_bytes)
+    merged["MaxRSS_MiB"] = merged["MaxRSS_bytes"] / (1024.0**2)
+
+    merged["ElapsedSeconds"] = merged["Elapsed"].apply(parse_elapsed_to_seconds)
+
+    out = merged[[
+        "BaseID","ElapsedSeconds","TotalCPU","MaxRSS","MaxRSS_bytes",
+        "MaxRSS_MiB","ReqMem","NodeList","State"
+    ]].rename(columns={"BaseID": "JobID"})
+
+    out.to_csv(OUTFILE, index=False)
+    print(f"Wrote {OUTFILE}")
+    print(out.head())
+
+
+if __name__ == "__main__":
+    main()
+}}}
+
+==== Plot progress ====
+
+Here is the listing for the file '''plot_progress.py''' used in [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] for generating .png plot files for run time and memory usage by the checkpoint runner jobs.
+
+===== plot_progress.py =====
+
+{{{
+#!/usr/bin/env python3
+"""
+plot_progress.py
+
+Produces:
+  - progress_with_anomalies.png     (dual-axis: checkpoints + MaxRSS MiB)
+  - resource_trend.png              (dual-axis: ElapsedSeconds + MaxRSS MiB)
+  - anomaly_details.csv             (per-job anomalies)
+  - anomaly_stats_table.png         (summary table)
+"""
+
+import pandas as pd
+import matplotlib.pyplot as plt
+import datetime as dt
+import numpy as np
+import glob
+import re
+import os
+
+USAGE_FILE = "worker_usage.csv"
+LOG_GLOB   = "log_runner_*.out"
+
+
+# ---------------------------------------------------------
+# Utilities
+# ---------------------------------------------------------
+def parse_worker_logs():
+    rows = []
+    for fp in glob.glob(LOG_GLOB):
+        m = re.search(r"log_runner_(\d+)\.out", os.path.basename(fp))
+        if not m: continue
+        jid = int(m.group(1))
+
+        with open(fp, "r", errors="ignore") as f:
+            for line in f:
+                if "Info[" in line:
+                    try:
+                        ts = line.split("Info[")[1].split("]")[0]
+                        t  = dt.datetime.strptime(ts,"%Y%m%d-%H:%M:%S")
+                        rows.append((jid,t))
+                        break
+                    except:
+                        pass
+    return pd.DataFrame(rows, columns=["JobID","StartTime"]) if rows else \
+           pd.DataFrame(columns=["JobID","StartTime"])
+
+
+def numeric_jobid(s):
+    m = re.match(r"^(\d+)", str(s))
+    return int(m.group(1)) if m else None
+
+
+def time_anomalies(elapsed):
+    s = elapsed.astype(float)
+    if s.std(ddof=0)==0:
+        return pd.Series(False,index=s.index), pd.Series(0.0,index=s.index)
+    z = (s-s.mean())/s.std(ddof=0)
+    return (z.abs()>3.5), z
+
+
+def mem_anomalies(mem_mib, window=10, thresh=3.5):
+    s = mem_mib.astype(float)
+    med = s.rolling(window,min_periods=3).median()
+    absd = (s-med).abs()
+    mad  = absd.rolling(window,min_periods=3).median()
+    modz = 0.6745 * absd / mad.replace(0,np.nan)
+    modz = modz.fillna(0)
+    return (modz>thresh), modz
+
+
+def save_summary(summary, details, fname="anomaly_stats_table.png"):
+    fig, ax = plt.subplots(figsize=(11,6))
+    ax.axis("off")
+
+    rows = [[k,str(v)] for k,v in summary.items()]
+    stab = ax.table(
+        cellText=rows,
+        colLabels=["Metric","Value"],
+        bbox=[0,0.55,0.45,0.4]
+    )
+    stab.auto_set_font_size(False)
+    stab.set_fontsize(11)
+
+    anom = details[details["AnomalyType"]!="none"]
+    if len(anom)>12: anom = anom.head(12)
+    if not anom.empty:
+        atab = ax.table(
+            cellText=anom.astype(str).values,
+            colLabels=list(anom.columns),
+            bbox=[0,0.05,1,0.45]
+        )
+        atab.auto_set_font_size(False)
+        atab.set_fontsize(9)
+
+    fig.tight_layout()
+    fig.savefig(fname,dpi=150)
+    plt.close(fig)
+    print(f"Saved {fname}")
+
+
+# ---------------------------------------------------------
+# Main
+# ---------------------------------------------------------
+def main():
+
+    df = pd.read_csv(USAGE_FILE)
+    df = df[df["ElapsedSeconds"].notnull()].reset_index(drop=True)
+
+    if "MaxRSS_MiB" not in df.columns:
+        print("ERROR: worker_usage.csv missing MaxRSS_MiB.")
+        return
+
+    df["JobID"] = df["JobID"].apply(numeric_jobid)
+    df = df[df["JobID"].notnull()].reset_index(drop=True)
+
+    logs = parse_worker_logs()
+    merged = pd.merge(df, logs, on="JobID", how="inner")
+
+    if merged.empty:
+        print("ERROR: No matching logs.")
+        return
+
+    merged = merged.sort_values("StartTime").reset_index(drop=True)
+
+    times   = merged["StartTime"].tolist()
+    idxs    = np.arange(len(merged))
+    elapsed = merged["ElapsedSeconds"].astype(float)
+    mem_mib = merged["MaxRSS_MiB"].astype(float)
+
+    # Detect anomalies
+    mtime, ztime = time_anomalies(elapsed)
+    mmem,  zmem  = mem_anomalies(mem_mib)
+
+    both  = mtime & mmem
+    tonly = mtime & ~mmem
+    monly = mmem  & ~mtime
+
+    # -----------------------------------------------------
+    # 1. progress_with_anomalies.png
+    # -----------------------------------------------------
+    fig, ax1 = plt.subplots(figsize=(12,6))
+
+    # Checkpoint curve with dots
+    ax1.plot(times, idxs, 'o-', color="gray", label="Checkpoint Progress")
+    ax1.set_xlabel("Wall Time")
+    ax1.set_ylabel("Checkpoint Number",color="gray")
+
+    # Anomaly markers
+    for i,t in enumerate(times):
+        if both[i]:
+            ax1.plot(t, idxs[i], 'o', color="purple", markersize=10,
+                     label="Both anomalies" if i==0 else "")
+        elif tonly[i]:
+            ax1.plot(t, idxs[i], 'o', color="orange", markersize=9,
+                     label="Time anomaly" if i==0 else "")
+        elif monly[i]:
+            ax1.plot(t, idxs[i], 'o', color="blue", markersize=9,
+                     label="Memory anomaly" if i==0 else "")
+
+    # MaxRSS dual-axis with dots
+    ax2 = ax1.twinx()
+    ax2.plot(times, mem_mib.values, 'bo-', markersize=5,
+             alpha=0.70, label="MaxRSS (MiB)")
+    ax2.set_ylabel("MaxRSS (MiB)", color="blue")
+
+    # Highlight memory anomalies on memory curve
+    for i,t in enumerate(times):
+        if monly[i]:
+            ax2.plot(t, mem_mib.values[i], 'bo', markersize=8)
+        elif both[i]:
+            ax2.plot(t, mem_mib.values[i], 'mo', markersize=10)
+
+    h1,l1 = ax1.get_legend_handles_labels()
+    h2,l2 = ax2.get_legend_handles_labels()
+    ax1.legend(h1+h2, l1+l2, loc="upper left")
+
+    fig.tight_layout()
+    fig.savefig("progress_with_anomalies.png")
+    plt.close(fig)
+    print("Saved progress_with_anomalies.png")
+
+    # -----------------------------------------------------
+    # 2. resource_trend.png  (Elapsed vs MaxRSS dual-axis)
+    # -----------------------------------------------------
+    fig2, axL = plt.subplots(figsize=(12,6))
+
+    axL.plot(elapsed.values, 'ko-', markersize=5,
+             label="Elapsed (sec)")
+    axL.set_xlabel("Worker Job Index (time-ordered)")
+    axL.set_ylabel("Elapsed Seconds",color="black")
+
+    axR = axL.twinx()
+    axR.plot(mem_mib.values, 'bo-', markersize=5,
+             alpha=0.7, label="MaxRSS (MiB)")
+    axR.set_ylabel("MaxRSS (MiB)",color="blue")
+
+    hL,lL = axL.get_legend_handles_labels()
+    hR,lR = axR.get_legend_handles_labels()
+    axL.legend(hL+hR, lL+lR, loc="upper left")
+
+    fig2.tight_layout()
+    fig2.savefig("resource_trend.png")
+    plt.close(fig2)
+    print("Saved resource_trend.png")
+
+    # -----------------------------------------------------
+    # 3. anomaly_details + summary table
+    # -----------------------------------------------------
+    details = pd.DataFrame({
+        "JobID": merged["JobID"],
+        "StartTime": merged["StartTime"],
+        "ElapsedSeconds": elapsed,
+        "MaxRSS_MiB": mem_mib,
+        "TimeZ": ztime,
+        "MemModZ": zmem
+    })
+
+    atype = np.where(both, "both",
+             np.where(tonly, "time",
+             np.where(monly,"memory","none")))
+    details["AnomalyType"] = atype
+
+    summary = {
+        "Total jobs": len(details),
+        "Anomalies": int((atype!="none").sum()),
+        "Time only": int((atype=="time").sum()),
+        "Memory only": int((atype=="memory").sum()),
+        "Both": int((atype=="both").sum())
+    }
+
+    details.to_csv("anomaly_details.csv", index=False)
+    print("Saved anomaly_details.csv")
+
+    save_summary(summary, details)
+
+
+if __name__ == "__main__":
+    main()
+}}}
+
 == Working Examples ==
 
-=== Make copies of checkpoint_submitter.sh and checkpoint_runner.sh
-
-All of the following examples make use of the two job scripts mentioned above.
-
-* the [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] and
-* the [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh] 
+=== Make copies of the BASH scripts checkpoint_submitter.sh and checkpoint_runner.sh
+
+All of the examples linked further below make use of the following two (2) BASH job scripts mentioned above.
+
+* [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh]
+* [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh] 
+
+=== Make copies of the Python scripts aggregate_usage.py and plot_progress.py
+
+Also, the '''checkpoint_submitter.sh''' script calls out the following two (2) Python scripts for post-workflow analysis of execution time and memory usage by the checkpoint runner jobs.
+
+* [wiki://Workshops/JobCheckpointing/Examples#aggregate_usage.py aggregate_usage.py]
+* [wiki://Workshops/JobCheckpointing/Examples#plot_progress.py plot_progress.py]
 
 === BASH Checkpointing Example ===