[Dataset] implement `from_spark`, `to_spark` and some optimizations #17340

kira-lin · 2021-07-26T10:16:07Z

Why are these changes needed?

@ericl Thanks for adding from_arrow, which makes from_spark much easier and faster. It currently only accepts List[ObjectRef[Pyarrow.Table]], however, in RayDP, as data is written to ray in java, the type of ObjectRef is bytes. It can be read to pyarrow.Table by s=pyarrow.ipc.open_stream(data); tb=s.read_all(). I added this to for_block function.

Related issue number

Checks

I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

kira-lin · 2021-07-26T10:16:21Z

Besides, we still have some overhead after data is written to Ray, due to getting metadata. I have a few questions:

We store schema in metadata for each block. But a spark dataframe has a uniform schema. Is this necessary? Similarly for input_files.
In the current implementation, we need to ray.get all of the blocks just to get the metadata. This is probably not ideal. In some cases, it might not make difference because anyway data needs to be fetched eventually. But if the dataset is huge, it might cause too much memory pressure.

ericl · 2021-07-26T17:48:04Z

We store schema in metadata for each block. But a spark dataframe has a uniform schema. Is this necessary? Similarly for input_files.

Yeah it's a good question. One advantage of keeping input files per block is that if the dataset is split, we can still show input files per split accuratley. Also, it avoids the need for both global and per-block metadata. But open to adding global metadata if storing per-block is overly redundant.

In the current implementation, we need to ray.get all of the blocks just to get the metadata. This is probably not ideal. In some cases, it might not make difference because anyway data needs to be fetched eventually. But if the dataset is huge, it might cause too much memory pressure.

Note that you only need to ray.get() the metadata, which is quite small. The actual blocks do not need to be fetched.

ericl · 2021-07-26T17:48:42Z

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):


Can you also change the definition of the Block type to include bytes?

kira-lin · 2021-07-27T02:57:22Z

Note that you only need to ray.get() the metadata, which is quite small. The actual blocks do not need to be fetched.

Yes, only metadata is collected to one node. But the data would be made local on their own nodes, right? In this part no communication is involved, but how about time and memory overhead? Is it omittable?

If not, maybe we can use a whole arrow table for the metadata, and this can be written in java.

ericl · 2021-07-27T03:59:31Z

Hmm you could always leave the metadata blank. It's preferable to fill it out if possible, but maybe that can be an option if it incurs extra performance overhead.

kira-lin · 2021-08-04T09:26:13Z

Hi @ericl , I've been benchmarking our udf to_spark implementation. And I have a few problems:

logging/printing in the udf seems buggy. We need to connect to the ray cluster in the udf function(which is executed in other processes) by either ray.init or ray.client. They both print a single line of log too much times. But I was able to bypass this bug by writing information to a named actor.
Our current approach for to_spark is to first create a spark dataframe, which contains the object ref in ray, and map this dataframe using a udf where the object is retrieved and converted. You can refer to here. The problem is when we use ray.cloudpickle.dumps to serialize the object ref so that we can put them into a dataframe(and be able to retrieve it in udf), it seems like the address of the object changes to the node where cloudpickle.dumps is run. Is it right? We want to have as little remote read as possible. How to solve this problem?

ericl · 2021-08-04T17:30:33Z

@stephanie-wang @rkooo567 any suggestions on how to handle the to_spark() issue above with serializing the object ref?

rkooo567 · 2021-08-04T17:46:21Z

it seems like the address of the object changes to the node where cloudpickle.dumps is run.

Are you saying the object is transferred to the node where it calls cloudpickle.dumps?

kira-lin · 2021-08-05T02:08:52Z

sorry, it turns out my timing went wrong 😭

ericl · 2021-08-13T18:26:13Z

@kira-lin sounds like the issue was resolved, is that right? Any other blocking issues?

kira-lin · 2021-08-16T01:26:36Z

Hi @ericl , we were optimizing to_spark's performance during these weeks. One problem is if block type is pyarrow.Table, we cannot fetch it in java directly, and need to first fetch in python then send to jvm through udf. This is not optimal. We want to be able to directly fetch the data in java. That means block type should be arrow streaming format(bytes). For now we can have these two implementations for two types. What do you think?

ericl · 2021-08-16T02:43:12Z

