feat: add extension to support DataFrame::checkpoint() - #1993
Conversation
sandugood
changed the title
DataFrame::checkpoint()DataFrame::checkpoint()
|
Is that OK that we'll hold the |
milenkovicm
left a comment
milenkovicm
left a comment
There was a problem hiding this comment.
sorry for very late review @sandugood I'm catching up with backlog after summer break.
the implementation does make sense, but it diverges from spark implementation as ballista checkpoint will materialise dag. perhaps if it is not too complicated we should consider implementing same behaviour like spark. wdyt?
bb83b59 to
d88bb6c
Compare
|
I think, same as |
|
I believe you could get away without physical node if you split into two jobs at planning |
Tried to introduce this concept and it worked out nicely For me it seemed to introduce new |
|
Also, I think it is good to have an example with remote-checkpoint, so user could see the output of it in the ballista-tui 
During this planning I have:
Hence having 3 separated jobs in the TUI |
|
thanks @sandugood will have a look |
github-actions
Bot
added
the
documentation
milenkovicm
left a comment
milenkovicm
left a comment
There was a problem hiding this comment.
thanks @sandugood, my appologise for slow rewview. i have few comments, please have a look when you get chance.
I'm still not sure if we should provide checkpoint_lazy or checkopint(lazy) i guess second approach would give possibility to users to change from ine to other type withut method change, but lets think about it
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn should_insert_checkpoint_node_without_executing() -> datafusion::error::Result<()> |
There was a problem hiding this comment.
should_insert_checkpoint_lazy as name?
| .await?; | ||
| let schema_before = df.schema().clone(); | ||
|
|
||
| let checkpointed = df.checkpoint_lazy()?; |
There was a problem hiding this comment.
should we have test to check execution of checkpoint_lazy or ive missed it
| let plan = match plan { | ||
| SubmitPlan::Logical(logical) if contains_checkpoint(logical) => { | ||
| resolved = | ||
| SubmitPlan::Logical(resolve_checkpoints(&ctx, logical, self).await?); |
There was a problem hiding this comment.
if im not mistaken, this will block job submission (of reading part) until checkpoint creation finishes ? can we create two jobs and link them using job notification subscriber
2 participants
Which issue does this PR close?
Closes #1937 .
Rationale for this change
Before this change, there was no
checkpoint()method for the DataFrame in Ballista (where we break lineage of the plan and store intermediate results to specified location)What changes are included in this PR?
Introducing new
DataFrameExttrait in the client, also introducing bothcheckpoint()andcheckpoint_lazy()there.A new entry to the
BallistaConfigwas added as well. When trying to callcheckpoint()orcheckpoint_lazy()without the config entry set user will get an error.Also added new logic to a separate file (
checkpoint.rs) on theScheduler's side, which is being used during planning in thesubmit_plan()Are there any user-facing changes?
New methods introduced.
checkpoint()method is eager. It breaks lineage and doesn't wait for an action (i.e:show()) => so it executes immediatelycheckpoint_lazy()method is lazy. It breaks lineage but does wait for an action