Class HLGExpr

dask/_expr.py:969–1077

class HLGExpr(Expr):
    _parameters = [
        "dsk",
        "low_level_optimizer",
        "output_keys",
        "postcompute",
        "_cached_optimized",
    ]
    _defaults = {
        "low_level_optimizer": None,
        "output_keys": None,
        "postcompute": None,
        "_cached_optimized": None,
    }

    @property
    def hlg(self):
        return self.operand("dsk")

    @staticmethod
    def from_collection(collection, optimize_graph=True):
        from dask.highlevelgraph import HighLevelGraph

        if hasattr(collection, "dask"):
            dsk = collection.dask.copy()
        else:
            dsk = collection.__dask_graph__()

        # Delayed objects still ship with low level graphs as `dask` when going
        # through optimize / persist
        if not isinstance(dsk, HighLevelGraph):

            dsk = HighLevelGraph.from_collections(
                str(id(collection)), dsk, dependencies=()
            )
        if optimize_graph and not hasattr(collection, "__dask_optimize__"):
            warnings.warn(
                f"Collection {type(collection)} does not define a "
                "`__dask_optimize__` method. In the future this will raise. "
                "If no optimization is desired, please set this to `None`.",
                PendingDeprecationWarning,
            )
            low_level_optimizer = None
        else:
            low_level_optimizer = (
                collection.__dask_optimize__ if optimize_graph else None
            )
        return HLGExpr(
            dsk=dsk,
            low_level_optimizer=low_level_optimizer,
            output_keys=collection.__dask_keys__(),
            postcompute=collection.__dask_postcompute__(),
        )

    def finalize_compute(self):
        return HLGFinalizeCompute(
            self,
            low_level_optimizer=self.low_level_optimizer,