I see, that makes sense. Would you just add a case for 'bytes' to BlockAccessor.for_block()? I'm wondering if it makes sense to always use the cross language format and not pyarrow.Table at all

…

On Sun, Aug 15, 2021, 6:26 PM Zhi Lin ***@***.***> wrote: Hi @ericl <https://github.com/ericl> , we were optimizing to_spark's performance during these weeks. One problem is if block type is pyarrow.Table, we cannot fetch it in java directly, and need to first fetch in python then send to jvm through udf. This is not optimal. We want to be able to directly fetch the data in java. That means block type should be arrow streaming format(bytes). For now we can have these two implementations for two types. What do you think? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#17340 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAADUSW5KQH7N57Q2L7TYMLT5BSNRANCNFSM5A7ZGCQQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email> .

kira-lin · 2021-08-16T03:26:09Z

Would you just add a case for 'bytes' to BlockAccessor.for_block()?

Sure.

I'm wondering if it makes sense to always use the cross language format and not pyarrow.Table at all

I think having these two types might confuse users. But I'm not sure about the overhead now. I'm wondering if it is necessary to use cross language format. Does Plasma have any optimization for such cases?

ericl · 2021-08-16T05:22:54Z

By cross language format, I just meant always using the bytes format.

…

On Sun, Aug 15, 2021, 8:26 PM Zhi Lin ***@***.***> wrote: Would you just add a case for 'bytes' to BlockAccessor.for_block()? Sure. I'm wondering if it makes sense to always use the cross language format and not pyarrow.Table at all I think having these two types might confuse users. But I'm not sure about the overhead now. I'm wondering if it is necessary to use cross language format. Does Plasma have any optimization for such cases? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#17340 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAADUSQ27Y4ONZ6OYM6MJA3T5CANZANCNFSM5A7ZGCQQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email> .

kira-lin · 2021-08-16T09:51:15Z

Yeah, what I wanted to say is that using streaming format(bytes) introduces some overhead. If we can directly read/write the arrow standard columnar format with Plasma, then no conversions are needed.

ericl · 2021-08-16T16:57:10Z

I see, why not use the random access format then? There should be no overhead with that one.

…

On Mon, Aug 16, 2021, 2:51 AM Zhi Lin ***@***.***> wrote: Yeah, what I wanted to say is that using streaming format(bytes) introduces some overhead. If we can directly read/write the arrow standard columnar format with Plasma, then no conversions are needed. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#17340 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAADUSVQLZWFAADQKJNP7ADT5DNR5ANCNFSM5A7ZGCQQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email> .

kira-lin · 2021-08-17T01:51:51Z

I'm not sure, but Spark uses streaming format to communicate between python and jvm. I guess we'll just use this for now, and update whenever optimization is possible.

kira-lin · 2021-08-18T02:42:27Z

python/ray/_raylet.pyx

+    def get_owner_address(self, ObjectRef object_ref):
+        cdef:
+            CObjectID c_object_id = object_ref.native()
+        return CCoreWorkerProcess.GetCoreWorker().GetOwnerAddress(
+                c_object_id).SerializeAsString()


This function is added because locations of objects need to be passed to java, so that we can ensure locality in spark.

Are you calling this from RayDP? This isn't a public API, so we should call it only from within Ray code (and pass the addresses to external libraries).

Is this comment addressed?

Well I added the developer api, I'm going to directly call it in to_spark now.

kira-lin · 2021-08-19T07:19:23Z

Hi @ericl , I'm wondering what happens when pyarrow.Table is put into Ray object store? Normally objects are serialized, but how does pyarrow.Table get serialized? Because from my understanding, serializing an arrow Table/Record Batch is just to write to IPC stream/file format(this is what pyarrow.RecordBatch.serialize do though it is now deprecated). In other words, the block type of the dataset is still pyarrow.Table, but we use arrow's serialization instead of ray's.

Maybe we should create a plasma buffer and use StreamWriter to write into it? I'm not sure whether this is better than just writing to bytes.

BTW, as RecordBatch is for contiguous memory, I guess we should use it instead of Table for each block?

ericl · 2021-08-19T20:06:06Z

In other words, the block type of the dataset is still pyarrow.Table, but we use arrow's serialization instead of ray's.

