{"record":{"id":"cec024f57b925781","repo":"apache/beam","slug":"collecting-metrics-will-come-later","errorCode":null,"errorMessage":"collecting metrics will come later!","messagePattern":"collecting metrics will come later!","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"warning","filePath":"sdks/python/apache_beam/runners/dask/dask_runner.py","lineNumber":163,"sourceCode":"        # not actually need to use the results here, so we just pass. to gather,\n        # we use the iterative `as_completed(..., with_results=True)`, instead\n        # of aggregate `client.gather`, to minimize memory footprint of results.\n        pass\n      self._state = PipelineState.DONE\n    except:  # pylint: disable=broad-except\n      self._state = PipelineState.FAILED\n      raise\n    return self._state\n\n  def cancel(self) -> str:\n    self._state = PipelineState.CANCELLING\n    self.client.cancel(self.futures)\n    self._state = PipelineState.CANCELLED\n    return self._state\n\n  def metrics(self):\n    # TODO(alxr): Collect and return metrics...\n    raise NotImplementedError('collecting metrics will come later!')\n\n\nclass DaskRunner(BundleBasedDirectRunner):\n  \"\"\"Executes a pipeline on a Dask distributed client.\"\"\"\n  @staticmethod\n  def to_dask_bag_visitor(bag_kwargs=None) -> PipelineVisitor:\n    from dask import bag as db\n\n    if bag_kwargs is None:\n      bag_kwargs = {}\n\n    @dataclasses.dataclass\n    class DaskBagVisitor(PipelineVisitor):\n      bags: dict[AppliedPTransform, db.Bag] = dataclasses.field(\n          default_factory=collections.OrderedDict)\n\n      def visit_transform(self, transform_node: AppliedPTransform) -> None:\n        op_class = TRANSLATIONS.get(transform_node.transform.__class__, NoOp)","sourceCodeStart":145,"sourceCodeEnd":181,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/dask/dask_runner.py#L145-L181","documentation":"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.","triggerScenarios":"Calling `DaskRunner.metrics()` (or the pipeline result path that invokes it) after running/cancelling a pipeline on the Dask runner.","commonSituations":"Developers switching a pipeline from DirectRunner/Dataflow to DaskRunner and expecting metrics collection to keep working.","solutions":["Don't call metrics() with DaskRunner; obtain metrics via Dask's distributed dashboard/client diagnostics instead.","Use a runner with metrics support (DirectRunner, DataflowRunner) if metrics are required.","Implement/contribute metrics collection using dask.distributed client scheduler_info()."],"exampleFix":"// before\nresult = dask_runner.metrics()\n// after\nclient = dask.distributed.Client(...)\nmetrics = client.run_on_scheduler(lambda dask_scheduler: dask_scheduler.total_occupancy)","handlingStrategy":"fallback","validationCode":"from apache_beam.runners.dask.dask_runner import DaskRunner\nif isinstance(runner, DaskRunner):\n    logging.warning('DaskRunner does not support metrics()')","typeGuard":null,"tryCatchPattern":"try:\n    metrics = runner.metrics()\nexcept NotImplementedError:\n    metrics = None  # use dask client diagnostics instead","preventionTips":["Check runner feature support before relying on metrics.","Use the Dask dashboard/client scheduler info for Dask-runner metrics.","Pin expectations in code review when switching runners."],"tags":["python","apache-beam","dask","not-implemented"],"backgroundTag":"method-not-implemented","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}