[jira] [Resolved] (SPARK-46361) Add spark dataset chunk read API (python only)

["Weichen Xu (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-46361?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Weichen Xu resolved SPARK-46361.
--------------------------------
    Resolution: Won't Do

> Add spark dataset chunk read API (python only)
> ----------------------------------------------
>
>                 Key: SPARK-46361
>                 URL: https://issues.apache.org/jira/browse/SPARK-46361
>             Project: Spark
>          Issue Type: Improvement
>          Components: PySpark, Spark Core
>    Affects Versions: 4.0.0
>            Reporter: Weichen Xu
>            Assignee: Weichen Xu
>            Priority: Major
>              Labels: pull-request-available
>
> *Design doc:*
> h1. [https://docs.google.com/document/d/1LHzwCjm2SluHkta_08cM3jxFSgfF-niaCZbtIThG-H8/edit#heading=h.cxcvohcybvo2]
>  
> *Proposed API:*
> {code:java}
> def persist_dataframe_as_chunks(dataframe: DataFrame) -> list[str]:
>     """
>     Persist and materialize the spark dataframe as chunks, each chunk is an arrow batch.
>     It tries to persist data to spark worker memory firstly, if memory is not sufficient,
>     then it fallbacks to persist spilled data to spark worker local disk.
>     Return the list of chunk ids.
>     This function is only available when it is called from spark driver process.
>     """
> def read_chunk(chunk_id):
>     """
>     Read chunk by id, return this chunk as an arrow table.
>     You can call this function from spark driver, spark python UDF python,
>     descendant process of spark driver, or descendant process of spark python UDF worker.
>     """
> def unpersist_chunks(chunk_ids: list[str]) -> None:
>     """
>     Remove chunks by chunk ids.
>     This function is only available when it is called from spark driver process.
>     """{code}
> *Motivation:*
> (1)
> In Ray on spark, we want to support loading Ray data from arbitrary spark Dataframe with in-memory conversion,
> for Ray on spark, Ray datasource read-task runs as child process of Ray worker node, and in Ray on spark, we launch Ray worker node as child process of pyspark UDF worker.
> So that the above proposed API allows descendent python process of pyspark UDF worker to read a chunk data of given spark dataframe, based on this, we can achieve efficient "spark DataFrame" to "Ray dataset" conversion.
> (2)
> For spark torch distributor, we want to implement an efficient {{Torch DataLoader}} that loads data from spark dataframe without saving spark dataframe to cloud storage. This API makes it feasible. This issue has a similar pattern to use-case (1)
> (3)
> For petastorm spark dataset converter (see [https://www.databricks.com/blog/2020/06/16/simplify-data-conversion-from-apache-spark-to-tensorflow-and-pytorch.html]) , using the added API, we can achieve better performance



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org