There's no such thing as "Ray vs Arrow serialization" here; Ray is supporting pickle5 serialization natively. Hence we use the serializer defined by pyarrow.Table and it writes to Ray shared memory via the pickle5 protocol. Hence, the pickling of pyarrow.Table is already optimal in performance.

BTW, as RecordBatch is for contiguous memory, I guess we should use it instead of Table for each block?

A table is just a convenience wrapper around a list of RecordBatches, it's the same thing.

kira-lin · 2021-09-06T06:38:54Z

@ericl I'm not sure whether the tests I added get tested, can you please take a look? Where should I install raydp for testing?

python/ray/data/dataset.py

ericl · 2021-09-06T19:39:31Z

python/requirements.txt

@@ -59,6 +59,8 @@ moto
 mypy
 networkx
 numba
+raydp-nightly; platform_system != "Windows"


@clarkzinzow @iycheng could you provide the right place to add this for dataset tests?

python/ray/tests/test_raydp_dataset.py

python/ray/data/dataset.py

ericl · 2021-09-07T06:31:57Z

//python/ray/data:tests/test_raydp_dataset PASSED in 35.5s

Seems it worked?

kira-lin · 2021-09-07T06:35:41Z

Seems it worked?

Yes, I just moved raydp to /python/requirements/data_processing/requirements.txt, which is installed only in that test. Hope it still works.

python/ray/data/read_api.py

ericl · 2021-09-08T01:38:29Z

python/ray/data/read_api.py

        df: A Spark dataframe, which must be created by RayDP (Spark-on-Ray).
        parallelism: The amount of parallelism to use for the dataset.
+        If not provided, it will be equal to the number of partitions of the


ericl · 2021-09-08T01:38:58Z

python/ray/data/read_api.py

-def from_arrow(tables: List[ObjectRef["pyarrow.Table"]]) -> Dataset[ArrowRow]:
+def from_arrow(tables: List[ObjectRef[Union["pyarrow.Table", bytes]]],
+               *,
+               parallelism: int = 200) -> Dataset[ArrowRow]:


Remove? This doesn't seem to be used.

ericl · 2021-09-08T01:39:03Z

python/ray/data/read_api.py

-        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.


ericl

LGTM with minor comments.

Co-authored-by: Eric Liang <ekhliang@gmail.com>

ericl · 2021-09-08T17:44:16Z

def spark_dataframe_to_ray_dataset(df: sql.DataFrame, parallelism: int = 0):
num_part = df.rdd.getNumPartitions()
if parallelism in (0, num_part):
pass

  elif parallelism < num_part:

E TypeError: '<' not supported between instances of 'NoneType' and 'int'

ericl · 2021-09-08T22:28:25Z

Also LINT error.

kira-lin · 2021-09-09T05:00:34Z

All fixed, @ericl

ericl self-assigned this Jul 26, 2021

from arrow need a from bytes for java written data

afb1752

ericl reviewed Jul 26, 2021

View reviewed changes

ericl added the @author-action-required The PR author is responsible for the next step. Remove tag to send back to the reviewer. label Jul 26, 2021

add bytes to block type

619404a

ericl added this to the Datasets Beta milestone Jul 30, 2021

kira-lin marked this pull request as ready for review August 17, 2021 08:33

kira-lin added 3 commits August 17, 2021 14:13

owner address is needed for ensure locality

0c8de45

Merge remote-tracking branch 'upstream/master' into fix-from-arrow

0958dbe

use raydp provided functions

cf9f4f4

kira-lin commented Aug 18, 2021

View reviewed changes

format

68e109f

kira-lin added 4 commits September 2, 2021 14:25

update

6200f00

format

6a27cac

add raydp in test dependency

2bfc10a

Merge remote-tracking branch 'upstream/master' into fix-from-arrow

8a7e3bf

Merge remote-tracking branch 'upstream/master' into fix-from-arrow

6f31d7e

ericl reviewed Sep 6, 2021

View reviewed changes

python/ray/data/dataset.py Outdated Show resolved Hide resolved

ericl reviewed Sep 6, 2021

View reviewed changes

ericl reviewed Sep 7, 2021

View reviewed changes

python/ray/tests/test_raydp_dataset.py Outdated Show resolved Hide resolved

ericl reviewed Sep 7, 2021

View reviewed changes

python/ray/data/dataset.py Outdated Show resolved Hide resolved

kira-lin added 5 commits September 7, 2021 11:09

get object locations in ray

