[Dataset] implement from_spark, to_spark and some optimizations

python/ray/experimental/data/block.py

@@ -112,6 +112,10 @@ def for_block(block: Block) -> "BlockAccessor[T]":
             from ray.experimental.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.experimental.data.impl.block_builder import \
                 SimpleBlockAccessor

python/ray/experimental/data/impl/arrow_block.py

@@ -124,6 +124,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/experimental/data/read_api.py

@@ -345,7 +345,7 @@ def df_to_block(df: "pandas.DataFrame") -> Block[ArrowRow]:
 
 
 @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.