[Umbrella] Object Storage Support (Help Wanted)

[jerqi]

Code of Conduct

• [X] I agree to follow this project's Code of Conduct

Search before asking

• [X] I have searched in the issues and found no similar issues.

Describe the proposal

Now, remote shuffle storage only support Hadoop Compatible Filesystem. Object storage is also important and widely used in the big data system. Object storage has different implement. Some systems don't support the method of list or have bad performance of the method list. Some systems need to consider file names to use more buckets of object. Some object storage don't support the method append. Now, we use store index and data separately, it will cause many small index files if we don't support the method append. So we should consider merging index files and data files. To have better performance, we should support object storage to adapt different object storage systems. https://docs.google.com/document/d/1E88wZA9Yhr-pGeUEfxo6uSgsIXxg_ivPYBNcTOeaaZA/edit

Task list

• [ ] https://github.com/apache/incubator-uniffle/issues/892
• [ ] https://github.com/apache/incubator-uniffle/issues/391
• [ ] documents for object storage (need to create an issue)
• [ ] https://github.com/apache/incubator-uniffle/issues/1133

Are you willing to submit PR?

• [ ] Yes I am willing to submit a PR!

[jerqi]

I'm not familiar with object storage. Could you give me more inputs? @hiboyang @pspoerri @melodyyangaws @zhaohc10 @LantaoJin @yuyang733

[jerqi]

cc @xianjingfeng @zuston Could we finish this issue together?

[pspoerri]

Regarding upload to S3: As long as you use the Apache HDFS S3A adapter you can stream data to an object store. However you can only append as long as you keep the stream open and you can only do so from a single client. The S3A filesystem implementation uses buffered multi-part uploads to stream a file to an object store. Streaming from multiple clients should be possible in principle, but the coordination overhead and the way Java streams are implemented make things tricky.

Regarding .index files: For better performance you can always cache the index files, or serve them from a different location (Redis, Uniffle Coordinator, etc...).

Regarding list support: You can always store the list of objects somewhere else if you want to avoid any expensive file-listing operations. spark-s3-shuffle only uses listings when it needs to delete objects.

[jerqi]

Regarding upload to S3: As long as you use the Apache HDFS S3A adapter you can stream data to an object store. However you can only append as long as you keep the stream open and you can only do so from a single client. The S3A filesystem implementation uses buffered multi-part uploads to stream a file to an object store. Streaming from multiple clients should be possible in principle, but the coordination overhead and the way Java streams are implemented make things tricky.

Regarding .index files: For better performance you can always cache the index files, or serve them from a different location (Redis, Uniffle Coordinator, etc...).

Regarding list support: You can always store the list of objects somewhere else if you want to avoid any expensive file-listing operations. spark-s3-shuffle only uses listings when it needs to delete objects.

Thanks for your input.

[zuston]

cc @xianjingfeng @zuston Could we finish this issue together?

Yes. I will

Regarding upload to S3: As long as you use the Apache HDFS S3A adapter you can stream data to an object store. However you can only append as long as you keep the stream open and you can only do so from a single client. The S3A filesystem implementation uses buffered multi-part uploads to stream a file to an object store. Streaming from multiple clients should be possible in principle, but the coordination overhead and the way Java streams are implemented make things tricky.

I hope the append could be avoided in this design. And I think it's OK to store same partition data into different files in object store. Like this:

s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/0.index
s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/0.data
s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/1.index
s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/1.data
....


....
s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/990.index
s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/990.data

The one flush of shuffle-server for one partition could be flushed into one file. But this is ensured by the following rules.

1. The partition must be managed by single shuffle-server. Because the id of file prefix name only known with shuffle-server

For reader, it could get the endId (s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/endId.data) from the shuffle-server. That means we need not list operation

If I'm wrong, feel free to point out

[xianjingfeng]

cc @xianjingfeng @zuston Could we finish this issue together?

It is ok for me.

[xianjingfeng]

s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/990.data

s3a://xxxxxxxxxx/{app_id}/{shuffle_id}/{partition_id}/{shuffle_server_id}/990.data may be better?

Regarding .index files: For better performance you can always cache the index files, or serve them from a different location (Redis, Uniffle Coordinator, etc...)

Agree. And I think we can let users to choose whether store index files in memory or other external system.

[hiboyang]

Yeah, a lot of small index files will not work well in object storage like S3. Maybe good idea to store in other places. Or is it possible to serve index file from Spark driver?

[pegasas]

With my limited experience, I think s3 is a good choice in this issue.

If we choose to support s3, then it will easily extend to other filesystems (NFS, CIFS, EFS, GCS fuse, Azure File System) by using a solution like MinIO.

Yeah, a lot of small index files will not work well in object storage like S3. Maybe a good idea to store it in other places. Or is it possible to serve index files from Spark driver?

I think we may have other solution for merging small index files like application-and-practice-of-spark-small-file-merging-function-on-aws-s3?

Feel free to correct me if I am wrong.

[jerqi]

Yeah, a lot of small index files will not work well in object storage like S3. Maybe good idea to store in other places. Or is it possible to serve index file from Spark driver?

Thanks for your input.

[jerqi]

With my limited experience, I think s3 is a good choice in this issue.

If we choose to support s3, then it will easily extend to other filesystems (NFS, CIFS, EFS, GCS fuse, Azure File System) by using a solution like MinIO.

Yeah, a lot of small index files will not work well in object storage like S3. Maybe a good idea to store it in other places. Or is it possible to serve index files from Spark driver?

I think we may have other solution for merging small index files like application-and-practice-of-spark-small-file-merging-function-on-aws-s3?

Feel free to correct me if I am wrong.

I will propose a document at this weekend. First, we can only define some interfaces. Then, we will implement some object systems according to the need of xianjingfeng and zuston.

@xianjingfeng What object system do you want to implement?

[jerqi]

@jiafuzha Do you have extra input?

[xianjingfeng]

@xianjingfeng What object system do you want to implement?

s3

[jiafuzha]

@jiafuzha Do you have extra input?

I was on vacation last week. Do we have any interface defined for RemoteStorageManager? I am looking forward to it since our DAOS is pure remote storage.

[zuston]

Is this on the roadmap? @jerqi @xianjingfeng

[zuston]

I'm interested on this proposal, and will implement this in rust side. @jerqi