feat: event log and history server (Spark History Server equivalent) by andygrove · Pull Request #1925 · apache/datafusion-ballista

andygrove · 2026-07-03T01:48:06Z

Which issue does this PR close?

Closes #1923.

Rationale for this change

Ballista has a live TUI that shows jobs, stages, tasks, executors, and metrics by
reading the scheduler's REST API — but that state is ephemeral. Completed jobs are
cleaned up after finished_job_state_clean_up_interval_seconds, and everything is
gone when the scheduler restarts. There is no way to inspect a job after the fact,
the way Spark's History Server lets you.

This PR adds a Spark-History-Server equivalent: the scheduler can record a durable
per-job event log during execution, and a standalone history server replays those
logs and serves the same /api/* responses the live scheduler does — so the
existing TUI browses completed jobs unchanged, with no scheduler running.

What changes are included in this PR?

New ballista-history crate (a leaf crate; depends only on serde,
serde_json, tokio):

A versioned, JSONL on-disk event schema (HistoryEvent): JobStart,
StageStart/StageEnd, TaskEnd (the incremental timeline), and a terminal
JobEnd that embeds the fully-built REST DTOs.
The REST response DTO types (JobResponse, QueryStagesResponse, …), shared so
the live scheduler and the history server serialize byte-identical JSON.
An async, buffered EventLogWriter (appends never block the scheduler hot path)
and a reader that folds a completed log into the DTO bundle the history server
serves.

Scheduler:

REST DTOs moved to the shared crate; the graph→DTO construction extracted into
reusable dto_build builders (behavior-preserving — existing API output and
tests are unchanged).
A new event_log_dir config option (off by default). When set, an
EventLogWriter is tee'd off the internal QueryStageSchedulerEvent bus in
QueryStageScheduler::on_receive, mapping JobSubmitted→JobStart,
TaskUpdating→TaskEnd, and JobFinished/JobRunningFailed→JobEnd. When
disabled there is no channel, no task, no file, and no per-event work beyond one
Option check.
The terminal JobEnd is enqueued with a blocking send and flushed before the
loop proceeds, so a completed job's record cannot be dropped under load; its file
handle is then closed.

New ballista-history-server binary:

ballista-history-server --event-log-dir <dir> --bind-host <h> --bind-port <p>
loads completed logs and serves an axum router over the same /api/* paths from
the stored DTOs. Corrupt/partial logs are skipped rather than failing startup;
missing jobs return 404 like the live scheduler.

Scope for this first cut: local-filesystem storage and completed jobs only.
Incremental TaskEnd timeline events are captured but not yet surfaced in a UI.

Verification: the DTO builders are shared by both the live handler and the writer,
and an end-to-end test asserts the history server serves byte-identical JSON to
the live scheduler for the same job.

Are there any user-facing changes?

Yes, all additive and opt-in:

New scheduler flag --event-log-dir <dir> (default: disabled).
New ballista-history-server binary.
The existing TUI can point at a history server (--host/--port) to browse
completed jobs with no live scheduler.

No breaking changes to public APIs; live REST responses are unchanged.

Move the scheduler REST API's JobResponse/TaskSummary/TaskStatus/Percentiles/ QueryStageSummary/QueryStagesResponse types onto the ballista-history crate's shared DTOs, and extract the graph-to-DTO construction logic that previously lived inline in the handlers into pub(crate) builder functions in a new api::dto_build module. Handlers become thin wrappers over these builders, which a later event-log writer will also call. Behavior is unchanged; the existing handler unit tests move to dto_build.rs alongside the moved helper functions they exercise, plus one new plan-string assertion test.

Emit HistoryEvents from the QueryStageScheduler event loop: JobSubmitted -> JobStart, TaskUpdating -> TaskEnd (one per finished task), and JobFinished/JobRunningFailed -> JobEnd. Event builders live in a new scheduler_server::event_log module and reuse the same dto_build DTOs the REST API serves, so a job's JobEnd event matches its live GET /api/job/{id} response. The EventLogWriter is constructed from SchedulerConfig::event_log_dir and threaded into QueryStageScheduler; emission is best-effort and a no-op when event_log_dir is unset.

… missing jobs - HistoryStore::load now skips unreadable/corrupt .eventlog files with a warning instead of failing the whole load, so one bad log can't hide every other completed job. - Job endpoints (/api/job/{id}, /stages, /config, /dot) now return 404 for unknown job ids instead of 200 with a null body, matching the live scheduler's behavior. - Add /api/state with a static payload matching the shape the TUI deserializes at startup. - Strengthen router tests: assert stage-response body contents, cover the corrupt-eventlog skip path, and cover the 404 behavior.

…e scheduler Add an end-to-end DTO parity test that builds live JobResponse/ QueryStagesResponse DTOs, writes the same JobEnd event through the real EventLogWriter, reloads it via HistoryStore::load, and asserts the serialized JSON matches exactly -- proving the history server would serve the same data a running scheduler produced.

EventLogWriter::append enqueues via non-blocking try_send, which drops the event when the channel is saturated. Using it for the terminal JobEnd event meant a completed job's log could end up with JobStart and TaskEnds but no JobEnd, making read_completed_job return None and the job silently disappear from the history server. Add append_final, which awaits channel capacity instead of dropping, and finish_job, which flushes and then closes the per-job file handle (also fixing fd accumulation). Wire both into the JobFinished and JobRunningFailed on_receive arms in place of append + flush_job.

…-api All ballista_history usage in the scheduler is behind the rest-api feature (DTO builders, event-log wiring, history module). Gating the dependency on rest-api keeps it out of the graph for consumers that build the scheduler with default-features = false (e.g. pyballista), and out of non-rest-api builds.

serde_json is only used in production by the rest-api-gated history module (all other uses are in tests, covered by the dev-dependency). Gating it keeps it out of the dependency graph for default-features = false consumers such as pyballista, so python/Cargo.lock stays in sync.

martin-g · 2026-07-10T12:43:10Z

+                    }
+                    log.finish_job(job_id.as_str()).await;
+                }
+                _ => {}


Shouldn't JobCancel and JobPlanningFailed be handled too ? IMO they should call log.finish_job(job_id.as_str()).await; too

Good catch, and it was worse than just a missing record. Cancellation is terminal for a job (the handler below removes the graph), so a cancelled job never got a JobEnd, which meant it was invisible to the history server and its file handle stayed open for the life of the process. Added a JobCancel arm that appends a terminal event and calls finish_job.

One wrinkle: the tee runs before the cancel is applied to the graph, so the graph still reports the job as running at that point. The new job_cancel_event stamps the cancelled status onto the DTO so the history view does not show a cancelled job as perpetually Running. Added JobEndStatus::Cancelled plus a test for it.

JobPlanningFailed I left alone on purpose, and added a comment saying why: it is only posted when submit_job fails, so it happens instead of JobSubmitted. The job has neither an execution graph nor an open log file at that point, so there is nothing to write or close.

martin-g · 2026-07-10T12:45:50Z

+    let percent_complete =
+        ((completed_stages as f32 / num_stages as f32) * 100_f32) as u8;


num_stages could be 0

Suggested change

let percent_complete =

((completed_stages as f32 / num_stages as f32) * 100_f32) as u8;

let percent_complete = if num_stages == 0 {

0

} else {

((completed_stages as f32 / num_stages as f32) * 100_f32) as u8

};

Applied. It does not actually panic today since this is float division and NaN as u8 saturates to 0 in Rust, but relying on that is not great, and build_job_response_from_overview right below already guards the same way. Now they match.

martin-g · 2026-07-10T12:49:50Z

+    if !handles.contains_key(job_id) {
+        let path = log_dir.join(format!("{job_id}.eventlog"));
+        match tokio::fs::OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(&path)
+            .await
+        {
+            Ok(f) => {
+                handles.insert(job_id.to_string(), f);
+            }
+            Err(e) => {
+                eprintln!("event-log writer: cannot open {}: {e}", path.display());
+                return None;
+            }
+        }
+    }
+    handles.get_mut(job_id)


Using https://doc.rust-lang.org/stable/std/collections/struct.HashMap.html#method.entry will be more idiomatic:

Suggested change

if !handles.contains_key(job_id) {

let path = log_dir.join(format!("{job_id}.eventlog"));

match tokio::fs::OpenOptions::new()

.create(true)

.append(true)

.open(&path)

.await

{

Ok(f) => {

handles.insert(job_id.to_string(), f);

}

Err(e) => {

eprintln!("event-log writer: cannot open {}: {e}", path.display());

return None;

}

}

}

handles.get_mut(job_id)

match handles.entry(job_id.to_string()) {

std::collections::hash_map::Entry::Occupied(entry) => Some(entry.into_mut()),

std::collections::hash_map::Entry::Vacant(entry) => {

let path = log_dir.join(format!("{job_id}.eventlog"));

match tokio::fs::OpenOptions::new()

.create(true)

.append(true)

.open(&path)

.await

{

Ok(f) => Some(entry.insert(f)),

Err(e) => {

eprintln!("event-log writer: cannot open {}: {e}", path.display());

None

}

}

}

}

I would rather keep this one as is. entry() takes an owned key, so it allocates a String on every event, including the already-open case which is the overwhelming majority of calls. The contains_key plus get_mut shape is the usual workaround for the borrow checker rejecting the get_mut-then-insert form, and the extra lookup only happens on the first event for a job.

Happy to switch if you feel strongly, the allocation is small, it just seemed like paying it on every event to save one lookup per job.

martin-g · 2026-07-10T12:51:35Z

+
+    let args = Args::parse();
+
+    let store = Arc::new(HistoryStore::load(&args.event_log_dir)?);


execute the synchronous HistoryStore::load() in spawn_blocking():

Suggested change

let store = Arc::new(HistoryStore::load(&args.event_log_dir)?);

let event_log_dir = args.event_log_dir.clone();

let store = Arc::new(

tokio::task::spawn_blocking(move || HistoryStore::load(&event_log_dir))

.await

.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))??,

);

or make load() async and use tokio::fs::** APIs inside it.

Agreed on the problem, though I went a slightly different way: the load now happens in main() before the tokio runtime is built at all, rather than in spawn_blocking. Nothing else is running at startup, so there is no reason for a runtime to exist yet, and it avoids having to map a JoinError into BallistaError.

martin-g · 2026-07-10T12:53:30Z

+        status: String,
+    },
+    TaskEnd {
+        stage_id: u32,


Why u32 ?
Above it uses usize (lines 55 and 59)

The inconsistency was real. TaskEnd came straight off the protobuf TaskStatus (u32) while the stage events came off the graph (usize), and I never reconciled them.

Resolved it toward fixed width: StageStart and StageEnd are now u32 too, with a doc comment on the enum stating the rule. This is a durable, cross-machine format, so a pointer-width type does not really belong in it. Callers holding the scheduler usize stage ids cast on the way in.

martin-g · 2026-07-10T12:57:48Z

+                .status
+                .as_ref()
+                .map(to_api_task_status)
+                .unwrap_or(ApiTaskStatus::Running);


At line 72 you filter out the Running tasks but here if the status is None you map it to Running. The method doc also says that Running should be skipped

Right, that fallback contradicted both the filter above it and the doc comment. Rewrote it as a filter_map so a status-less update is skipped outright rather than being recorded as a TaskEnd that claims Running. Added a test for the status: None case.

martin-g · 2026-07-10T13:00:10Z

+                    if let Ok(Some(graph)) = self
+                        .state
+                        .task_manager
+                        .get_job_execution_graph(job_id)


What if this returns Err ?
job_end_event() won't be called but log.finish_job(job_id.as_str()).await; is still executed.

The finish_job call is intentional there, and I have added a comment explaining it. It closes the per-job file handle, so a job whose JobEnd could not be built still releases its fd instead of leaking one. The resulting log has no JobEnd line, and the reader treats a log without one as incomplete and skips it, so the history server never serves a half-written job.

What was wrong is that the Err was swallowed silently. Pulled the lookup into an event_log_graph helper that logs both the Err case and the Ok(None) case at warn level, and reused it in all three arms. Event logging still never fails scheduling, it just costs the job its event.

- record cancelled jobs: `JobCancel` is terminal for a job, so without a `JobEnd` the log was left without a terminal record (invisible to the history server) and its file handle stayed open for the life of the process. The tee runs before the cancel is applied to the graph, so the event stamps the cancelled status onto the DTO rather than recording the job as perpetually `Running`. - guard `percent_complete` against a zero stage count, matching `build_job_response_from_overview`. - skip status-less task updates in `task_end_events` instead of recording them as `Running`, which contradicted both the filter and the doc. - report (rather than swallow) a failed or missing execution-graph lookup on the event-log path, and document why `finish_job` still runs. - use fixed-width stage/partition ids throughout the on-disk event schema. - load the history store before the tokio runtime starts, so its blocking directory walk does not park a runtime worker.

andygrove · 2026-08-08T15:33:45Z

Thanks for the reviews so far @martin-g. I'm going to break this work down into some smaller PRs to make the review process easier.

andygrove · 2026-08-08T21:28:24Z

First PR split out of this one is #2256

andygrove · 2026-08-09T13:16:53Z

More PRs split out from this one:

andygrove · 2026-08-09T14:38:06Z

Final PRs, which replace this one

andygrove added 13 commits July 2, 2026 18:10

feat(history): add ballista-history crate with event-log schema

7cbcfea

feat(history): add async buffered event-log writer

564a999

feat(scheduler): add event_log_dir config option

c542809

feat(history): add event-log reader producing ReplayedJob

1eda517

feat(scheduler): add ballista-history-server binary and router

af575ef

chore(history): add Apache license header to Cargo.toml

d919d1b

martin-g reviewed Jul 10, 2026

View reviewed changes

milenkovicm requested a review from martin-g August 7, 2026 15:12

andygrove marked this pull request as draft August 8, 2026 15:33

This was referenced Aug 8, 2026

refactor(scheduler): extract REST API wire types into ballista-api-types #2256

Merged

feat: event log and history server (Spark History Server equivalent) #2259

Closed

feat(history): add the event-log schema, writer, and reader #2260

Merged

This was referenced Aug 9, 2026

feat(scheduler): record a per-job event log behind --event-log-dir #2264

Merged

feat(scheduler): add the history server #2265

Open

andygrove closed this Aug 9, 2026

		let percent_complete =
		((completed_stages as f32 / num_stages as f32) * 100_f32) as u8;


		let args = Args::parse();

		let store = Arc::new(HistoryStore::load(&args.event_log_dir)?);

-    let store = Arc::new(HistoryStore::load(&args.event_log_dir)?);
+    let event_log_dir = args.event_log_dir.clone();
+    let store = Arc::new(
+        tokio::task::spawn_blocking(move || HistoryStore::load(&event_log_dir))
+            .await
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))??,
+    );

Uh oh!

Conversation

andygrove commented Jul 3, 2026

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andygrove Aug 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andygrove commented Aug 8, 2026

Uh oh!

andygrove commented Aug 8, 2026

Uh oh!

andygrove commented Aug 9, 2026

Uh oh!

andygrove commented Aug 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

andygrove Aug 5, 2026 •

edited

Loading