Add Spark-like scheduling mode with shuffle files between stages

[andygrove]

DFRay currently creates "stage processors" for each stage in a query plan and eagerly executes queries. This is good for low-latency use cases but can require a large amount of memory.

I would like to have the option to execute using a Spark-like approach where the output of each query stage is streamed to shuffle files and then the next stage streams from that shuffle file via flight (this is the design implemented in Ballista).

I would also like to have a fixed number of "executors" so that concurrency can be controlled more easily.

This would be helpful to cases where users are migrating away from Spark and want a similar execution model.

[robtandy]

This makes sense to me. DFRay as written tries to lean into low latency execution at all costs, though it does look like it will provide a lot of utility for longer running distributed execution as well.

One of the things DFRay is trying to do is not rewrite or replace any of the ExecutionPlans in DataFusion, including RepartitionExecs. I am not sure if this is the same approach that Ballista uses.

I think, ideally, this new mode of execution diverges as little as possible from the existing mode of execution and we can use the exact same physical plans in both cases.

For this to be true, once all stages are created by the DFRayDataFrame, we should arrange them into a DAG and execute them leaves up.

Assumptions, to make sure I understand what you're thinking here:

• we would execute each stage to completion before moving up the DAG
• we would change (new mode) the ProcessorService to, instead of making the stage partitions available for streaming via ArrowFlight, it would be asked to execute and materialize the results fully.
• these results would be in Arrow IPC format
• The DFRayStageReaderExec would be changed (new mode) to, instead of streaming input from the remote stage via Arrow Flight, stream input from the previous materialized values

Being that this is on Ray we have some choices:

• Use the Ray object store for intermediate results:
  • Pros:
    • results are in memory only
    • Easy to do from within the python code DFRayProcessor
    • Confirms to Ray Cluster resource allocations and likely won't surprise cluster operators
  • Cons:
    • finite size compared to object storage
  • Notable
    • We would have DFRayStageReaderExec call back into python to get the results from Ray Object store
• Use object store interface only
  • Pros:
    • larger available size
    • consumes less cluster resources
  • Cons:
    • likely slower, perhaps costlier

Being that I used the object store in a previous iteration for DFRay that didn't make it into this repo, I have experience with the first option and would want to try that first. The second option could be a future user facing choice.

Priority wise, I'd like to release 0.1.0 first and then add this functionality. WDYT

[robtandy]

This would allow for a user to specify an exact number of DFRayProcessors to allocate and the stages would execute on those as available, otherwise they wait.

[andygrove]

I think, ideally, this new mode of execution diverges as little as possible from the existing mode of execution and we can use the exact same physical plans in both cases.

Yes.

For this to be true, once all stages are created by the DFRayDataFrame, we should arrange them into a DAG and execute them leaves up.

Yes.

Assumptions, to make sure I understand what you're thinking here:

• we would execute each stage to completion before moving up the DAG

Yes, although we can execute all leaf plans in parallel, if resource is available.

• we would change (new mode) the ProcessorService to, instead of making the stage partitions available for streaming via ArrowFlight, it would be asked to execute and materialize the results fully.

Yes, but I think the materialized results would still be streamed via Arrow Flight. The only difference is that the source of the stream is materialized results rather than live execution. The reader probably does not need to change.

• these results would be in Arrow IPC format

Yes.

• The DFRayStageReaderExec would be changed (new mode) to, instead of streaming input from the remote stage via Arrow Flight, stream input from the previous materialized values

Yes.

Being that this is on Ray we have some choices:

• Use the Ray object store for intermediate results:

I was thinking that each stage would just stream results directly to disk shuffle files, but we can try some different approaches.

Priority wise, I'd like to release 0.1.0 first and then add this functionality. WDYT

Absolutely.

[robtandy]

Yes, but I think the materialized results would still be streamed via Arrow Flight. The only difference is that the source of the stream is materialized results rather than live execution. The reader probably does not need to change.

Agree, I like that.