Method join

dask/bag/core.py:1151–1217

Joins collection with another collection. Other collection must be one of the following: 1. An iterable. We recommend tuples over lists for internal performance reasons. 2. A delayed object, pointing to a tuple. This is recommended if the other c

(self, other, on_self, on_other=None)

        return self.var(ddof=ddof).apply(math.sqrt)

    def join(self, other, on_self, on_other=None):
        """Joins collection with another collection.

        Other collection must be one of the following:

        1.  An iterable.  We recommend tuples over lists for internal
            performance reasons.
        2.  A delayed object, pointing to a tuple.  This is recommended if the
            other collection is sizable and you're using the distributed
            scheduler.  Dask is able to pass around data wrapped in delayed
            objects with greater sophistication.
        3.  A Bag with a single partition

        You might also consider Dask Dataframe, whose join operations are much
        more heavily optimized.

        Parameters
        ----------
        other: Iterable, Delayed, Bag
            Other collection on which to join
        on_self: callable
            Function to call on elements in this collection to determine a
            match
        on_other: callable (defaults to on_self)
            Function to call on elements in the other collection to determine a
            match

        Examples
        --------
        >>> import dask.bag as db
        >>> people = db.from_sequence(['Alice', 'Bob', 'Charlie'])
        >>> fruit = ['Apple', 'Apricot', 'Banana']
        >>> list(people.join(fruit, lambda x: x[0]))
        [('Apple', 'Alice'), ('Apricot', 'Alice'), ('Banana', 'Bob')]
        """
        token = tokenize(self, other, on_self, on_other)
        name = f"join-{token}"
        dsk = {}
        if isinstance(other, Bag):
            if other.npartitions == 1:
                dsk.update(other.dask)
                other = other.__dask_keys__()[0]
                dsk[f"join-{name}-other"] = (list, other)
            else:
                msg = (
                    "Multi-bag joins are not implemented. "
                    "We recommend Dask dataframe if appropriate"
                )
                raise NotImplementedError(msg)
        elif isinstance(other, Delayed):
            dsk.update(other.dask)
            other = other._key
        elif not isinstance(other, Iterable):
            msg = (
                "Joined argument must be single-partition Bag, "
                f" delayed object, or Iterable, got {type(other).__name}"
            )
            raise TypeError(msg)