# Spans

Note

This is an experimental feature and may rapidly change without a deprecation cycle.

Dask offers various diagnostics and Fine Performance Metrics about tasks, grouping them by their prefix (commonly, the name of the function being called). This can be suboptimal:

- your client code may be sophisticated enough that correlating lines on the client and tasks being run on the cluster may prove difficult;

- the same function may be applied to different parts of your workflow, with a different performance profile;

- you may be in a multitenancy setup, where part of the load on the cluster was not caused by your client code.

In these situations, it may be useful to attach meaningful tags to your workflow, or segments of it. To do so, you should use the `span()` context manager inside the client code.

For example:

    import dask.config
    import dask.array as da
    from distributed import Client, span

    # Read important note below
    dask.config.set({"optimization.fuse.active": False})
    client = Client()

    with span("Alice's workflow"):
        with span("data load"):
            a = da.read_zarr(...)
        with span("ML preprocessing"):
            a = preprocess(a)
        with span("Model training"):
            model = train(a)
        model = model.compute()

Note how the `span()` context manager can be nested. The example will create the following spans on the scheduler:

- `("Alice's workflow", )`

- `("Alice's workflow", "data load")`

- `("Alice's workflow", "ML preprocessing")`

- `("Alice's workflow", "Model training")`

Each of the spans will be mapped to the tasks matching the segment of the graph that was defined inside its context manager. The parent span will be mapped to all tasks of its children.

Tags are arbitrary and nothing stops you from parameterizing them; for example

    >>> with span(f"{user}'s workflow"):
    ...     ...

Which may give you

- `("Alice's workflow", "data load")`

- `("Bob's workflow", "data load")`

- etc.

This is useful for example if you want to observe either all the workload submitted by Alice, while hiding Bob’s activity, or alternatively all the data loading activity, regardless of who submitted it.

The possibilities are more or less endless - instead or in addition to a username at the top, you could store information on what dataset you’re processing, etc.

## The default span

If you don’t use the `span()` context manager, your tasks will be automatically attributed to the `("default", )` span.

## Viewing the spans

You can filter by span tags in the Fine Performance Metrics dashboard widget to filter your workload:

Additionally, spans can be queried using scheduler extensions or `run_on_scheduler()`; see Dask Developer API.

## User API

Important

Dataframes have a minimum granularity of a single call to compute() or persist() and can’t break it down further into groups of operations - if the example above used dataframes, everything would have been uniformly tagged as “Alice’s Workflow”, as it is the span that’s active during compute().

In other collections, such as arrays and delayed objects, spans that don’t wrap around a call to compute() or persist() can get lost during the optimization phase. To prevent this issue, you must set

    >>> dask.config.set({"optimization.fuse.active": False})

Or in dask.yaml:

    optimization:
      fuse:
        active: false

A possible workaround, that also works for dataframes, can be to perform intermediate calls to persist(). Note however that this can significantly impact optimizations and reduce overall performance.

    with span("Alice's workflow"):
        with span("data load"):
            a = dd.read_parquet(...).persist()
        with span("ML preprocessing"):
            a = preprocess(a).persist()
            del a  # Release distributed memory for a as soon as possible
        with span("Model training"):
            model = train(b).persist()
            del b  # Release distributed memory for b as soon as possible
            model = model.compute()

distributed.span(*tags: str) → Iterator[str]

  Tag group of tasks to be part of a certain group, called a span.

  This context manager can be nested, thus creating sub-spans. If you close and re-open a span context manager with the same tag, you’ll end up with two separate spans.

  Every cluster defines a global “default” span when no span has been defined by the client; the default span is automatically closed and reopened when all tasks associated to it have been completed; in other words the cluster is idle save for tasks that are explicitly annotated by a span. Note that, in some edge cases, you may end up with overlapping default spans, e.g. if a worker crashes and all unique tasks that were in memory on it need to be recomputed.

  You may capture the ID of a span on the client to match it with the `Span` objects the scheduler:

      >>> client = Client()
      >>> with span("my workflow") as span_id:
      ...     client.submit(lambda: "Hello world!").result()
      >>> client.cluster.scheduler.extensions["spans"].spans[span_id]
      Span<name=('my workflow',), id=5dc9b908-116b-49a5-b0d7-5a681f49a111>

  Notes

  You may retrieve the current span with `dask.get_annotations().get("span")`. You can do so in the client code as well as from inside a task.

