python/ray/data/read_api.py
def read_datasource(
    datasource: Datasource[T],
    *,
    parallelism: int = 200,
    ray_remote_args: Dict[str, Any] = None,
    **read_args,
) -> Dataset[T]:
    """Read a dataset from a custom data source.

    Args:
        datasource: The datasource to read data from.
        parallelism: The requested parallelism of the read. Parallelism may be
            limited by the available partitioning of the datasource.
        read_args: Additional kwargs to pass to the datasource impl.
        ray_remote_args: kwargs passed to ray.remote in the read tasks.

    Returns:
        Dataset holding the data read from the datasource.
    """
    # TODO(ekl) remove this feature flag.
    force_local = "RAY_DATASET_FORCE_LOCAL_METADATA" in os.environ
    pa_ds = _lazy_import_pyarrow_dataset()
    if pa_ds:
        partitioning = read_args.get("dataset_kwargs", {}).get("partitioning", None)
        if isinstance(partitioning, pa_ds.Partitioning):
            logger.info(
                "Forcing local metadata resolution since the provided partitioning "
                f"{partitioning} is not serializable."
            )
            force_local = True

    if force_local:
        read_tasks = datasource.prepare_read(parallelism, **read_args)
    else:
        # Prepare read in a remote task so that in Ray client mode, we aren't
        # attempting metadata resolution from the client machine.
        ctx = DatasetContext.get_current()
        prepare_read = cached_remote_fn(
            _prepare_read, retry_exceptions=False, num_cpus=0
        )
        read_tasks = ray.get(
            prepare_read.remote(
                datasource,
                ctx,
                parallelism,
                _wrap_arrow_serialization_workaround(read_args),
            )
        )

    if len(read_tasks) < parallelism and (
        len(read_tasks) < ray.available_resources().get("CPU", parallelism) // 2
    ):
        logger.warning(
            "The number of blocks in this dataset ({}) limits its parallelism to {} "
            "concurrent tasks. This is much less than the number of available "
            "CPU slots in the cluster. Use `.repartition(n)` to increase the number of "
            "dataset blocks.".format(len(read_tasks), len(read_tasks))
        )

    if ray_remote_args is None:
        ray_remote_args = {}
    if "scheduling_strategy" not in ray_remote_args:
        ray_remote_args["scheduling_strategy"] = "SPREAD"

    block_list = LazyBlockList(read_tasks, ray_remote_args=ray_remote_args)
    block_list.compute_first_block()
    block_list.ensure_metadata_for_first_block()

    return Dataset(
        ExecutionPlan(block_list, block_list.stats()),
        0,
        False,
    )
Read a dataset from a custom data source.

    Args:
        datasource: The datasource to read data from.
        parallelism: The requested parallelism of the read. Parallelism may be
            limited by the available partitioning of the datasource.
        read_args: Additional kwargs to pass to the datasource impl.
        ray_remote_args: kwargs passed to ray.remote in the read tasks.

    Returns:
        Dataset holding the data read from the datasource.
    
Read a dataset from a custom data source.
Read a dataset from a custom data source .
def read_datasource(
    datasource: Datasource[T],
    *,
    parallelism: int = 200,
    ray_remote_args: Dict[str, Any] = None,
    **read_args,
) -> Dataset[T]:
    force_local = "RAY_DATASET_FORCE_LOCAL_METADATA" in os.environ
    pa_ds = _lazy_import_pyarrow_dataset()
    if pa_ds:
        partitioning = read_args.get("dataset_kwargs", {}).get("partitioning", None)
        if isinstance(partitioning, pa_ds.Partitioning):
            logger.info(
                "Forcing local metadata resolution since the provided partitioning "
                f"{partitioning} is not serializable."
            )
            force_local = True
    if force_local:
        read_tasks = datasource.prepare_read(parallelism, **read_args)
    else:
        ctx = DatasetContext.get_current()
        prepare_read = cached_remote_fn(
            _prepare_read, retry_exceptions=False, num_cpus=0
        )
        read_tasks = ray.get(
            prepare_read.remote(
                datasource,
                ctx,
                parallelism,
                _wrap_arrow_serialization_workaround(read_args),
            )
        )
    if len(read_tasks) < parallelism and (
        len(read_tasks) < ray.available_resources().get("CPU", parallelism) // 2
    ):
        logger.warning(
            "The number of blocks in this dataset ({}) limits its parallelism to {} "
            "concurrent tasks. This is much less than the number of available "
            "CPU slots in the cluster. Use `.repartition(n)` to increase the number of "
            "dataset blocks.".format(len(read_tasks), len(read_tasks))
        )
    if ray_remote_args is None:
        ray_remote_args = {}
    if "scheduling_strategy" not in ray_remote_args:
        ray_remote_args["scheduling_strategy"] = "SPREAD"
    block_list = LazyBlockList(read_tasks, ray_remote_args=ray_remote_args)
    block_list.compute_first_block()
    block_list.ensure_metadata_for_first_block()
    return Dataset(
        ExecutionPlan(block_list, block_list.stats()),
        0,
        False,
    )
