Distribute S3 select over multiple paths and scan ranges #1445
Conversation
AWS CodeBuild CI Report
Powered by github-codebuild-logs, available on the AWS Serverless Application Repository |
AWS CodeBuild CI Report
Powered by github-codebuild-logs, available on the AWS Serverless Application Repository |
AWS CodeBuild CI Report
Powered by github-codebuild-logs, available on the AWS Serverless Application Repository |
| @@ -532,6 +532,20 @@ def pyarrow_types_from_pandas( # pylint: disable=too-many-branches | |||
| return columns_types | |||
|
|
|||
|
|
|||
| def pyarrow2pandas_defaults(use_threads: Union[bool, int], kwargs: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: | |||
| futures.append(self._exec.apply_async(func, arg)) | ||
| return [f.get() for f in futures] | ||
| # Discard boto3.Session object & return futures | ||
| return list(func(*arg) for arg in zip(itertools.repeat(None), *args)) |
There was a problem hiding this comment.
So you're essentially using naive .remote() implementation instead of the pool just wrapping it in the same interface...
Considering the issues we've had with the Pool so far I think this might be the way to go, the only thing is that I wanted pool to handle the results & avoid having to call ray.get if config.distributed outside thus moving as much ray machinery as possible into this class for the benefit of SOC.
This would require a feeding all tasks into the pool at the same time similar to what was done here. Thoughts?
There was a problem hiding this comment.
Agreed - as I said I am not a fan of having to call a wrapper to just do a ray.get and also to flatten a list, just couldn't see a way around it for this use case where we have two ray remote methods (path and scan range)
| return _arrow_refs_to_df(arrow_refs=tables, kwargs=kwargs) # type: ignore | ||
|
|
||
|
|
||
| def list_to_arrow_table( |
There was a problem hiding this comment.
Do you want to keep this version of the function with the schema handling? this is just doing the minimum
There was a problem hiding this comment.
Yeah it does a bit more than what we need it to do right now. Happy to limit it, just couldn't see the harm in expanding it a bit either
| return _arrow_refs_to_df(arrow_refs=tables, kwargs=kwargs) # type: ignore | ||
|
|
||
|
|
||
| def list_to_arrow_table( |
There was a problem hiding this comment.
Yeah it does a bit more than what we need it to do right now. Happy to limit it, just couldn't see the harm in expanding it a bit either
| futures.append(self._exec.apply_async(func, arg)) | ||
| return [f.get() for f in futures] | ||
| # Discard boto3.Session object & return futures | ||
| return list(func(*arg) for arg in zip(itertools.repeat(None), *args)) |
There was a problem hiding this comment.
Agreed - as I said I am not a fan of having to call a wrapper to just do a ray.get and also to flatten a list, just couldn't see a way around it for this use case where we have two ray remote methods (path and scan range)
Feature or Bugfix
Detail
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.