[Dataset] implement from_spark, to_spark and some optimizations

python/ray/_raylet.pyx

@@ -1667,6 +1667,12 @@ cdef class CoreWorker:
             CCoreWorkerProcess.GetCoreWorker().RemoveLocalReference(
                 c_object_id)
 
+    def get_owner_address(self, ObjectRef object_ref):
+        cdef:
+            CObjectID c_object_id = object_ref.native()
+        return CCoreWorkerProcess.GetCoreWorker().GetOwnerAddress(
+                c_object_id).SerializeAsString()
+
     def serialize_and_promote_object_ref(self, ObjectRef object_ref):
         cdef:
             CObjectID c_object_id = object_ref.native()

python/ray/data/block.py

@@ -16,7 +16,7 @@
 #
 # Block data can be accessed in a uniform way via ``BlockAccessors`` such as
 # ``SimpleBlockAccessor``, ``ArrowBlockAccessor``, and ``TensorBlockAccessor``.
-Block = Union[List[T], np.ndarray, "pyarrow.Table"]
+Block = Union[List[T], np.ndarray, "pyarrow.Table", bytes]
 
 
 @DeveloperAPI
@@ -118,6 +118,10 @@ def for_block(block: Block) -> "BlockAccessor[T]":
             from ray.data.impl.arrow_block import \
                 ArrowBlockAccessor
             return ArrowBlockAccessor(block)
+        elif isinstance(block, bytes):
+            from ray.experimental.data.impl.arrow_block import \
+                ArrowBlockAccessor
+            return ArrowBlockAccessor.from_bytes(block)
         elif isinstance(block, list):
             from ray.data.impl.simple_block import \
                 SimpleBlockAccessor

python/ray/data/dataset.py

@@ -1275,15 +1275,17 @@ def to_modin(self) -> "modin.DataFrame":
         """
         raise NotImplementedError  # P1
 
-    def to_spark(self) -> "pyspark.sql.DataFrame":
+    def to_spark(self,
+                 spark: "pyspark.sql.SparkSession") -> "pyspark.sql.DataFrame":
         """Convert this dataset into a Spark dataframe.
 
         Time complexity: O(dataset size / parallelism)
 
         Returns:
             A Spark dataframe created from this dataset.
         """
-        raise NotImplementedError  # P2
+        import raydp
+        raydp.spark.ray_dataset_to_spark_dataframe(spark, self)
 
     def to_pandas(self) -> List[ObjectRef["pandas.DataFrame"]]:
         """Convert this dataset into a distributed set of Pandas dataframes.

python/ray/data/impl/arrow_block.py

@@ -135,6 +135,11 @@ def __init__(self, table: "pyarrow.Table"):
             raise ImportError("Run `pip install pyarrow` for Arrow support")
         self._table = table
 
+    @classmethod
+    def from_bytes(cls, data: bytes):
+        reader = pyarrow.ipc.open_stream(data)
+        return cls(reader.read_all())
+
     def iter_rows(self) -> Iterator[ArrowRow]:
         outer = self
 

python/ray/data/read_api.py

@@ -477,13 +477,14 @@ def from_pandas(dfs: List[ObjectRef["pandas.DataFrame"]],
 
 
 @PublicAPI(stability="beta")
-def from_arrow(tables: List[ObjectRef["pyarrow.Table"]],
+def from_arrow(tables: List[ObjectRef[Union["pyarrow.Table", bytes]]],
                *,
                parallelism: int = 200) -> Dataset[ArrowRow]:
     """Create a dataset from a set of Arrow tables.
 
     Args:
-        dfs: A list of Ray object references to Arrow tables.
+        tables: A list of Ray object references to Arrow tables,
+                or its streaming format in bytes.
         parallelism: The amount of parallelism to use for the dataset.
 
     Returns:
@@ -501,13 +502,15 @@ def from_spark(df: "pyspark.sql.DataFrame", *,
     """Create a dataset from a Spark dataframe.
 
     Args:
+        spark: A SparkSession, which must be created by RayDP (Spark-on-Ray).
         df: A Spark dataframe, which must be created by RayDP (Spark-on-Ray).
         parallelism: The amount of parallelism to use for the dataset.
 
     Returns:
         Dataset holding Arrow records read from the dataframe.
     """
-    raise NotImplementedError  # P2
+    import raydp
+    return raydp.spark.spark_dataframe_to_ray_dataset(df, parallelism)
 
 
 def _df_to_block(df: "pandas.DataFrame") -> Block[ArrowRow]: