Function from_delayed

dask/bag/core.py:1888–1937

Create bag from many dask Delayed objects. These objects will become the partitions of the resulting Bag. They should evaluate to a ``list`` or some other concrete sequence. Parameters ---------- values: list of delayed values An iterable of dask Delayed objects. Each

(values)

def from_delayed(values):
    """Create bag from many dask Delayed objects.

    These objects will become the partitions of the resulting Bag.  They should
    evaluate to a ``list`` or some other concrete sequence.

    Parameters
    ----------
    values: list of delayed values
        An iterable of dask Delayed objects.  Each evaluating to a list.

    Returns
    -------
    Bag

    Examples
    --------
    >>> x, y, z = [delayed(load_sequence_from_file)(fn)
    ...             for fn in filenames] # doctest: +SKIP
    >>> b = from_delayed([x, y, z])  # doctest: +SKIP

    See also
    --------
    dask.delayed
    """
    from dask.delayed import Delayed, delayed

    if isinstance(values, Delayed):
        values = [values]
    futures = [v for v in values if isinstance(v, TaskRef)]
    if all_futures := (len(futures) == len(values)):
        # All futures. Fast path
        values = futures
    else:
        # Every Delayed generates a Layer, i.e. this path is much more expensive
        # if there are many input values.
        values = [
            delayed(v) if not isinstance(v, (Delayed,)) and hasattr(v, "key") else v
            for v in values
        ]

    token = tokenize(*values)
    name = f"bag-from-delayed-{token}"
    tasks = [Task((name, i), reify, TaskRef(v.key)) for i, v in enumerate(values)]
    dsk = {t.key: t for t in tasks}

    graph = HighLevelGraph.from_collections(
        name, dsk, dependencies=values if not all_futures else ()
    )
    return Bag(graph, name, len(values))


def chunk_distinct(seq, key=None):