Job Management Library

1 Overview

The jd-worker package exposes two complementary APIs:

2 Setup

Create a .env file

Keep secrets out of source code. Point jd.init() at this file:

# bloomf.env — do not commit to git
JD_API_KEY=jd_your_hub_api_key_here
JD_EXP_ID=bloomf
JD_WORKSPACE_PATH=/Users/you/projects          # optional; default ~
# JD_HUB_URL=https://hub.jobdistributor.net    # optional

Create an API key in the Hub under API Keys. JD_EXP_ID must match your experiment name exactly. JD_WORKSPACE_PATH is the parent directory where local downloads land (…/jd_data/<expId>/<job_id>/downloaded/).

jd.init()

Call once at the start of your orchestrator script or notebook:

import jd

jd.init("bloomf.env")   # or jd.init() if .env is in cwd

# Optional overrides:
# jd.init("bloomf.env", exp_id="other_exp", hub_url="https://hub.jobdistributor.net")

init exchanges your Hub API key for a short-lived worker JWT and discovers the experiment's job-server URL. Tokens refresh automatically for long-running scripts.

3 Path helpers

These resolve local directories on your machine (not on workers):

jd.data_root()              # …/jd_data
jd.exp_path()               # …/jd_data/<expId>/
jd.job_path(42)             # …/jd_data/<expId>/42/
jd.job_download_dir(42)     # …/jd_data/<expId>/42/downloaded/

download_result writes into job_download_dir by default, keeping one folder per job id for easy inspection.

4 create_jobs()

Submit jobs in three equivalent ways:

Parameter grid (Cartesian product)

jd.create_jobs({
    "lr":     [1e-3, 1e-4],
    "epochs": [10, 20],
})
# → 4 jobs

Explicit job list

jd.create_jobs([
    {"lr": "0.001", "epochs": "10", "seed": "1"},
    {"lr": "0.0001", "epochs": "20", "seed": "2"},
])

File upload

jd.create_jobs("batch1.csv")    # header row = param names
jd.create_jobs("batch1.json")   # {"jobs": [...]} or {"parameters": {...}}
jd.create_jobs("batch1.tsv")

By default jobs are appended to the queue. Pass replace=True to wipe existing jobs first (same as dashboard "Replace all").

result = jd.create_jobs({"lr": [1e-3]}, replace=True)
print(result["total_jobs"], result["action"])  # e.g. 1, "Created"

Getting the IDs of newly created jobs

For append operations the response always includes start_id — the first ID assigned to the batch. Because the server assigns IDs as a contiguous range in a single atomic INSERT, you can reconstruct the full list without any additional request:

result = jd.create_jobs([
    {"lr": "0.001", "seed": "1"},
    {"lr": "0.0001", "seed": "2"},
    {"lr": "0.00001", "seed": "3"},
])
start_id   = result["start_id"]    # e.g. 42
total_jobs = result["total_jobs"]  # e.g. 3
job_ids    = list(range(start_id, start_id + total_jobs))  # [42, 43, 44]

5 list_jobs()

page = jd.list_jobs(status="DONE", page=1, per_page=50)
for job in page["jobs"]:
    print(job["id"], job["status"], job["parameters"])

# Fetch every matching job (auto-paginate):
all_done = jd.list_jobs(status="DONE", fetch_all=True)

6 download_result()

Download the latest version of a logical filename (e.g. metrics.json even if the worker uploaded metrics_v2.json):

path = jd.download_result(job_id=42, filename="metrics.json")
# → …/jd_data/<expId>/42/downloaded/metrics.json

# Custom destination:
path = jd.download_result(42, "metrics.json", dest="./analysis/run42.json")

7 wait_for_jobs()

Poll until specific job ids reach a terminal status:

job_ids = [101, 102, 103]
finished = jd.wait_for_jobs(job_ids, status="DONE", timeout=3600, poll_interval=30)
for jid, job in finished.items():
    print(jid, job["status"], job.get("message"))

When waiting for DONE, ABORTED jobs are also treated as finished so your script does not hang on failed runs.

8 Iterative batch workflow

The pattern below is the core use case: run a batch on workers, analyze results locally, decide the next hyperparameters, and repeat. This example performs a simple two-round learning-rate search.

Worker entry script (train.py)

Runs on each worker — reads job parameters from the environment, writes a result file:

"""train.py — invoked by jd_worker_cli for each job."""
import json
import os
import random

from jd import jd_job_dir, jd_upload

