python/ray/data/dataset.py
def union(self, *other: List["Dataset[T]"]) -> "Dataset[T]":
        """Combine this dataset with others of the same type.

        The order of the blocks in the datasets is preserved, as is the
        relative ordering between the datasets passed in the argument list.

        Args:
            other: List of datasets to combine with this one. The datasets
                must have the same schema as this dataset, otherwise the
                behavior is undefined.

        Returns:
            A new dataset holding the union of their data.
        """

        start_time = time.perf_counter()
        context = DatasetContext.get_current()
        tasks: List[ReadTask] = []
        block_partition_refs: List[ObjectRef[BlockPartition]] = []
        block_partition_meta_refs: List[ObjectRef[BlockPartitionMetadata]] = []

        datasets = [self] + list(other)
        for ds in datasets:
            bl = ds._plan.execute()
            if isinstance(bl, LazyBlockList):
                tasks.extend(bl._tasks)
                block_partition_refs.extend(bl._block_partition_refs)
                block_partition_meta_refs.extend(bl._block_partition_meta_refs)
            else:
                tasks.extend([ReadTask(lambda: None, meta) for meta in bl._metadata])
                if context.block_splitting_enabled:
                    block_partition_refs.extend(
                        [ray.put([(b, m)]) for b, m in bl.get_blocks_with_metadata()]
                    )
                else:
                    block_partition_refs.extend(bl.get_blocks())
                block_partition_meta_refs.extend(
                    [ray.put(meta) for meta in bl._metadata]
                )

        epochs = [ds._get_epoch() for ds in datasets]
        max_epoch = max(*epochs)
        if len(set(epochs)) > 1:
            global _epoch_warned
            if not _epoch_warned:
                logger.warning(
                    "Dataset contains data from multiple epochs: {}, "
                    "likely due to a `rewindow()` call. The higher epoch "
                    "number {} will be used. This warning will not "
                    "be shown again.".format(set(epochs), max_epoch)
                )
                _epoch_warned = True
        dataset_stats = DatasetStats(
            stages={"union": []},
            parent=[d._plan.stats() for d in datasets],
        )
        dataset_stats.time_total_s = time.perf_counter() - start_time
        return Dataset(
            ExecutionPlan(
                LazyBlockList(tasks, block_partition_refs, block_partition_meta_refs),
                dataset_stats,
            ),
            max_epoch,
            self._lazy,
        )
Combine this dataset with others of the same type.

        The order of the blocks in the datasets is preserved, as is the
        relative ordering between the datasets passed in the argument list.

        Args:
            other: List of datasets to combine with this one. The datasets
                must have the same schema as this dataset, otherwise the
                behavior is undefined.

        Returns:
            A new dataset holding the union of their data.
        
Combine this dataset with others of the same type.
The order of the blocks in the datasets is preserved, as is the
relative ordering between the datasets passed in the argument list.
Combine this dataset with others of the same type . The order of the blocks in the datasets is preserved as is the relative ordering between the datasets passed in the argument list .
def union(self, *other: List["Dataset[T]"]) -> "Dataset[T]":
        start_time = time.perf_counter()
        context = DatasetContext.get_current()
        tasks: List[ReadTask] = []
        block_partition_refs: List[ObjectRef[BlockPartition]] = []
        block_partition_meta_refs: List[ObjectRef[BlockPartitionMetadata]] = []
        datasets = [self] + list(other)
        for ds in datasets:
            bl = ds._plan.execute()
            if isinstance(bl, LazyBlockList):
                tasks.extend(bl._tasks)
                block_partition_refs.extend(bl._block_partition_refs)
                block_partition_meta_refs.extend(bl._block_partition_meta_refs)
            else:
                tasks.extend([ReadTask(lambda: None, meta) for meta in bl._metadata])
                if context.block_splitting_enabled:
                    block_partition_refs.extend(
                        [ray.put([(b, m)]) for b, m in bl.get_blocks_with_metadata()]
                    )
                else:
                    block_partition_refs.extend(bl.get_blocks())
                block_partition_meta_refs.extend(
                    [ray.put(meta) for meta in bl._metadata]
                )
        epochs = [ds._get_epoch() for ds in datasets]
        max_epoch = max(*epochs)
        if len(set(epochs)) > 1:
            global _epoch_warned
            if not _epoch_warned:
                logger.warning(
                    "Dataset contains data from multiple epochs: {}, "
                    "likely due to a `rewindow()` call. The higher epoch "
                    "number {} will be used. This warning will not "
                    "be shown again.".format(set(epochs), max_epoch)
                )
                _epoch_warned = True
        dataset_stats = DatasetStats(
            stages={"union": []},
            parent=[d._plan.stats() for d in datasets],
        )
        dataset_stats.time_total_s = time.perf_counter() - start_time
        return Dataset(
            ExecutionPlan(
                LazyBlockList(tasks, block_partition_refs, block_partition_meta_refs),
                dataset_stats,
            ),
            max_epoch,
            self._lazy,
        )
