Function read_datasource

python/ray/data/read_api.py:593–720

Read a stream from a custom :class:`~ray.data.Datasource`. Args: datasource: The :class:`~ray.data.Datasource` to read data from. parallelism: This argument is deprecated. Use ``override_num_blocks`` argument. num_cpus: The number of CPUs to reserve for each parallel rea

(
    datasource: Datasource,
    *,
    parallelism: int = -1,
    num_cpus: Optional[float] = None,
    num_gpus: Optional[float] = None,
    memory: Optional[float] = None,
    ray_remote_args: Optional[Dict[str, Any]] = None,
    concurrency: Optional[int] = None,
    compute: Optional[ComputeStrategy] = None,
    override_num_blocks: Optional[int] = None,
    **read_args,
)

@PublicAPI
@wrap_auto_init
def read_datasource(
    datasource: Datasource,
    *,
    parallelism: int = -1,
    num_cpus: Optional[float] = None,
    num_gpus: Optional[float] = None,
    memory: Optional[float] = None,
    ray_remote_args: Optional[Dict[str, Any]] = None,
    concurrency: Optional[int] = None,
    compute: Optional[ComputeStrategy] = None,
    override_num_blocks: Optional[int] = None,
    **read_args,
) -> Dataset:
    """Read a stream from a custom :class:`~ray.data.Datasource`.

    Args:
        datasource: The :class:`~ray.data.Datasource` to read data from.
        parallelism: This argument is deprecated. Use ``override_num_blocks`` argument.
        num_cpus: The number of CPUs to reserve for each parallel read worker.
        num_gpus: The number of GPUs to reserve for each parallel read worker. For
            example, specify `num_gpus=1` to request 1 GPU for each parallel read
            worker.
        memory: The heap memory in bytes to reserve for each parallel read worker.
        ray_remote_args: kwargs passed to :func:`ray.remote` in the read tasks.
        concurrency: The maximum number of Ray tasks to run concurrently. Set this
            to control number of tasks to run concurrently. This doesn't change the
            total number of tasks run or the total number of output blocks. By default,
            concurrency is dynamically decided based on the available resources.
        compute: The compute strategy to use for reading. Pass an
            :class:`~ray.data.ActorPoolStrategy` instance to use an actor pool,
            or a :class:`~ray.data.TaskPoolStrategy` instance (default) to use Ray tasks.
            If not specified, defaults to ``TaskPoolStrategy(concurrency)``. If both
            ``compute`` and ``concurrency`` are specified, ``concurrency`` takes precedence.
        override_num_blocks: Override the number of output blocks from all read tasks.
            By default, the number of output blocks is dynamically decided based on
            input data size and available resources. You shouldn't manually set this
            value in most cases.
        **read_args: Additional kwargs to pass to the :class:`~ray.data.Datasource`
            implementation.

    Returns:
        :class:`~ray.data.Dataset` that reads data from the :class:`~ray.data.Datasource`.

    Examples:
        Read using default task-based execution:

        >>> import ray
        >>> from ray.data._internal.datasource.range_datasource import RangeDatasource
        >>> datasource = RangeDatasource(n=1000, block_format="arrow")
        >>> ds = ray.data.read_datasource(datasource) # doctest: +SKIP

        Read using actors for stateful operations:

        >>> from ray.data import ActorPoolStrategy
        >>> ds = ray.data.read_datasource( # doctest: +SKIP
        ...     datasource,
        ...     compute=ActorPoolStrategy(size=4)  # Use 4 actors
        ... )