def main():
    params = json.loads(os.environ.get("JD_JOB_PARAMETERS", "{}"))
    lr = float(params.get("lr", 1e-3))
    epochs = int(params.get("epochs", 5))
    seed = int(params.get("seed", 0))

    random.seed(seed)
    # … your real training loop here …
    # Stand-in metric: lower lr → slightly better loss in this toy example
    loss = abs(lr - 0.0003) + random.uniform(0, 0.01)

    out = jd_job_dir() / "metrics.json"
    out.write_text(json.dumps({
        "lr": lr,
        "epochs": epochs,
        "seed": seed,
        "loss": loss,
    }))
    jd_upload(str(out))

if __name__ == "__main__":
    main()

Start workers as usual:

jd_worker_cli run --expId bloomf --script train.py --workers 4

Orchestrator script (orchestrate.py)

Runs on your machine — drives the experiment in rounds:

"""
orchestrate.py — iterative hyperparameter search using the job-management library.

Round 1: coarse grid → pick best lr
Round 2: fine grid around winner → pick final config
"""
import json
from pathlib import Path

import jd

jd.init("bloomf.env")

RESULT_FILE = "metrics.json"
TOP_K = 2


def job_ids_from(result: dict) -> list[int]:
    """Reconstruct the exact IDs created by an append create_jobs() call."""
    start_id = result["start_id"]
    total    = result["total_jobs"]
    return list(range(start_id, start_id + total))


def collect_results(job_ids: list[int]) -> list[dict]:
    """Download metrics.json for each job and parse."""
    rows = []
    for jid in job_ids:
        path = jd.download_result(jid, RESULT_FILE)
        data = json.loads(Path(path).read_text())
        data["job_id"] = jid
        rows.append(data)
    return rows


def pick_best_lrs(rows: list[dict], k: int = TOP_K) -> list[float]:
    rows.sort(key=lambda r: r["loss"])
    return [r["lr"] for r in rows[:k]]


def fine_grid_around(center: float, deltas=(0.5, 0.25, 0.75)) -> list[float]:
    return sorted({round(center * d, 6) for d in deltas})


# ── Round 1: coarse search ───────────────────────────────────────────────
print("=== Round 1: coarse grid ===")
result1    = jd.create_jobs({
    "lr":     [1e-2, 1e-3, 1e-4, 1e-5],
    "epochs": [5],
    "seed":   [0, 1],
}, replace=True)
batch1_ids = job_ids_from(result1)
print(f"Created {len(batch1_ids)} jobs: {batch1_ids}")

jd.wait_for_jobs(batch1_ids, timeout=7200)
round1   = collect_results(batch1_ids)
best_lrs = pick_best_lrs(round1)
print(f"Round 1 best lrs: {best_lrs}")

# ── Round 2: refine around winners ─────────────────────────────────────────
print("\n=== Round 2: fine grid ===")
round2_jobs = [
    {"lr": str(fine_lr), "epochs": "10", "seed": "0", "round": "2"}
    for lr in best_lrs
    for fine_lr in fine_grid_around(lr)
]

result2    = jd.create_jobs(round2_jobs)  # append — keeps round 1 history in DB
batch2_ids = job_ids_from(result2)
print(f"Appended {len(batch2_ids)} jobs: {batch2_ids}")

jd.wait_for_jobs(batch2_ids, timeout=7200)
round2 = collect_results(batch2_ids)
round2.sort(key=lambda r: r["loss"])

winner = round2[0]
print(f"\nWinner: job {winner['job_id']}  lr={winner['lr']}  loss={winner['loss']:.4f}")

# Save summary next to downloaded artifacts
summary_path = jd.exp_path() / "search_summary.json"
summary_path.write_text(json.dumps({
    "round1_best_lrs": best_lrs,
    "winner": winner,
    "round2_top3": round2[:3],
}, indent=2))
print(f"Summary written to {summary_path}")

Flow diagram

┌─────────────┐     create_jobs      ┌──────────────┐
│ Orchestrator│ ───────────────────► │  Job server  │
│  (your PC)  │                      │   + queue    │
└──────┬──────┘                      └──────┬───────┘
       │                                    │ serve jobs
       │ wait_for_jobs                      ▼
       │◄─────────────────────────── ┌──────────────┐
       │                             │   Workers    │
       │ download_result             │  (train.py)  │
       │◄─────────────────────────── │  jd_upload   │
       │                             └──────────────┘
       │ post-process (Python)
       │ decide next batch
       └──────── create_jobs (append) ──► … repeat …

9 Patterns & tips

• Separate concerns: workers run train.py; your laptop runs orchestrate.py. Never mix Hub API keys into worker scripts.
• Use CSV for large batches: export a spreadsheet as batch.csv and pass it to create_jobs("batch.csv").
• Inspect downloads: open jd.exp_path() in your file manager — each job id has its own folder under downloaded/.
• Combine with dashboard: you can still monitor workers and jobs in the web UI while your orchestrator runs in a terminal.
• Worker library reference: see Library Reference for jd_upload, checkpoints, and path helpers used inside entry scripts.