## Dask Developer API

Intended audience

This section is only of interest to developers maintaining Dask or writing scheduler extensions, e.g. to create an alternative dashboard or to store metrics long-term.

Spans can be accessed on the scheduler through `Scheduler.extensions["spans"]`, which contains a singleton instance of `SpansSchedulerExtension`. In turn, the extension contains a mapping of all `Span` objects, plus a variety of convenience methods to access and aggregate them.

Note how `Span` objects offer a variety of methods that the dashboard currently doesn’t use - such as start/stop times, tasks counts, and size of the output.

class distributed.spans.Span(name: tuple[str, ...], id_: str, parent: Span | None, total_nthreads_history: list[tuple[float, int]])
  property active_cpu_seconds: float

    Return number of CPU seconds that were made available on the cluster while this Span was running; in other words `(Span.stop - Span.enqueued) * Scheduler.total_nthreads`.

    This accounts for workers joining and leaving the cluster while this Span was active. If this Span is the output of `merge()`, do not count gaps between input spans.

    See also

    `enqueued`
    `stop`
    `nthreads_intervals`
    `distributed.scheduler.SchedulerState.total_nthreads`
  add_metadata(metadata: SpanMetadata) → None

    Add metadata to the span, e.g. code snippets

  property all_durations: dict[str, float]

    Cumulative duration of all completed actions in this span tree, by action

    See also

    `duration`
    `distributed.scheduler.TaskGroup.all_durations`
  property annotation: dict[str, tuple[str, ...]] | None

    Rebuild the dask graph annotation which contains the full id history

    Note that this may not match the original annotation in case of TaskGroup collision.

  children: list[Span]

    Direct children of this span, sorted by creation time

  property code: list[tuple[SourceCode, ...]]

    Code snippets, sent by the client on compute(), persist(), and submit().

    Only populated if `distributed.diagnostics.computations.nframes` is non-zero.

  property cumulative_worker_metrics: dict[tuple[Hashable, ...], float]

    Replica of `Worker.digests_total` and `Scheduler.cumulative_worker_metrics`, but only for the metrics that can be attributed to the current span tree. The span id has been removed from the key.

    At the moment of writing, all keys are `("execute", <task prefix>, <activity>, <unit>)` or `("p2p", <where>, <activity>, <unit>)` but more may be added in the future with a different format; please test e.g. for `k[0] == "execute"`.

  property done: bool

    Return True if all tasks in this span tree are completed; False otherwise.

    See also

    `distributed.scheduler.TaskGroup.done`

    Notes

    This property may transition from True to False, e.g. when a new sub-span is added or when a worker that contained the only replica of a task in memory crashes and the task need to be recomputed.

  property duration: float

    The total amount of time spent on all tasks in this span tree

    See also

    `all_durations`
    `distributed.scheduler.TaskGroup.duration`
  enqueued: float

    Time when the span first appeared on the scheduler. The same property on parent spans is always less than or equal to this.

    See also

    `start`
    `stop`
  groups: set[TaskGroup]

    Notes

    TaskGroups are forgotten by the Scheduler when the last task is forgotten, but remain referenced here indefinitely. If a user calls compute() twice on the same collection, you’ll have more than one group with the same tg.name in this set! For the same reason, while the same TaskGroup object is guaranteed to be attached to exactly one Span, you may have different TaskGroups with the same key attached to different Spans.

  id: str

    Unique ID, generated by `span()` and taken from `TaskState.annotations["span"]["id"][-1]`. Matches `distributed.scheduler.TaskState.group.span_id` and `distributed.worker_state_machine.TaskState.span_id`.

  static merge(*items: Span) → Span

    Merge multiple spans into a synthetic one. The input spans must not be related with each other.

  name: tuple[str, ...]

    (<tag>, <tag>, …) Matches `TaskState.annotations["span"]["name"]`, both on the scheduler and the worker.

  property nbytes_total: int

    The total number of bytes that this span tree has produced

    See also

    `distributed.scheduler.TaskGroup.nbytes_total`
  property nthreads_intervals: list[tuple[float, float, int]]
    Returns:
      List of tuples:
      - begin timestamp
      - end timestamp
      - Scheduler.total_nthreads during this interval
      When the Span is the output of `merge()`, the intervals may not be
      contiguous.

    See also

    `enqueued`
    `stop`
    `active_cpu_seconds`
    `distributed.scheduler.SchedulerState.total_nthreads`
  property start: float

    Earliest time when a task belonging to this span tree started computing; 0 if no task has finished computing yet.

    See also

    `enqueued`
    `stop`
    `distributed.scheduler.TaskGroup.start`

    Notes

    This is not updated until at least one task has finished computing. It could move backwards as tasks complete.

  property states: dict[TaskStateState, int]

    The number of tasks currently in each state in this span tree; e.g. `{"memory": 10, "processing": 3, "released": 4, ...}`.

    See also

    `distributed.scheduler.TaskGroup.states`
  property stop: float

    When this span tree finished computing, or current timestamp if it didn’t finish yet.

    See also

    `enqueued`
    `start`
    `done`
    `distributed.scheduler.TaskGroup.stop`

    Notes

    This differs from `TaskGroup.stop` when there aren’t unfinished tasks; is also will never be zero.

  traverse_groups() → Iterator[TaskGroup]

    All TaskGroups belonging to this branch of span tree

  traverse_spans() → Iterator[Span]

    Top-down recursion of all spans belonging to this branch off span tree, including self

