diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index dc46000..d150723 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -5,11 +5,17 @@ in hours, so nothing here waits for one. The way to follow a run is to poll it or to listen on the flow socket, which carries its start and finish. """ +import csv +import io +import json +from collections.abc import Iterator from datetime import UTC, datetime +from itertools import groupby from typing import Any, Literal from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.concurrency import run_in_threadpool +from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field, model_validator from sqlalchemy import func from sqlalchemy import select as sa_select @@ -35,6 +41,11 @@ def elapsed_ms(since: datetime) -> float: return round((datetime.now(UTC) - start).total_seconds() * 1000, 2) +def _aware(when: datetime) -> datetime: + """A bound as the columns store it. A naive one is read as UTC.""" + return when if when.tzinfo else when.replace(tzinfo=UTC) + + #: Where a caller may say a run came from. "sweep" is not here because the #: sweep route writes it itself, and neither is a value a client made up: the #: column is only worth a table row if it means the same thing every time. @@ -252,10 +263,19 @@ def read_runs( status: str | None = None, group: str | None = None, digest: str | None = None, + since: datetime | None = None, + before: datetime | None = None, limit: int = 50, offset: int = 0, ) -> Any: - """Runs, newest first. The queryable table an experiment log needs.""" + """Runs, newest first. The queryable table an experiment log needs. + + ``before`` is the cursor a long history is paged by: rows are newest + first, so handing back the last row's ``created_at`` reads the next page + whatever landed meanwhile — which ``offset`` cannot, since a run submitted + between two pages shifts every row down one. ``since`` bounds the other + end and is inclusive. + """ statement = select(Run).order_by(col(Run.created_at).desc()) if flow: statement = statement.where(col(Run.flow) == flow) @@ -265,6 +285,10 @@ def read_runs( statement = statement.where(col(Run.group_id) == group) if digest: statement = statement.where(col(Run.params_digest) == digest) + if since: + statement = statement.where(col(Run.created_at) >= _aware(since)) + if before: + statement = statement.where(col(Run.created_at) < _aware(before)) statement = statement.offset(max(0, offset)).limit(min(limit, 500)) return list(session.exec(statement)) @@ -296,6 +320,232 @@ def read_overview(session: SessionDep) -> Any: return sorted(rows.values(), key=lambda row: row.last_created_at, reverse=True) +# --------------------------------------------------------------------------- +# Export +# +# Both routes are declared above `/{run_id}`, or "export" is read as the id of +# a run nobody has. What they are for: an analysis wants a dataframe, and the +# alternatives are a call per run or somebody reading our schema out of +# `fluksio.db`. The two shapes below are what an analysis actually asks for. +# --------------------------------------------------------------------------- + +#: What an export is written as. Parquet is a conversion the client does over +#: jsonl, because keeping dtypes is worth a dependency only to whoever wants it. +ExportFormat = Literal["csv", "jsonl"] + +#: The columns of the long table, in order. The run is on every row: it is the +#: join back to the run page and to what the run made, and it is what makes an +#: exported file auditable rather than loose. +METRIC_COLUMNS = ("run", "name", "step", "ts", "value") + +#: A run's own columns in the wide table. Its inputs and its final numbers +#: follow, prefixed, so an input named "status" cannot collide with the run's. +RUN_COLUMNS = ( + "id", + "flow", + "status", + "created_at", + "started_at", + "finished_at", + "duration_ms", + "seed", + "group_id", + "code_digest", + "origin_commit", +) + + +def _selected( + session: Session, + flow: str | None, + status: str | None, + group: str | None, + ids: str, + since: datetime | None, + until: datetime | None, +) -> list[Run]: + """The runs an export covers, newest first — the filters the list takes.""" + statement = select(Run).order_by(col(Run.created_at).desc()) + if flow: + statement = statement.where(col(Run.flow) == flow) + if status: + statement = statement.where(col(Run.status) == status) + if group: + statement = statement.where(col(Run.group_id) == group) + named = [part for part in ids.split(",") if part] + if named: + statement = statement.where(col(Run.id).in_(named)) + if since: + statement = statement.where(col(Run.created_at) >= _aware(since)) + if until: + statement = statement.where(col(Run.created_at) < _aware(until)) + return list(session.exec(statement)) + + +def _cell(value: Any) -> Any: + """A value as a table holds it: a scalar, or JSON when it is not one.""" + if hasattr(value, "isoformat"): + return value.isoformat() + if value is None or isinstance(value, (str, int, float, bool)): + return value + return json.dumps(value) + + +def _varying(runs: list[Run]) -> list[str]: + """The inputs that differ across these runs — the axis of a sweep. + + What a reader comparing arms wants as columns. Under two runs nothing can + differ, and a table of one run with none of its inputs in it is not worth + reading, so all of them are kept. + """ + keys = sorted({key for run in runs for key in run.params}) + if len(runs) < 2: + return keys + return [ + key + for key in keys + if len({json.dumps(run.params.get(key), sort_keys=True) for run in runs}) > 1 + ] + + +def _scored(runs: list[Run]) -> list[str]: + """A run's final numbers: every scalar its declared outputs carry.""" + return sorted( + { + key + for run in runs + for key, value in run.result.items() + if isinstance(value, (int, float)) and not isinstance(value, bool) + } + ) + + +def _drain(buffer: io.StringIO) -> str: + text = buffer.getvalue() + buffer.seek(0) + buffer.truncate(0) + return text + + +def _csv(columns: list[str], chunks: Iterator[list[dict[str, Any]]]) -> Iterator[str]: + """Header first, then a chunk at a time through one reused buffer.""" + buffer = io.StringIO() + writer = csv.DictWriter(buffer, fieldnames=columns) + writer.writeheader() + yield _drain(buffer) + for chunk in chunks: + writer.writerows(chunk) + yield _drain(buffer) + + +def _stream( + fmt: str, name: str, columns: list[str], chunks: Iterator[list[dict[str, Any]]] +) -> StreamingResponse: + """The rows out, a chunk at a time rather than a list. + + Streaming because the point of an export is that it is bigger than what a + screen reads: a sweep's curves are millions of rows. + """ + if fmt == "jsonl": + body: Iterator[str] = ( + "".join(json.dumps(row) + "\n" for row in chunk) for chunk in chunks + ) + media = "application/x-ndjson" + else: + body = _csv(columns, chunks) + media = "text/csv; charset=utf-8" + return StreamingResponse( + body, + media_type=media, + headers={"Content-Disposition": f'attachment; filename="{name}.{fmt}"'}, + ) + + +@router.get("/export/metrics") +def export_metrics( + session: SessionDep, + flow: str | None = None, + status: str | None = None, + group: str | None = None, + ids: str = "", + since: datetime | None = None, + until: datetime | None = None, + name: str = "", + stride: int = 1, + format: ExportFormat = "csv", +) -> Any: + """Every selected run's series as one long table: run, name, step, ts, value. + + The tidy shape a plotting library takes without reshaping. ``name`` keeps + the metrics it lists; ``stride`` thins each curve — per series, so asking + for every tenth point of two metrics gives every tenth point of both. + """ + runs = _selected(session, flow, status, group, ids, since, until) + wanted = [part for part in name.split(",") if part] + step = max(1, stride) + + def chunks() -> Iterator[list[dict[str, Any]]]: + for run in runs: + rows: list[dict[str, Any]] = [] + for series, points in groupby( + _series(session, run.id), key=lambda row: row.name + ): + if wanted and series not in wanted: + continue + rows.extend( + { + "run": run.id, + "name": row.name, + "step": row.step, + "ts": row.ts, + "value": row.value, + } + for row in list(points)[::step] + ) + yield rows + + return _stream(format, "metrics", list(METRIC_COLUMNS), chunks()) + + +@router.get("/export/runs") +def export_runs( + session: SessionDep, + flow: str | None = None, + status: str | None = None, + group: str | None = None, + ids: str = "", + since: datetime | None = None, + until: datetime | None = None, + params: str = "", + metrics: str = "", + format: ExportFormat = "csv", +) -> Any: + """One row per run: what it was given, what it scored, what code it ran. + + The arm-comparison table. Inputs are columns rather than one JSON blob — + by default the ones that vary across the selection, which is the sweep + axis; ``params`` names them instead. ``metrics`` narrows the final numbers + to a few of a run's declared outputs. + """ + runs = _selected(session, flow, status, group, ids, since, until) + inputs = [part for part in params.split(",") if part] or _varying(runs) + scores = [part for part in metrics.split(",") if part] or _scored(runs) + columns = [ + *RUN_COLUMNS, + *(f"param.{key}" for key in inputs), + *(f"metric.{key}" for key in scores), + ] + + def chunks() -> Iterator[list[dict[str, Any]]]: + for run in runs: + row = {column: _cell(getattr(run, column)) for column in RUN_COLUMNS} + row.update((f"param.{k}", _cell(run.params.get(k))) for k in inputs) + row.update((f"metric.{k}", _cell(run.result.get(k))) for k in scores) + yield [row] + + return _stream(format, "runs", columns, chunks()) + + @router.get("/{run_id}", response_model=RunDetail) def read_run(run_id: str, session: SessionDep) -> Any: """One run in full: what it was asked, what each node did, what it made.""" diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 7b69f1c..c7ce659 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -8,6 +8,7 @@ does, imports it inside the branch that asked for it. from __future__ import annotations import argparse +import csv import getpass import importlib import itertools @@ -16,7 +17,7 @@ import pkgutil import sys import time from collections.abc import Iterator -from contextlib import contextmanager +from contextlib import contextmanager, nullcontext from pathlib import Path from typing import Any @@ -25,6 +26,7 @@ import httpx from fluksio.sdk import FLOWS, Flow, SyncError from fluksio.sdk.client import ( GLOBAL_DATA_DIR, + RETRIES, WAIT_TOLERANCE, ApiError, Client, @@ -216,13 +218,18 @@ def _engine_client() -> Iterator[Client]: @contextmanager -def _client_for(args: argparse.Namespace) -> Iterator[Client]: - """The engine this command talks to: one running somewhere, or this one.""" +def _client_for(args: argparse.Namespace, retries: int = RETRIES) -> Iterator[Client]: + """The engine this command talks to: one running somewhere, or this one. + + ``retries=0`` is what a read passes: an engine that is not there otherwise + takes the backoff — seconds — to say so, and nothing in a read is worth + waiting out a restart for. A command that submits keeps them. + """ if getattr(args, "local", False): with _engine_client() as client: yield client else: - yield Client(url=args.url, token=args.token) + yield Client(url=args.url, token=args.token, retries=retries) # --------------------------------------------------------------------------- @@ -708,7 +715,9 @@ def cmd_status(args: argparse.Namespace) -> int: console = Console() try: - with _client_for(args) as client: + # A watch outlives a blip and should ride one out; a single screen is + # a read, and says "not answering" at once. + with _client_for(args, retries=RETRIES if args.watch else 0) as client: if not args.watch: console.print(_status_screen(client)) return 0 @@ -751,7 +760,7 @@ def _stamp(row: dict[str, Any]) -> str: def cmd_runs(args: argparse.Namespace) -> int: try: - with _client_for(args) as client: + with _client_for(args, retries=0) as client: rows = client.runs(flow=args.flow, limit=args.limit) except (SyncError, ApiError) as exc: return _fail(str(exc)) @@ -771,7 +780,7 @@ def cmd_runs(args: argparse.Namespace) -> int: def cmd_flavors(args: argparse.Namespace) -> int: """The named sizes a node can ask for.""" try: - with _client_for(args) as client: + with _client_for(args, retries=0) as client: rows = client.flavors() except (SyncError, ApiError) as exc: return _fail(str(exc)) @@ -850,6 +859,91 @@ def cmd_sweep(args: argparse.Namespace) -> int: return _unreachable(exc, "The runs are still on the engine; `fluksio runs`.") +def _selection(args: argparse.Namespace) -> dict[str, Any]: + """The filters both exports share, minus the ones left empty.""" + return { + key: value + for key, value in ( + ("status", args.status), + ("group", args.group), + ("since", args.since), + ("until", args.until), + ) + if value + } + + +def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str) -> int: + """The exported rows, in the format asked for, to a file or to stdout. + + The engine settles the columns over the whole selection before it sends + anything, so every row carries the same keys in the same order and the + first row's are the header. + """ + if not rows: + print("fluksio: nothing matched, so nothing was written", file=sys.stderr) + return 0 + if fmt == "parquet": + try: + import pyarrow + import pyarrow.parquet + except ImportError: + return _fail( + "parquet needs pyarrow — `pip install 'fluksio[parquet]'`, or " + "export csv and convert it" + ) + pyarrow.parquet.write_table(pyarrow.Table.from_pylist(rows), out) + return 0 + with open(out, "w", newline="") if out else nullcontext(sys.stdout) as handle: + if fmt == "jsonl": + handle.writelines(json.dumps(row) + "\n" for row in rows) + else: + writer = csv.DictWriter(handle, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + return 0 + + +def cmd_export_metrics(args: argparse.Namespace) -> int: + """Every selected run's numbers as one long table.""" + if args.format == "parquet" and not args.out: + return _fail("--format parquet writes a file; name it with -o FILE") + try: + with _client_for(args, retries=0) as client: + rows = client.export_metrics( + flow=args.flow, + ids=args.run, + name=args.name, + stride=args.stride, + **_selection(args), + ) + except (SyncError, ApiError) as exc: + return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc) + return _write_rows(rows, args.format, args.out) + + +def cmd_export_runs(args: argparse.Namespace) -> int: + """One row per run: its inputs, its final numbers, what it ran.""" + if args.format == "parquet" and not args.out: + return _fail("--format parquet writes a file; name it with -o FILE") + try: + with _client_for(args, retries=0) as client: + rows = client.export_runs( + flow=args.flow, + ids=args.run, + params=args.params, + metrics=args.metrics, + **_selection(args), + ) + except (SyncError, ApiError) as exc: + return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc) + return _write_rows(rows, args.format, args.out) + + # --------------------------------------------------------------------------- # Wiring # --------------------------------------------------------------------------- @@ -1014,3 +1108,59 @@ def add_parsers(subparsers: Any) -> None: ) with_engine(parser, local=True) parser.set_defaults(func=cmd_sweep) + + parser = subparsers.add_parser( + "export", help="runs and their numbers as a table an analysis reads" + ) + exports = parser.add_subparsers(dest="table", required=True) + + def with_selection(sub: argparse.ArgumentParser) -> None: + sub.add_argument("--flow", default="", help="only runs of this flow") + sub.add_argument( + "--run", + action="append", + default=[], + metavar="ID", + help="only this run; repeat for several", + ) + sub.add_argument("--group", default="", help="only the runs of this sweep") + sub.add_argument("--status", default="", help="only runs that ended this way") + sub.add_argument( + "--since", default="", metavar="TS", help="only runs created at or after" + ) + sub.add_argument( + "--until", default="", metavar="TS", help="only runs created before" + ) + sub.add_argument("--format", default="csv", choices=("csv", "jsonl", "parquet")) + sub.add_argument( + "-o", "--out", default="", metavar="FILE", help="write here, not to stdout" + ) + with_engine(sub, local=True) + + sub = exports.add_parser( + "metrics", help="one row per run, metric and step — the tidy shape" + ) + with_selection(sub) + sub.add_argument("--name", default="", metavar="A,B", help="only these metrics") + sub.add_argument( + "--stride", + type=int, + default=1, + help="keep every Nth point of each curve", + ) + sub.set_defaults(func=cmd_export_metrics) + + sub = exports.add_parser( + "runs", help="one row per run: its inputs, its final numbers, its status" + ) + with_selection(sub) + sub.add_argument( + "--params", + default="", + metavar="A,B", + help="the inputs to put in columns (default: the ones that vary)", + ) + sub.add_argument( + "--metrics", default="", metavar="A,B", help="the final numbers to keep" + ) + sub.set_defaults(func=cmd_export_runs) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 973cd76..335212b 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -397,6 +397,75 @@ class Client: params={"ids": ",".join(ids), "metric": metric}, ) + def export_metrics( + self, + flow: str = "", + ids: Iterable[str] = (), + name: str = "", + stride: int = 1, + **filters: Any, + ) -> list[dict[str, Any]]: + """Runs' series as long rows: run, name, step, ts, value. + + ``pd.DataFrame(client.export_metrics(flow="train"))`` is every loss + curve of that flow with the run id on each row. ``name`` keeps the + metrics it lists, ``stride`` thins each curve, and the selection is + narrowed the way :meth:`runs` is — ``status=``, ``group=``, + ``since=``, ``until=``. + """ + query: dict[str, Any] = {"stride": stride} + if name: + query["name"] = name + return self._export("/runs/export/metrics", flow, ids, query, filters) + + def export_runs( + self, + flow: str = "", + ids: Iterable[str] = (), + params: str = "", + metrics: str = "", + **filters: Any, + ) -> list[dict[str, Any]]: + """One row per run: its inputs as columns, its final numbers, its code. + + The arm-comparison table. The inputs kept are the ones that vary + across the selection unless ``params`` names them, which is the axis a + sweep is read along. + """ + query: dict[str, Any] = {} + if params: + query["params"] = params + if metrics: + query["metrics"] = metrics + return self._export("/runs/export/runs", flow, ids, query, filters) + + def _export( + self, + path: str, + flow: str, + ids: Iterable[str], + query: dict[str, Any], + filters: dict[str, Any], + ) -> list[dict[str, Any]]: + """An export as parsed rows. + + jsonl on the wire rather than csv: it keeps the types the engine has, + and it is the format the CLI converts to parquet from. Read whole — + the streaming is the engine's side, and a notebook wants a list. + """ + if flow: + query["flow"] = flow + named = ",".join(ids) + if named: + query["ids"] = named + for key, value in filters.items(): + query[key] = value.isoformat() if hasattr(value, "isoformat") else value + query["format"] = "jsonl" + response = self._request("GET", path, idempotent=True, params=query) + if response.status_code >= 400: + raise ApiError(response.status_code, _detail(response)) + return [json.loads(line) for line in response.text.splitlines() if line] + def cancel(self, run_id: str) -> Any: # Cancelling a cancelled run is cancelled. return self._call("POST", f"/runs/{run_id}/cancel", idempotent=True) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 61811dd..3ea0651 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -57,6 +57,12 @@ dependencies = [ "rich>=13", ] +[project.optional-dependencies] +# `fluksio export --format parquet` converts the exported rows with it. An +# extra rather than a dependency: csv and jsonl need nothing, and pyarrow is +# tens of megabytes for whoever wants dtypes kept. +parquet = ["pyarrow>=17"] + [project.urls] Homepage = "https://fluksio.com" Documentation = "https://docs.fluksio.com" @@ -103,6 +109,12 @@ untyped_calls_exclude = ["influxdb_client"] module = ["influxdb_client"] implicit_reexport = true +# The parquet extra. Not installed here, and the import that uses it is inside +# the try/except that says so. +[[tool.mypy.overrides]] +module = ["pyarrow", "pyarrow.*"] +ignore_missing_imports = true + [tool.ruff] target-version = "py312" diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 1bfac14..d3e32f6 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -642,3 +642,146 @@ def test_a_tree_that_did_not_move_is_not_written_again(): run = session.get(Run, "stamp-2") assert service._restamp(run, "same") == "same" + + +# ----------------------------------------------------------------------------- +# Export +# +# Two tables an analysis reads: the long one a curve is plotted from, and the +# wide one arms are compared in. Both carry the run id on every row. +# ----------------------------------------------------------------------------- + + +@pytest.fixture +def exported(): + """Two runs of one flow, each with two curves and one input that varies.""" + made = datetime.now(UTC) + with Session(db_engine) as session: + for index, (lr, acc) in enumerate([(0.1, 0.5), (0.01, 0.25)]): + run_id = f"exp-{index}" + session.add( + Run( + id=run_id, + flow="export-study", + status="ok", + params={"lr": lr, "epochs": 10}, + result={"acc": acc, "note": "n/a"}, + created_at=made + timedelta(seconds=index), + ) + ) + for name in ("study.loss", "study.val"): + for step in range(4): + session.add(_metric(run_id, name, step, float(step), 100.0 + step)) + session.commit() + yield ["exp-0", "exp-1"] + with Session(db_engine) as session: + for run_id in ("exp-0", "exp-1"): + for row in session.exec( + select(RunMetric).where(col(RunMetric.run_id) == run_id) + ).all(): + session.delete(row) + run = session.get(Run, run_id) + if run is not None: + session.delete(run) + session.commit() + + +def _lines(answer) -> list[dict]: + return [json.loads(line) for line in answer.text.splitlines() if line] + + +def test_an_export_strides_each_series_and_names_its_run( + client, superuser_token_headers, exported +): + """Every second point of every curve — not every second row of all of them.""" + answer = client.get( + f"{settings.API_V1_STR}/runs/export/metrics", + params={"ids": ",".join(exported), "stride": 2, "format": "jsonl"}, + headers=superuser_token_headers, + ) + + assert answer.status_code == 200 + rows = _lines(answer) + assert {row["run"] for row in rows} == set(exported) + curves: dict[tuple[str, str], list[int]] = {} + for row in rows: + curves.setdefault((row["run"], row["name"]), []).append(row["step"]) + assert len(curves) == 4 + assert all(steps == [0, 2] for steps in curves.values()) + + +def test_an_exported_run_row_carries_the_inputs_that_vary( + client, superuser_token_headers, exported +): + """The sweep axis becomes columns; what every run shares stays out of them.""" + + def export(**extra): + answer = client.get( + f"{settings.API_V1_STR}/runs/export/runs", + params={"ids": ",".join(exported), **extra}, + headers=superuser_token_headers, + ) + assert answer.status_code == 200 + return answer + + rows = _lines(export(format="jsonl")) + assert [row["id"] for row in rows] == ["exp-1", "exp-0"] + assert {row["param.lr"] for row in rows} == {0.1, 0.01} + # `epochs` is the same on both runs, so it is not what they differ by; a + # string in the result is not one of the run's numbers. + assert "param.epochs" not in rows[0] + assert rows[0]["metric.acc"] == 0.25 + assert "metric.note" not in rows[0] + assert "param.epochs" in _lines(export(format="jsonl", params="epochs"))[0] + + # csv is the default, and the columns are in the order the header names. + header = export().text.splitlines()[0] + assert header.startswith("id,flow,status,") + assert header.endswith("param.lr,metric.acc") + + +@pytest.fixture +def paged(): + """Three runs of one flow, a minute apart.""" + made = datetime.now(UTC).replace(microsecond=0) + with Session(db_engine) as session: + for index in range(3): + session.add( + Run( + id=f"page-{index}", + flow="paged-study", + status="ok", + created_at=made + timedelta(minutes=index), + ) + ) + session.commit() + yield made + with Session(db_engine) as session: + for index in range(3): + run = session.get(Run, f"page-{index}") + if run is not None: + session.delete(run) + session.commit() + + +def test_runs_page_by_when_they_were_created(client, superuser_token_headers, paged): + """`before` is the cursor: the last row's own timestamp reads the next page.""" + + def listed(**params): + answer = client.get( + f"{settings.API_V1_STR}/runs", + params={"flow": "paged-study", **params}, + headers=superuser_token_headers, + ) + assert answer.status_code == 200 + return [row["id"] for row in answer.json()] + + assert listed() == ["page-2", "page-1", "page-0"] + assert listed(before=(paged + timedelta(minutes=2)).isoformat()) == [ + "page-1", + "page-0", + ] + assert listed(since=(paged + timedelta(minutes=1)).isoformat()) == [ + "page-2", + "page-1", + ] diff --git a/backend/tests/sdk/test_client.py b/backend/tests/sdk/test_client.py index 71b58d5..70c1ab9 100644 --- a/backend/tests/sdk/test_client.py +++ b/backend/tests/sdk/test_client.py @@ -130,3 +130,25 @@ def test_a_run_that_is_gone_stops_the_wait_at_once(monkeypatch): handle.wait(poll=0) assert caught.value.status == 404 assert len(calls) == 1, "a 404 is an answer, not a blip" + + +def test_an_export_is_read_as_rows(monkeypatch): + """jsonl on the wire: a stream of documents, not one.""" + seen = {} + + def handler(request): + seen.update(request.url.params) + return httpx.Response( + 200, + text='{"run": "r1", "step": 0}\n{"run": "r1", "step": 1}\n', + headers={"content-type": "application/x-ndjson"}, + ) + + client = a_client(handler, monkeypatch) + rows = client.export_metrics(flow="train", ids=["r1", "r2"], status="ok") + + assert seen["format"] == "jsonl" + assert seen["ids"] == "r1,r2" + assert seen["flow"] == "train" + assert seen["status"] == "ok" + assert [row["step"] for row in rows] == [0, 1] diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 45d56fa..aa7edbd 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -312,3 +312,36 @@ def test_the_run_prompt_names_an_answer_of_the_wrong_type() -> None: with patch("builtins.input", side_effect=["fast"]): with pytest.raises(SyncError, match="lr"): _ask_params(definition) + + +def test_an_export_is_parsed_with_its_selection() -> None: + """`export` is a group of two tables, and both take the same filters.""" + from fluksio.cli import _parser + from fluksio.sdk.cli import cmd_export_metrics, cmd_export_runs + + parser = _parser() + metrics = parser.parse_args( + [ + "export", + "metrics", + "--flow", + "train", + "--run", + "a", + "--run", + "b", + "--name", + "loss,val", + "--stride", + "5", + ] + ) + assert metrics.func is cmd_export_metrics + assert metrics.run == ["a", "b"] + assert metrics.stride == 5 + assert metrics.format == "csv" + + runs = parser.parse_args(["export", "runs", "--params", "lr", "--format", "jsonl"]) + assert runs.func is cmd_export_runs + assert runs.params == "lr" + assert runs.format == "jsonl" diff --git a/docs/code/api.md b/docs/code/api.md index 39d51f6..48b3161 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -104,8 +104,10 @@ published to. Flows own the namespace; everything else is a client of it. |---|---|---| | `POST` | `/runs/flows/{name}` | queue one run — `{"params": {...}, "seed": 7, "draft": false, "no_cache": false}`. `"cause"` says where it came from — `api` (the default), `cli` or `sdk` | | `POST` | `/runs/flows/{name}/sweep` | queue many, sharing a `group_id` | -| `GET` | `/runs` | the queryable history: `?flow=`, `?status=`, `?group=`, `?digest=`, `?limit=`, `?offset=` | +| `GET` | `/runs` | the queryable history: `?flow=`, `?status=`, `?group=`, `?digest=`, `?since=`, `?before=`, `?limit=`, `?offset=` | | `GET` | `/runs/overview` | one row per flow that has runs, with how many are running or queued | +| `GET` | `/runs/export/metrics?…&name=&stride=&format=` | every selected run's series as one long table: `run, name, step, ts, value` | +| `GET` | `/runs/export/runs?…¶ms=&metrics=&format=` | one row per run: its inputs as columns, its final numbers, its status and provenance | | `GET` | `/runs/{id}` | one run in full: params, result, per-node record, artifacts | | `POST` | `/runs/{id}/cancel` | stop it | | `GET` | `/runs/{id}/metrics?name=&stride=` | one metric's series, in step order; every series of the run without `name` | @@ -118,6 +120,18 @@ problem, before anything executes. `?digest=` filters by the hash of the parameters, which is how you find "every run that used exactly this configuration". +`?before=` is how a long history is paged: rows come newest first, so handing +back the last row's `created_at` reads the next page whatever landed +meanwhile — which `?offset=` cannot, since a run submitted between two pages +shifts every row down one. `?since=` is inclusive and bounds the other end. + +Both exports stream `csv` (the default) or `jsonl`, and take the selection the +history takes plus `?ids=a,b,c`, `?since=` and `?until=`. `export/runs` puts +the inputs that *vary* across the selection in `param.` columns — the sweep +axis — unless `?params=` names them, and the run's numbers in `metric.` +columns. The run id is on every row of both, which is what makes an exported +file a join back to the run rather than a loose table. + `compare` answers in the same shape a chart widget draws, so three training curves side by side is a widget binding rather than a screen of its own. diff --git a/docs/code/cli.md b/docs/code/cli.md index 01b7f9f..5700a17 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -304,6 +304,36 @@ flow's inputs, the same as `run`'s are, and `--seed`, `--no-sync`, `--no-cache` and `--local` mean what they do there. `--wait` blocks until all of them are finished and exits non-zero if any failed. +### `fluksio export` + +```sh +fluksio export metrics --flow train --name train_loss,val_loss --stride 10 -o curves.csv +fluksio export runs --flow train --status ok > arms.csv +``` + +The two tables an analysis reads. `export metrics` is the long one — a row per +run, metric and step — which is what a plotting library takes without +reshaping; `--name` keeps the metrics it lists and `--stride` keeps every Nth +point of *each* curve. `export runs` is the wide one: a row per run with its +inputs as columns, its final numbers, its status, its duration and the commit +and digest of the code it ran. + +The inputs that become columns are the ones that **vary** across the selected +runs — the axis of the sweep, which is what a comparison is read along — +unless `--params lr,seed` names them. `--metrics` narrows the final numbers +the same way. + +Both take `--flow`, `--run ID` (repeat it), `--group`, `--status`, `--since`, +`--until` and `--local`, and both put the run id on every row: it is the join +back to the run page and to what the run made. + +`--format` is `csv` (the default), `jsonl` or `parquet`; output goes to stdout +unless `-o FILE` names somewhere. Parquet keeps the types and needs pyarrow — +`pip install 'fluksio[parquet]'` — and a file to write, since it is not a +stream. In a notebook, `Client.export_metrics()` and `Client.export_runs()` +answer the same rows as a list of dicts, which `pandas.DataFrame` takes +directly. + ## What lives in the data directory ```text diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 0bb0dc2..d00289e 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -318,6 +318,28 @@ an epoch, or samples seen — joined on the step the two share. One run in full is params, the per-node record with its logs and traceback, the artifacts it made, its metrics and its result. +### Taking it into a dataframe + +An analysis wants a table rather than a screen, and there are two it usually +wants. `fluksio export metrics` is the long one — a row per run, metric and +step — and `fluksio export runs` is the wide one, a row per run with the +inputs that varied as columns beside its final numbers: + +```python +import pandas as pd +from fluksio.sdk.client import Client + +client = Client() +curves = pd.DataFrame(client.export_metrics(flow="train")) +arms = pd.DataFrame(client.export_runs(flow="train", status="ok")) +``` + +The run id is on every row of both, so a curve joins to the arm it came from +and to the run page it was recorded on, and the wide table carries the commit +and the code digest — an exported file says what produced its numbers. +[The CLI](../code/cli.md#fluksio-export) writes the same rows as csv, jsonl or +parquet, which is where an export belongs: in the script beside the analysis. + ### A dashboard, read against runs A run records values under the same names a dashboard binds to — a run of diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index 3bfb8f3..2a7cf88 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -605,6 +605,22 @@ curl -s "$FLUKSIO/runs/series/compare?ids=$A,$B,$C&metric=train.loss" \ which answers in exactly the shape a chart widget draws. +## Take it into a dataframe + +The analysis itself wants a table, and `export` writes the two an analysis +asks for — the curves long, and one row per run with the parameters that +varied beside its final numbers: + +```sh +fluksio export metrics --flow train --name train.loss -o curves.csv +fluksio export runs --flow train --status ok -o arms.csv +``` + +Both carry the run id on every row, and `arms.csv` carries the commit and the +code digest beside it, so an exported file still says what produced its +numbers. `Client.export_metrics()` and `Client.export_runs()` answer the same +rows to a notebook, ready for `pandas.DataFrame`. + ## Looking at them in the portal None of the above needs a browser, but there is one. **Runs** is the table of diff --git a/uv.lock b/uv.lock index 80ba2a0..3a17484 100644 --- a/uv.lock +++ b/uv.lock @@ -899,6 +899,11 @@ dependencies = [ { name = "uv" }, ] +[package.optional-dependencies] +parquet = [ + { name = "pyarrow" }, +] + [package.dev-dependencies] dev = [ { name = "coverage" }, @@ -925,6 +930,7 @@ requires-dist = [ { name = "numpy", specifier = ">=2.2.6" }, { name = "orjson", specifier = ">=3.10" }, { name = "pwdlib", extras = ["argon2", "bcrypt"], specifier = ">=0.3.0" }, + { name = "pyarrow", marker = "extra == 'parquet'", specifier = ">=17" }, { name = "pydantic", specifier = ">2.0" }, { name = "pydantic-settings", specifier = ">=2.2.1,<3.0.0" }, { name = "pyjwt", specifier = ">=2.8.0,<3.0.0" }, @@ -936,6 +942,7 @@ requires-dist = [ { name = "tenacity", specifier = ">=8.2.3,<9.0.0" }, { name = "uv", specifier = ">=0.5" }, ] +provides-extras = ["parquet"] [package.metadata.requires-dev] dev = [ @@ -1928,6 +1935,42 @@ bcrypt = [ { name = "bcrypt" }, ] +[[package]] +name = "pyarrow" +version = "25.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/3d/e3/27f57f80141379d60defe6703eb50a707325706f07fedfd1312c7a751995/pyarrow-25.0.1.tar.gz", hash = "sha256:9150a83248bfed9813ea3c3af74c3856c1984d444aa28e58bf7733b9750ddf6a", size = 1201653, upload-time = "2026-08-10T12:40:53.904Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a6/e2/9ab15b88cbfac28e16419ce5439ec29234c5172cb8259301b4ba639bdec0/pyarrow-25.0.1-cp312-cp312-macosx_12_0_arm64.whl", hash = "sha256:df961f2e7ae9cf496459259d798652c70625f6c080650d6952f8c04053c58ee9", size = 35861559, upload-time = "2026-08-10T12:38:02.567Z" }, + { url = "https://files.pythonhosted.org/packages/58/79/a0036dbe1eabe1f73127427342f1d99982584c4a2cde2651d6c93499c6f6/pyarrow-25.0.1-cp312-cp312-macosx_12_0_x86_64.whl", hash = "sha256:cc4aa407fde9fc660be3939e49ea31f50f3e9fec17c0ec63159f7711edd3efc9", size = 37628383, upload-time = "2026-08-10T12:38:09.083Z" }, + { url = "https://files.pythonhosted.org/packages/13/49/d93a57d375f4bf0cf82913dd6bb54acafde83dd993be2282c81ac5616cad/pyarrow-25.0.1-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:4340f0ba6c1d2e13f21658de1d7c662ca2545018568d0030a1e9afca159d87e3", size = 46820190, upload-time = "2026-08-10T12:38:15.458Z" }, + { url = "https://files.pythonhosted.org/packages/60/c9/711ca85d79f1ec98f29a5eae2b051e25b4ecec5de3e3c0e2d5c5dcb15664/pyarrow-25.0.1-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:5389cdf79447ed1515c9e31620e6e1e2302249564d603f2ad727d4f6d313e4c3", size = 50102437, upload-time = "2026-08-10T12:38:22.487Z" }, + { url = "https://files.pythonhosted.org/packages/80/53/8fb8359ff17cfb6263a1cf3ebf7caec9fe197de118719e84fcb1d0618026/pyarrow-25.0.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:d51592cb7561e87877c506113e7adbf1342ab579e6c21f0ef44b8ba41cb74c80", size = 49942424, upload-time = "2026-08-10T12:38:28.755Z" }, + { url = "https://files.pythonhosted.org/packages/e8/83/4e5ae02a9341571b18a6fca380ac7a58ce6ddae7ab3c060208c0a1e79f02/pyarrow-25.0.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:6109c94d8b9f3b17a041daca16cacb2f651ad8f1ef70a4232c2c0f37a23da2a8", size = 53144206, upload-time = "2026-08-10T12:38:34.862Z" }, + { url = "https://files.pythonhosted.org/packages/65/ee/197cbf47e49f83e6ebeb946a5259a48a638dea27ac774db42fe78022179d/pyarrow-25.0.1-cp312-cp312-win_amd64.whl", hash = "sha256:8858d7bfc22e3f51529aeaa4077225029724623e4595dc9eff8c793935c34140", size = 27953934, upload-time = "2026-08-10T12:38:39.808Z" }, + { url = "https://files.pythonhosted.org/packages/cc/8d/8f271a7a034c834910ec925d56fa4b29733b1380f5289419f5aaa3b02777/pyarrow-25.0.1-cp313-cp313-macosx_12_0_arm64.whl", hash = "sha256:c7c534ec03c358a76ea3e505e74c1b6aef290af90c444dfd092dbfe23e755b85", size = 35855328, upload-time = "2026-08-10T12:38:45.489Z" }, + { url = "https://files.pythonhosted.org/packages/d2/cd/5bac242f4e841b9971d5eb94fdfe2577e2b70be983e27401e72055786037/pyarrow-25.0.1-cp313-cp313-macosx_12_0_x86_64.whl", hash = "sha256:dda9470024204d7bbf2042b47c6e8a0e47a3eeb8e34405882dfaea6577e0c153", size = 37622415, upload-time = "2026-08-10T12:38:51.107Z" }, + { url = "https://files.pythonhosted.org/packages/63/1f/96d03b4e1506524f7087adb0fd6b2f69f0c9c7aaff1ec36d8030082e15a5/pyarrow-25.0.1-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:44a9120ce5bd81936b8ab9a88076e3fd47c2c6838e0e43630fed83626aca81d9", size = 46813813, upload-time = "2026-08-10T12:38:57.773Z" }, + { url = "https://files.pythonhosted.org/packages/98/d6/33a411115b61dbfc16ad6ad73e71730f6fea654ee3667673bc53ab0e2fe7/pyarrow-25.0.1-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:0befcf816e45a1af33ac775a9970b749e4868a230c7372f0ae5e932bee27039f", size = 50104452, upload-time = "2026-08-10T12:39:04.579Z" }, + { url = "https://files.pythonhosted.org/packages/33/ae/b1b97c9ca87f9f9ddbb5230c798df94eccce61bd79b9b45458c69a478588/pyarrow-25.0.1-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:3f89685964f46e4216103c75483aac0c0692a5f72212d7ca835adba5ede56ce3", size = 49951343, upload-time = "2026-08-10T12:39:11.8Z" }, + { url = "https://files.pythonhosted.org/packages/98/9e/a112df5cfd5a68cb1d9fc31cfe38c28d5aec9f10865ce37ecef2e4450873/pyarrow-25.0.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:6943e2fe7954d29d84de45d29d34c8dc36ce96570e67d89aa9976e650a4a9138", size = 53144784, upload-time = "2026-08-10T12:39:20.503Z" }, + { url = "https://files.pythonhosted.org/packages/31/24/97e8bd98f1e3b07e2ba08bcdff690674fbe16d69a7d2712cc3884665e615/pyarrow-25.0.1-cp313-cp313-win_amd64.whl", hash = "sha256:31e49a7888fcdf3a835da33ae777f6bb9a866334e5a789282fc26dcf426f7f15", size = 27870159, upload-time = "2026-08-10T12:39:26.161Z" }, + { url = "https://files.pythonhosted.org/packages/36/4c/b525824ad3094076919273cd97db61fb3d78252dee76fa3b8dc8f76774aa/pyarrow-25.0.1-cp314-cp314-macosx_12_0_arm64.whl", hash = "sha256:bf0b672390cdcb640d7288f96b826d71ff4e9abb254a86c89890baf51a29cee6", size = 35885255, upload-time = "2026-08-10T12:39:32.366Z" }, + { url = "https://files.pythonhosted.org/packages/08/62/448bb0e940de41aec31d1a956e63ad9c54afdf122a103cc3ab20c2a3ce33/pyarrow-25.0.1-cp314-cp314-macosx_12_0_x86_64.whl", hash = "sha256:38a9a4b4b9613380e200641891495a56c3d5a98a092db4a870af9975e220471d", size = 37644461, upload-time = "2026-08-10T12:39:38.142Z" }, + { url = "https://files.pythonhosted.org/packages/6e/9a/13587e38bd4806fd218f50fd13b8903fab60588a699ff0c406372e5b4043/pyarrow-25.0.1-cp314-cp314-manylinux_2_28_aarch64.whl", hash = "sha256:0b726ad7e7b669be982b0c71c07fe4b037d654354130da79a7902a669e93a66b", size = 46877146, upload-time = "2026-08-10T12:39:43.722Z" }, + { url = "https://files.pythonhosted.org/packages/8d/61/1c5d1229fa21da4cff5365e41e57177aaac57c563c727f35419b8513d1c1/pyarrow-25.0.1-cp314-cp314-manylinux_2_28_x86_64.whl", hash = "sha256:9171748cdf796972d85a4b60157c279913e242992e350c90c7450182a9838b2a", size = 50131616, upload-time = "2026-08-10T12:39:49.304Z" }, + { url = "https://files.pythonhosted.org/packages/43/20/291e1d65cc0b09aa19f03cf25cf51a2f5fa94b5db315178f2d254ed5cad4/pyarrow-25.0.1-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:b7a296aac7a71fa0886c08e155ddb6c636a50013f801f6178daafa0f9e726188", size = 50008879, upload-time = "2026-08-10T12:39:56.891Z" }, + { url = "https://files.pythonhosted.org/packages/8b/7c/1b7c9ec28e76576337e4f97b31141c9a181b89b6d1d6221e9d8205621a58/pyarrow-25.0.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:0fe7c8b6c03969b49c8c66182e4a18e3819ab92d07cfab5d8370c531b9369ef0", size = 53170864, upload-time = "2026-08-10T12:40:04.918Z" }, + { url = "https://files.pythonhosted.org/packages/b7/75/f3d789dc06011a765d14d86bda799cf72ac1d715b6a6edecaa0d73d95062/pyarrow-25.0.1-cp314-cp314-win_amd64.whl", hash = "sha256:f729cfdbd36fd99d543b67a914d2de044c84ebe45be8b34902b299b608c15c8f", size = 28620729, upload-time = "2026-08-10T12:40:51.41Z" }, + { url = "https://files.pythonhosted.org/packages/fc/05/647a8ee6f7c2662feb6921315617bc04dcd6034763fb61b1199720bf6162/pyarrow-25.0.1-cp314-cp314t-macosx_12_0_arm64.whl", hash = "sha256:59a2de54c0cbd954da861eee4d1d330f8e909c45b53455baef696380f2c55033", size = 36130288, upload-time = "2026-08-10T12:40:11.014Z" }, + { url = "https://files.pythonhosted.org/packages/93/f8/c9ee997554d7bea94520667dd1933f109ac1da3ee3556d2b49381e023484/pyarrow-25.0.1-cp314-cp314t-macosx_12_0_x86_64.whl", hash = "sha256:35935cd5de130aa5cf4dea052a63e6bf2e17006c35c3a468194242b9b2bf5956", size = 37762187, upload-time = "2026-08-10T12:40:16.592Z" }, + { url = "https://files.pythonhosted.org/packages/a2/08/a28c01c7fe9e96e8233ce2d13df1d402f4f999f848f51d2daacd6bb4c036/pyarrow-25.0.1-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:f3831aaa25c67a99f99dc8b05873cb9d64560390372e2aa197ce9dd4a3f06a44", size = 46888003, upload-time = "2026-08-10T12:40:23.242Z" }, + { url = "https://files.pythonhosted.org/packages/1b/b9/58612e977d28dc58c878448866838369ee8da2f1e7cc8ed2c84b952aafee/pyarrow-25.0.1-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:6a1fdfc6659b6b19022f2e50627fb5cf7156a66c46bf4299379955cbe742382a", size = 50079036, upload-time = "2026-08-10T12:40:29.169Z" }, + { url = "https://files.pythonhosted.org/packages/72/13/66e1402dcc860e1dc2760b1e0292c9a569b62b3bccab69def1b3e907d006/pyarrow-25.0.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:169d3429d5be7c752125890620f75a60776d38b0035eddae939651640822332e", size = 50040226, upload-time = "2026-08-10T12:40:35.186Z" }, + { url = "https://files.pythonhosted.org/packages/78/10/3f1a5497a7ef732ab0f03ecca3e66d89d9c0f57fdc61b4794c456b781f01/pyarrow-25.0.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:119297a6dc197e45d9c6d4415f7814a67ffa36c180d26f68c154c58067ae782d", size = 53149035, upload-time = "2026-08-10T12:40:41.454Z" }, + { url = "https://files.pythonhosted.org/packages/93/c0/37d4a7e8e2f7a6076283673d5298018ca26478b934c6ee369e10505ab32c/pyarrow-25.0.1-cp314-cp314t-win_amd64.whl", hash = "sha256:4288f27577352d608ca08553b0865e4a9b3aa14820c5d95b53337218d609835b", size = 28753071, upload-time = "2026-08-10T12:40:46.623Z" }, +] + [[package]] name = "pycparser" version = "3.0"