apache/beam · warning · NotImplementedError

collecting metrics will come later!

Error message

collecting metrics will come later!

What it means

DaskRunner.metrics() is a stub — collecting pipeline metrics from a Dask cluster is not implemented yet, so it raises NotImplementedError. The API exists to satisfy the runner interface but has no functionality.

Solutions

1. Don't call metrics() with DaskRunner; obtain metrics via Dask's distributed dashboard/client diagnostics instead.
2. Use a runner with metrics support (DirectRunner, DataflowRunner) if metrics are required.
3. Implement/contribute metrics collection using dask.distributed client scheduler_info().

Example fix

// before
result = dask_runner.metrics()
// after
client = dask.distributed.Client(...)
metrics = client.run_on_scheduler(lambda dask_scheduler: dask_scheduler.total_occupancy)

from apache_beam.runners.dask.dask_runner import DaskRunner
if isinstance(runner, DaskRunner):
    logging.warning('DaskRunner does not support metrics()')

try:
    metrics = runner.metrics()
except NotImplementedError:
    metrics = None  # use dask client diagnostics instead

When it happens

Trigger: Calling `DaskRunner.metrics()` (or the pipeline result path that invokes it) after running/cancelling a pipeline on the Dask runner.

Common situations: Developers switching a pipeline from DirectRunner/Dataflow to DaskRunner and expecting metrics collection to keep working.

Related errors

• interactive support will come later!
• Assigning an index is not yet supported. Consider using…
• by
• concat(ignore_index)
• concat(levels)

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/cec024f57b925781. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/runners/dask/dask_runner.py:163

        # not actually need to use the results here, so we just pass. to gather,
        # we use the iterative `as_completed(..., with_results=True)`, instead
        # of aggregate `client.gather`, to minimize memory footprint of results.
        pass
      self._state = PipelineState.DONE
    except:  # pylint: disable=broad-except
      self._state = PipelineState.FAILED
      raise
    return self._state

  def cancel(self) -> str:
    self._state = PipelineState.CANCELLING
    self.client.cancel(self.futures)
    self._state = PipelineState.CANCELLED
    return self._state

  def metrics(self):
    # TODO(alxr): Collect and return metrics...
    raise NotImplementedError('collecting metrics will come later!')


class DaskRunner(BundleBasedDirectRunner):
  """Executes a pipeline on a Dask distributed client."""
  @staticmethod
  def to_dask_bag_visitor(bag_kwargs=None) -> PipelineVisitor:
    from dask import bag as db

    if bag_kwargs is None:
      bag_kwargs = {}

    @dataclasses.dataclass
    class DaskBagVisitor(PipelineVisitor):
      bags: dict[AppliedPTransform, db.Bag] = dataclasses.field(
          default_factory=collections.OrderedDict)

      def visit_transform(self, transform_node: AppliedPTransform) -> None:
        op_class = TRANSLATIONS.get(transform_node.transform.__class__, NoOp)

View on GitHub (pinned to 12126d8942)