def union ( self , * other : List [ "Dataset[T]" ] ) -> "Dataset[T]" : start_time = time . perf_counter ( ) context = DatasetContext . get_current ( ) tasks : List [ ReadTask ] = [ ] block_partition_refs : List [ ObjectRef [ BlockPartition ] ] = [ ] block_partition_meta_refs : List [ ObjectRef [ BlockPartitionMetadata ] ] = [ ] datasets = [ self ] + list ( other ) for ds in datasets : bl = ds . _plan . execute ( ) if isinstance ( bl , LazyBlockList ) : tasks . extend ( bl . _tasks ) block_partition_refs . extend ( bl . _block_partition_refs ) block_partition_meta_refs . extend ( bl . _block_partition_meta_refs ) else : tasks . extend ( [ ReadTask ( lambda : None , meta ) for meta in bl . _metadata ] ) if context . block_splitting_enabled : block_partition_refs . extend ( [ ray . put ( [ ( b , m ) ] ) for b , m in bl . get_blocks_with_metadata ( ) ] ) else : block_partition_refs . extend ( bl . get_blocks ( ) ) block_partition_meta_refs . extend ( [ ray . put ( meta ) for meta in bl . _metadata ] ) epochs = [ ds . _get_epoch ( ) for ds in datasets ] max_epoch = max ( * epochs ) if len ( set ( epochs ) ) > 1 : global _epoch_warned if not _epoch_warned : logger . warning ( "Dataset contains data from multiple epochs: {}, " "likely due to a `rewindow()` call. The higher epoch " "number {} will be used. This warning will not " "be shown again." . format ( set ( epochs ) , max_epoch ) ) _epoch_warned = True dataset_stats = DatasetStats ( stages = { "union" : [ ] } , parent = [ d . _plan . stats ( ) for d in datasets ] , ) dataset_stats . time_total_s = time . perf_counter ( ) - start_time return Dataset ( ExecutionPlan ( LazyBlockList ( tasks , block_partition_refs , block_partition_meta_refs ) , dataset_stats , ) , max_epoch , self . _lazy , )
Combine this dataset with others of the same type.
Combine this dataset with others of the same type .
"""Combine this dataset with others of the same type.

        The order of the blocks in the datasets is preserved, as is the
        relative ordering between the datasets passed in the argument list.

        Args:
            other: List of datasets to combine with this one. The datasets
                must have the same schema as this dataset, otherwise the
                behavior is undefined.

        Returns:
            A new dataset holding the union of their data.
        """
A new dataset holding the union of their data.
A new dataset holding the union of their data .
self
other
List["Dataset[T]"]
List of datasets to combine with this one. The datasets
must have the same schema as this dataset, otherwise the
behavior is undefined.
List of datasets to combine with this one . The datasets must have the same schema as this dataset otherwise the behavior is undefined .