b2b5a6f

move test location

00272ca

Merge remote-tracking branch 'upstream/master' into fix-from-arrow

086f571

move raydp dependency

d99c7f4

passing only blocks and schema

c4f62d2

ericl reviewed Sep 8, 2021

View reviewed changes

python/ray/data/read_api.py Outdated Show resolved Hide resolved

ericl reviewed Sep 8, 2021

View reviewed changes

kira-lin and others added 2 commits September 8, 2021 10:29

Update python/ray/data/read_api.py

edfae6e

Co-authored-by: Eric Liang <ekhliang@gmail.com>

minor updates

c1a59d2

lint

2734119

ericl approved these changes Sep 9, 2021

View reviewed changes

ericl merged commit 2fcd1bc into ray-project:master Sep 9, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Dataset] implement `from_spark`, `to_spark` and some optimizations #17340

[Dataset] implement `from_spark`, `to_spark` and some optimizations #17340

kira-lin commented Jul 26, 2021

kira-lin commented Jul 26, 2021

ericl commented Jul 26, 2021 •

edited

Loading

ericl Jul 26, 2021

kira-lin commented Jul 27, 2021

ericl commented Jul 27, 2021

kira-lin commented Aug 4, 2021

ericl commented Aug 4, 2021

rkooo567 commented Aug 4, 2021

kira-lin commented Aug 5, 2021 •

edited

Loading

ericl commented Aug 13, 2021

kira-lin commented Aug 16, 2021

ericl commented Aug 16, 2021 via email

kira-lin commented Aug 16, 2021 •

edited

Loading

ericl commented Aug 16, 2021 via email

kira-lin commented Aug 16, 2021

ericl commented Aug 16, 2021 via email

kira-lin commented Aug 17, 2021

kira-lin Aug 18, 2021

ericl Aug 18, 2021

ericl Sep 6, 2021

kira-lin Sep 7, 2021

kira-lin commented Aug 19, 2021 •

edited

Loading

ericl commented Aug 19, 2021 •

edited

Loading

kira-lin commented Sep 6, 2021

ericl Sep 6, 2021

ericl commented Sep 7, 2021

kira-lin commented Sep 7, 2021

ericl Sep 8, 2021

ericl Sep 8, 2021

ericl Sep 8, 2021

ericl left a comment

ericl commented Sep 8, 2021

ericl commented Sep 8, 2021

kira-lin commented Sep 9, 2021

[Dataset] implement from_spark, to_spark and some optimizations #17340

[Dataset] implement from_spark, to_spark and some optimizations #17340

Conversation

kira-lin commented Jul 26, 2021

Why are these changes needed?

Related issue number

Checks

kira-lin commented Jul 26, 2021

ericl commented Jul 26, 2021 • edited Loading

Choose a reason for hiding this comment

kira-lin commented Jul 27, 2021

ericl commented Jul 27, 2021

kira-lin commented Aug 4, 2021

ericl commented Aug 4, 2021

rkooo567 commented Aug 4, 2021

kira-lin commented Aug 5, 2021 • edited Loading

ericl commented Aug 13, 2021

kira-lin commented Aug 16, 2021

ericl commented Aug 16, 2021 via email

kira-lin commented Aug 16, 2021 • edited Loading

ericl commented Aug 16, 2021 via email

kira-lin commented Aug 16, 2021

ericl commented Aug 16, 2021 via email

kira-lin commented Aug 17, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kira-lin commented Aug 19, 2021 • edited Loading

ericl commented Aug 19, 2021 • edited Loading

kira-lin commented Sep 6, 2021

Choose a reason for hiding this comment

ericl commented Sep 7, 2021

kira-lin commented Sep 7, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ericl left a comment

Choose a reason for hiding this comment

ericl commented Sep 8, 2021

ericl commented Sep 8, 2021

kira-lin commented Sep 9, 2021

[Dataset] implement `from_spark`, `to_spark` and some optimizations #17340

[Dataset] implement `from_spark`, `to_spark` and some optimizations #17340

ericl commented Jul 26, 2021 •

edited

Loading

kira-lin commented Aug 5, 2021 •

edited

Loading

kira-lin commented Aug 16, 2021 •

edited

Loading

kira-lin commented Aug 19, 2021 •

edited

Loading

ericl commented Aug 19, 2021 •

edited

Loading