def read_datasource ( datasource : Datasource [ T ] , * , parallelism : int = 200 , ray_remote_args : Dict [ str , Any ] = None , ** read_args , ) -> Dataset [ T ] : force_local = "RAY_DATASET_FORCE_LOCAL_METADATA" in os . environ pa_ds = _lazy_import_pyarrow_dataset ( ) if pa_ds : partitioning = read_args . get ( "dataset_kwargs" , { } ) . get ( "partitioning" , None ) if isinstance ( partitioning , pa_ds . Partitioning ) : logger . info ( "Forcing local metadata resolution since the provided partitioning " f"{partitioning} is not serializable." ) force_local = True if force_local : read_tasks = datasource . prepare_read ( parallelism , ** read_args ) else : ctx = DatasetContext . get_current ( ) prepare_read = cached_remote_fn ( _prepare_read , retry_exceptions = False , num_cpus = 0 ) read_tasks = ray . get ( prepare_read . remote ( datasource , ctx , parallelism , _wrap_arrow_serialization_workaround ( read_args ) , ) ) if len ( read_tasks ) < parallelism and ( len ( read_tasks ) < ray . available_resources ( ) . get ( "CPU" , parallelism ) // 2 ) : logger . warning ( "The number of blocks in this dataset ({}) limits its parallelism to {} " "concurrent tasks. This is much less than the number of available " "CPU slots in the cluster. Use `.repartition(n)` to increase the number of " "dataset blocks." . format ( len ( read_tasks ) , len ( read_tasks ) ) ) if ray_remote_args is None : ray_remote_args = { } if "scheduling_strategy" not in ray_remote_args : ray_remote_args [ "scheduling_strategy" ] = "SPREAD" block_list = LazyBlockList ( read_tasks , ray_remote_args = ray_remote_args ) block_list . compute_first_block ( ) block_list . ensure_metadata_for_first_block ( ) return Dataset ( ExecutionPlan ( block_list , block_list . stats ( ) ) , 0 , False , )
Read a dataset from a custom data source.
Read a dataset from a custom data source .
"""Read a dataset from a custom data source.

    Args:
        datasource: The datasource to read data from.
        parallelism: The requested parallelism of the read. Parallelism may be
            limited by the available partitioning of the datasource.
        read_args: Additional kwargs to pass to the datasource impl.
        ray_remote_args: kwargs passed to ray.remote in the read tasks.

    Returns:
        Dataset holding the data read from the datasource.
    """ # TODO(ekl) remove this feature flag. # Prepare read in a remote task so that in Ray client mode, we aren't # attempting metadata resolution from the client machine.
Dataset holding the data read from the datasource.
Dataset holding the data read from the datasource .
datasource
Datasource[T]
The datasource to read data from.
The datasource to read data from .
parallelism
int
The requested parallelism of the read. Parallelism may be
limited by the available partitioning of the datasource.
The requested parallelism of the read . Parallelism may be limited by the available partitioning of the datasource .
ray_remote_args
Dict[str, Any]
kwargs passed to ray.remote in the read tasks.
kwargs passed to ray . remote in the read tasks .
read_args
Additional kwargs to pass to the datasource impl.
Additional kwargs to pass to the datasource impl .