class distributed.spans.SpansSchedulerExtension(scheduler: Scheduler)

  Scheduler extension for spans support

  find_by_tags(*tags: str) → Iterator[Span]

    Yield all spans that contain any of the given tags. When a tag is shared both by a span and its (grand)children, only return the parent.

  heartbeat(ws: scheduler_module.WorkerState, data: dict[tuple[Hashable, ...], float]) → None

    Triggered by `SpansWorkerExtension.heartbeat()`.

    Populate `Span.cumulative_worker_metrics()` with data from the worker.

    See also

    `SpansWorkerExtension.heartbeat`
    `Span.cumulative_worker_metrics`
  merge_all() → Span

    Return a synthetic Span which is the sum of all spans

  merge_by_tags(*tags: str) → Span

    Return a synthetic Span which is the sum of all spans containing the given tags

  observe_tasks(tss: Iterable[scheduler_module.TaskState], code: tuple[SourceCode, ...], span_metadata: SpanMetadata) → dict[Key, dict]

    Acknowledge the existence of runnable tasks on the scheduler. These may either be new tasks, tasks that were previously unrunnable, or tasks that were already fed into this method already.

    Attach newly observed tasks to either the desired span or to (“default”, ). Update TaskGroup.span_id and wipe TaskState.annotations[“span”].

    Returns:
      Updated ‘span’ annotations: {key: {“name”: (…, …), “ids”: (…, …)}}
  root_spans: list[Span]

    Only the spans that don’t have any parents, sorted by creation time. This is a convenience helper structure to speed up searches.

  spans: dict[str, Span]

    All Span objects by id

  spans_search_by_name: defaultdict[tuple[str, ...], list[Span]]

    All spans, keyed by their full name and sorted by creation time. This is a convenience helper structure to speed up searches.

  spans_search_by_tag: defaultdict[str, list[Span]]

    All spans, keyed by the individual tags that make up their name and sorted by creation time. This is a convenience helper structure to speed up searches.

    See also

    `find_by_tags`
    `merge_by_tags`

class distributed.spans.SpansWorkerExtension(worker: Worker)

  Worker extension for spans support

  heartbeat() → dict[tuple[Hashable, ...], float]

    Apportion the metrics that do have a span to the Spans on the scheduler

    Returns:
      `{(context, span_id, prefix, activity, unit): value}}`

    See also

    `SpansSchedulerExtension.heartbeat`
    `Span.cumulative_worker_metrics`
    `distributed.worker.Worker.get_metrics`