python/ray/data/dataset.py
def write_datasource(self, datasource: Datasource[T], **write_args) -> None:
        """Write the dataset to a custom datasource.

        Examples:
            >>> import ray
            >>> from ray.data.datasource import Datasource
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> class CustomDatasource(Datasource): # doctest: +SKIP
            ...     # define custom data source
            ...     pass # doctest: +SKIP
            >>> ds.write_datasource(CustomDatasource(...)) # doctest: +SKIP

        Time complexity: O(dataset size / parallelism)

        Args:
            datasource: The datasource to write to.
            write_args: Additional write args to pass to the datasource.
        """

        ctx = DatasetContext.get_current()
        blocks, metadata = zip(*self._plan.execute().get_blocks_with_metadata())

        # TODO(ekl) remove this feature flag.
        if "RAY_DATASET_FORCE_LOCAL_METADATA" in os.environ:
            write_results: List[ObjectRef[WriteResult]] = datasource.do_write(
                blocks, metadata, **write_args
            )
        else:
            # Prepare write in a remote task so that in Ray client mode, we
            # don't do metadata resolution from the client machine.
            do_write = cached_remote_fn(_do_write, retry_exceptions=False, num_cpus=0)
            write_results: List[ObjectRef[WriteResult]] = ray.get(
                do_write.remote(
                    datasource,
                    ctx,
                    blocks,
                    metadata,
                    _wrap_arrow_serialization_workaround(write_args),
                )
            )

        progress = ProgressBar("Write Progress", len(write_results))
        try:
            progress.block_until_complete(write_results)
            datasource.on_write_complete(ray.get(write_results))
        except Exception as e:
            datasource.on_write_failed(write_results, e)
            raise
        finally:
            progress.close()
Write the dataset to a custom datasource.

        Examples:
            >>> import ray
            >>> from ray.data.datasource import Datasource
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> class CustomDatasource(Datasource): # doctest: +SKIP
            ...     # define custom data source
            ...     pass # doctest: +SKIP
            >>> ds.write_datasource(CustomDatasource(...)) # doctest: +SKIP

        Time complexity: O(dataset size / parallelism)

        Args:
            datasource: The datasource to write to.
            write_args: Additional write args to pass to the datasource.
        
Write the dataset to a custom datasource.
Write the dataset to a custom datasource .
def write_datasource(self, datasource: Datasource[T], **write_args) -> None:
        ctx = DatasetContext.get_current()
        blocks, metadata = zip(*self._plan.execute().get_blocks_with_metadata())
        if "RAY_DATASET_FORCE_LOCAL_METADATA" in os.environ:
            write_results: List[ObjectRef[WriteResult]] = datasource.do_write(
                blocks, metadata, **write_args
            )
        else:
            do_write = cached_remote_fn(_do_write, retry_exceptions=False, num_cpus=0)
            write_results: List[ObjectRef[WriteResult]] = ray.get(
                do_write.remote(
                    datasource,
                    ctx,
                    blocks,
                    metadata,
                    _wrap_arrow_serialization_workaround(write_args),
                )
            )
        progress = ProgressBar("Write Progress", len(write_results))
        try:
            progress.block_until_complete(write_results)
            datasource.on_write_complete(ray.get(write_results))
        except Exception as e:
            datasource.on_write_failed(write_results, e)
            raise
        finally:
            progress.close()
def write_datasource ( self , datasource : Datasource [ T ] , ** write_args ) -> None : ctx = DatasetContext . get_current ( ) blocks , metadata = zip ( * self . _plan . execute ( ) . get_blocks_with_metadata ( ) ) if "RAY_DATASET_FORCE_LOCAL_METADATA" in os . environ : write_results : List [ ObjectRef [ WriteResult ] ] = datasource . do_write ( blocks , metadata , ** write_args ) else : do_write = cached_remote_fn ( _do_write , retry_exceptions = False , num_cpus = 0 ) write_results : List [ ObjectRef [ WriteResult ] ] = ray . get ( do_write . remote ( datasource , ctx , blocks , metadata , _wrap_arrow_serialization_workaround ( write_args ) , ) ) progress = ProgressBar ( "Write Progress" , len ( write_results ) ) try : progress . block_until_complete ( write_results ) datasource . on_write_complete ( ray . get ( write_results ) ) except Exception as e : datasource . on_write_failed ( write_results , e ) raise finally : progress . close ( )
Write the dataset to a custom datasource.
Write the dataset to a custom datasource .
"""Write the dataset to a custom datasource.

        Examples:
            >>> import ray
            >>> from ray.data.datasource import Datasource
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> class CustomDatasource(Datasource): # doctest: +SKIP
            ...     # define custom data source
            ...     pass # doctest: +SKIP
            >>> ds.write_datasource(CustomDatasource(...)) # doctest: +SKIP

        Time complexity: O(dataset size / parallelism)

        Args:
            datasource: The datasource to write to.
            write_args: Additional write args to pass to the datasource.
        """ # TODO(ekl) remove this feature flag. # Prepare write in a remote task so that in Ray client mode, we # don't do metadata resolution from the client machine.
self
datasource
Datasource[T]
The datasource to write to.
The datasource to write to .
write_args
Additional write args to pass to the datasource.
Additional write args to pass to the datasource .
examples
>>> import ray
>>> from ray.data.datasource import Datasource
>>> ds = ray.data.range(100) # doctest: +SKIP
>>> class CustomDatasource(Datasource): # doctest: +SKIP
# define custom data source
pass # doctest: +SKIP
>>> ds.write_datasource(CustomDatasource(...)) # doctest: +SKIP
>>> import ray >>> from ray . data . datasource import Datasource >>> ds = ray . data . range ( 100 ) # doctest : + SKIP >>> class CustomDatasource ( Datasource ) : # doctest : + SKIP # define custom data source pass # doctest : + SKIP >>> ds . write_datasource ( CustomDatasource ( ... )) # doctest : + SKIP