This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow-ballista.git
The following commit(s) were added to refs/heads/master by this push:
new deac8775 Add list of jobs to scheduler UI (#241)
deac8775 is described below
commit deac877545eb7cd2243e7bd5435db817b0153320
Author: Andy Grove <[email protected]>
AuthorDate: Fri Sep 23 10:22:35 2022 -0600
Add list of jobs to scheduler UI (#241)
---
ballista/rust/scheduler/src/api/handlers.rs | 75 +++++++++++++++++++++-
ballista/rust/scheduler/src/api/mod.rs | 14 +++-
.../rust/scheduler/src/state/execution_graph.rs | 6 +-
.../src/state/execution_graph/execution_stage.rs | 14 ++--
ballista/rust/scheduler/src/state/task_manager.rs | 57 +++++++++++++++-
ballista/ui/scheduler/src/App.tsx | 55 +++++-----------
.../ui/scheduler/src/components/QueriesList.tsx | 40 ++++--------
7 files changed, 181 insertions(+), 80 deletions(-)
diff --git a/ballista/rust/scheduler/src/api/handlers.rs
b/ballista/rust/scheduler/src/api/handlers.rs
index 92d2e087..6854636e 100644
--- a/ballista/rust/scheduler/src/api/handlers.rs
+++ b/ballista/rust/scheduler/src/api/handlers.rs
@@ -31,12 +31,23 @@ pub struct ExecutorMetaResponse {
pub last_seen: u128,
}
-pub(crate) async fn scheduler_state<T: AsLogicalPlan, U: AsExecutionPlan>(
+#[derive(Debug, serde::Serialize)]
+pub struct JobResponse {
+ pub job_id: String,
+ pub job_status: String,
+ pub num_stages: usize,
+ pub completed_stages: usize,
+ pub percent_complete: u8,
+}
+
+/// Return current scheduler state, including list of executors and active,
completed, and failed
+/// job ids.
+pub(crate) async fn get_state<T: AsLogicalPlan, U: AsExecutionPlan>(
data_server: SchedulerServer<T, U>,
) -> Result<impl warp::Reply, Rejection> {
// TODO: Display last seen information in UI
- let executors: Vec<ExecutorMetaResponse> = data_server
- .state
+ let state = data_server.state;
+ let executors: Vec<ExecutorMetaResponse> = state
.executor_manager
.get_executor_state()
.await
@@ -49,6 +60,7 @@ pub(crate) async fn scheduler_state<T: AsLogicalPlan, U:
AsExecutionPlan>(
last_seen: duration.as_millis(),
})
.collect();
+
let response = StateResponse {
executors,
started: data_server.start_time,
@@ -56,3 +68,60 @@ pub(crate) async fn scheduler_state<T: AsLogicalPlan, U:
AsExecutionPlan>(
};
Ok(warp::reply::json(&response))
}
+
+/// Return list of jobs
+pub(crate) async fn get_jobs<T: AsLogicalPlan, U: AsExecutionPlan>(
+ data_server: SchedulerServer<T, U>,
+) -> Result<impl warp::Reply, Rejection> {
+ // TODO: Display last seen information in UI
+ let state = data_server.state;
+
+ let jobs = state
+ .task_manager
+ .get_jobs()
+ .await
+ .map_err(|_| warp::reject())?;
+
+ let jobs: Vec<JobResponse> = jobs
+ .iter()
+ .map(|job| JobResponse {
+ job_id: job.job_id.to_string(),
+ job_status: format!("{:?}", job.status),
+ num_stages: job.num_stages,
+ completed_stages: job.completed_stages,
+ percent_complete: ((job.completed_stages as f32 / job.num_stages
as f32)
+ * 100_f32) as u8,
+ })
+ .collect();
+
+ Ok(warp::reply::json(&jobs))
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct JobSummaryResponse {
+ /// Just show debug output for now but what we really want here is a list
of stages with
+ /// plans and metrics and the relationship between them
+ pub summary: String,
+}
+
+/// Get the execution graph for the specified job id
+pub(crate) async fn get_job_summary<T: AsLogicalPlan, U: AsExecutionPlan>(
+ data_server: SchedulerServer<T, U>,
+ job_id: String,
+) -> Result<impl warp::Reply, Rejection> {
+ let graph = data_server
+ .state
+ .task_manager
+ .get_job_execution_graph(&job_id)
+ .await
+ .map_err(|_| warp::reject())?;
+
+ match graph {
+ Some(x) => Ok(warp::reply::json(&JobSummaryResponse {
+ summary: format!("{:?}", x),
+ })),
+ _ => Ok(warp::reply::json(&JobSummaryResponse {
+ summary: "Not Found".to_string(),
+ })),
+ }
+}
diff --git a/ballista/rust/scheduler/src/api/mod.rs
b/ballista/rust/scheduler/src/api/mod.rs
index 2f19f68d..386511f0 100644
--- a/ballista/rust/scheduler/src/api/mod.rs
+++ b/ballista/rust/scheduler/src/api/mod.rs
@@ -85,8 +85,18 @@ fn with_data_server<T: AsLogicalPlan + Clone, U: 'static +
AsExecutionPlan>(
pub fn get_routes<T: AsLogicalPlan + Clone, U: 'static + AsExecutionPlan>(
scheduler_server: SchedulerServer<T, U>,
) -> BoxedFilter<(impl Reply,)> {
- let routes = warp::path!("api" / "state")
+ let route_state = warp::path!("api" / "state")
+ .and(with_data_server(scheduler_server.clone()))
+ .and_then(handlers::get_state);
+
+ let route_jobs = warp::path!("api" / "jobs")
+ .and(with_data_server(scheduler_server.clone()))
+ .and_then(|data_server| handlers::get_jobs(data_server));
+
+ let route_job_summary = warp::path!("api" / "job" / String)
.and(with_data_server(scheduler_server))
- .and_then(handlers::scheduler_state);
+ .and_then(|job_id, data_server| handlers::get_job_summary(data_server,
job_id));
+
+ let routes = route_state.or(route_jobs).or(route_job_summary);
routes.boxed()
}
diff --git a/ballista/rust/scheduler/src/state/execution_graph.rs
b/ballista/rust/scheduler/src/state/execution_graph.rs
index 3d49ee93..4835b850 100644
--- a/ballista/rust/scheduler/src/state/execution_graph.rs
+++ b/ballista/rust/scheduler/src/state/execution_graph.rs
@@ -44,7 +44,7 @@ use ballista_core::serde::{AsExecutionPlan, BallistaCodec};
use crate::display::print_stage_metrics;
use crate::planner::DistributedPlanner;
use crate::scheduler_server::event::QueryStageSchedulerEvent;
-use crate::state::execution_graph::execution_stage::{
+pub(crate) use crate::state::execution_graph::execution_stage::{
CompletedStage, ExecutionStage, FailedStage, ResolvedStage, StageOutput,
UnresolvedStage,
};
@@ -157,6 +157,10 @@ impl ExecutionGraph {
self.stages.len()
}
+ pub(crate) fn stages(&self) -> &HashMap<usize, ExecutionStage> {
+ &self.stages
+ }
+
/// An ExecutionGraph is complete if all its stages are complete
pub fn complete(&self) -> bool {
self.stages
diff --git
a/ballista/rust/scheduler/src/state/execution_graph/execution_stage.rs
b/ballista/rust/scheduler/src/state/execution_graph/execution_stage.rs
index b8d590cf..5ef4d4ad 100644
--- a/ballista/rust/scheduler/src/state/execution_graph/execution_stage.rs
+++ b/ballista/rust/scheduler/src/state/execution_graph/execution_stage.rs
@@ -49,7 +49,7 @@ use crate::display::DisplayableBallistaExecutionPlan;
/// ↓
/// CompletedStage
#[derive(Clone)]
-pub(super) enum ExecutionStage {
+pub(crate) enum ExecutionStage {
UnResolved(UnresolvedStage),
Resolved(ResolvedStage),
Running(RunningStage),
@@ -71,7 +71,7 @@ impl Debug for ExecutionStage {
/// For a stage whose input stages are not all completed, we say it's a
unresolved stage
#[derive(Clone)]
-pub(super) struct UnresolvedStage {
+pub(crate) struct UnresolvedStage {
/// Stage ID
pub(super) stage_id: usize,
/// Output partitioning for this stage.
@@ -89,7 +89,7 @@ pub(super) struct UnresolvedStage {
/// For a stage, if it has no inputs or all of its input stages are completed,
/// then we call it as a resolved stage
#[derive(Clone)]
-pub(super) struct ResolvedStage {
+pub(crate) struct ResolvedStage {
/// Stage ID
pub(super) stage_id: usize,
/// Total number of output partitions for this stage.
@@ -112,7 +112,7 @@ pub(super) struct ResolvedStage {
/// 3. manage the stage-level combined metrics
/// Running stages will only be maintained in memory and will not saved to the
backend storage
#[derive(Clone)]
-pub(super) struct RunningStage {
+pub(crate) struct RunningStage {
/// Stage ID
pub(super) stage_id: usize,
/// Total number of output partitions for this stage.
@@ -135,7 +135,7 @@ pub(super) struct RunningStage {
/// If a stage finishes successfully, its task statuses and metrics will be
finalized
#[derive(Clone)]
-pub(super) struct CompletedStage {
+pub(crate) struct CompletedStage {
/// Stage ID
pub(super) stage_id: usize,
/// Total number of output partitions for this stage.
@@ -158,7 +158,7 @@ pub(super) struct CompletedStage {
/// If a stage fails, it will be with an error message
#[derive(Clone)]
-pub(super) struct FailedStage {
+pub(crate) struct FailedStage {
/// Stage ID
pub(super) stage_id: usize,
/// Total number of output partitions for this stage.
@@ -987,7 +987,7 @@ impl Debug for FailedStage {
/// Each `ExecutionStage` will hold a `StageOutput`s for each of its child
stages.
/// When all tasks for the child stage are complete, it will mark the
`StageOutput`
#[derive(Clone, Debug, Default)]
-pub(super) struct StageOutput {
+pub(crate) struct StageOutput {
/// Map from partition -> partition locations
pub partition_locations: HashMap<usize, Vec<PartitionLocation>>,
/// Flag indicating whether all tasks are complete
diff --git a/ballista/rust/scheduler/src/state/task_manager.rs
b/ballista/rust/scheduler/src/state/task_manager.rs
index e34c04c8..9f9bcb5a 100644
--- a/ballista/rust/scheduler/src/state/task_manager.rs
+++ b/ballista/rust/scheduler/src/state/task_manager.rs
@@ -18,7 +18,7 @@
use crate::scheduler_server::event::QueryStageSchedulerEvent;
use crate::scheduler_server::SessionBuilder;
use crate::state::backend::{Keyspace, Lock, StateBackendClient};
-use crate::state::execution_graph::{ExecutionGraph, Task};
+use crate::state::execution_graph::{ExecutionGraph, ExecutionStage, Task};
use crate::state::executor_manager::{ExecutorManager, ExecutorReservation};
use crate::state::{decode_protobuf, encode_protobuf, with_lock};
use ballista_core::config::BallistaConfig;
@@ -106,6 +106,38 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> TaskManager<T, U>
Ok(())
}
+ /// Get a list of active job ids
+ pub async fn get_jobs(&self) -> Result<Vec<JobOverview>> {
+ let mut job_ids = vec![];
+ for job_id in self.state.scan_keys(Keyspace::ActiveJobs).await? {
+ job_ids.push(job_id);
+ }
+ for job_id in self.state.scan_keys(Keyspace::CompletedJobs).await? {
+ job_ids.push(job_id);
+ }
+ for job_id in self.state.scan_keys(Keyspace::FailedJobs).await? {
+ job_ids.push(job_id);
+ }
+
+ let mut jobs = vec![];
+ for job_id in &job_ids {
+ let graph = self.get_execution_graph(job_id).await?;
+ let mut completed_stages = 0;
+ for stage in graph.stages().values() {
+ if let ExecutionStage::Completed(_) = stage {
+ completed_stages += 1;
+ }
+ }
+ jobs.push(JobOverview {
+ job_id: job_id.clone(),
+ status: graph.status(),
+ num_stages: graph.stage_count(),
+ completed_stages,
+ });
+ }
+ Ok(jobs)
+ }
+
/// Get the status of of a job. First look in the active cache.
/// If no one found, then in the Active/Completed jobs, and then in Failed
jobs
pub async fn get_job_status(&self, job_id: &str) ->
Result<Option<JobStatus>> {
@@ -126,6 +158,22 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> TaskManager<T, U>
}
}
+ /// Get the execution graph of of a job. First look in the active cache.
+ /// If no one found, then in the Active/Completed jobs.
+ pub async fn get_job_execution_graph(
+ &self,
+ job_id: &str,
+ ) -> Result<Option<Arc<ExecutionGraph>>> {
+ if let Some(graph) = self.get_active_execution_graph(job_id).await {
+ Ok(Some(Arc::new(graph.read().await.clone())))
+ } else if let Ok(graph) = self.get_execution_graph(job_id).await {
+ Ok(Some(Arc::new(graph)))
+ } else {
+ // if the job failed then we return no graph for now
+ Ok(None)
+ }
+ }
+
/// Update given task statuses in the respective job and return a tuple
containing:
/// 1. A list of QueryStageSchedulerEvent to publish.
/// 2. A list of reservations that can now be offered.
@@ -550,3 +598,10 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> TaskManager<T, U>
.collect()
}
}
+
+pub struct JobOverview {
+ pub job_id: String,
+ pub status: JobStatus,
+ pub num_stages: usize,
+ pub completed_stages: usize,
+}
diff --git a/ballista/ui/scheduler/src/App.tsx
b/ballista/ui/scheduler/src/App.tsx
index 43d7aa1b..374104ed 100644
--- a/ballista/ui/scheduler/src/App.tsx
+++ b/ballista/ui/scheduler/src/App.tsx
@@ -24,46 +24,9 @@ import { Footer } from "./components/Footer";
import "./App.css";
-function uuidv4() {
- return "xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx".replace(/[xy]/g, function (c) {
- var r = (Math.random() * 16) | 0,
- v = c === "x" ? r : (r & 0x3) | 0x8;
- return v.toString(16);
- });
-}
-
-const getRandomQueries = (num: number): Query[] => {
- const nodes: Query[] = [];
-
- for (let i = 0; i < num; i++) {
- nodes.push({
- started: new Date().toISOString(),
- query:
- "SELECT \n" +
- " employee.id,\n" +
- " employee.first_name,\n" +
- " employee.last_name,\n" +
- ' SUM(DATEDIFF("SECOND", call.start_time, call.end_time)) AS
call_duration_sum\n' +
- "FROM call\n" +
- "INNER JOIN employee ON call.employee_id = employee.id\n" +
- "GROUP BY\n" +
- " employee.id,\n" +
- " employee.first_name,\n" +
- " employee.last_name\n" +
- "ORDER BY\n" +
- " employee.id ASC;",
- status: QueryStatus.RUNNING,
- progress: Math.round(Math.random() * 100),
- uuid: uuidv4(),
- });
- }
- return nodes;
-};
-
-const queries = getRandomQueries(17);
-
const App: React.FunctionComponent<any> = () => {
const [schedulerState, setSchedulerState] = useState(undefined);
+ const [jobs, setJobs] = useState(undefined);
function getSchedulerState() {
return fetch(`/api/state`, {
@@ -76,17 +39,31 @@ const App: React.FunctionComponent<any> = () => {
.then((res) => setSchedulerState(res));
}
+ function getJobs() {
+ return fetch(`/api/jobs`, {
+ method: "POST",
+ headers: {
+ Accept: "application/json",
+ },
+ })
+ .then((res) => res.json())
+ .then((res) => setJobs(res));
+ }
+
useEffect(() => {
getSchedulerState();
+ getJobs();
}, []);
+ console.log(JSON.stringify(schedulerState));
+
return (
<Box>
<Grid minH="100vh">
<VStack alignItems={"flex-start"} spacing={0} width={"100%"}>
<Header schedulerState={schedulerState} />
<Summary schedulerState={schedulerState} />
- <QueriesList queries={queries} />
+ <QueriesList queries={jobs} />
<Footer />
</VStack>
</Grid>
diff --git a/ballista/ui/scheduler/src/components/QueriesList.tsx
b/ballista/ui/scheduler/src/components/QueriesList.tsx
index 447a507f..a5d4e886 100644
--- a/ballista/ui/scheduler/src/components/QueriesList.tsx
+++ b/ballista/ui/scheduler/src/components/QueriesList.tsx
@@ -28,7 +28,7 @@ import {
} from "@chakra-ui/react";
import { Column, DateCell, DataTable, LinkCell } from "./DataTable";
import { FaStop } from "react-icons/fa";
-import { GrPowerReset } from "react-icons/gr";
+import { GrDocumentDownload, GrPowerReset } from "react-icons/gr";
export enum QueryStatus {
QUEUED = "QUEUED",
@@ -38,11 +38,10 @@ export enum QueryStatus {
}
export interface Query {
- uuid: string;
- query: string;
+ job_id: string;
status: QueryStatus;
- progress: number;
- started: string;
+ num_stages: number;
+ percent_complete: number;
}
export interface QueriesListProps {
@@ -54,7 +53,7 @@ export const ActionsCell: (props: any) => React.ReactNode =
(props: any) => {
<Flex>
<FaStop color={"red"} title={"stop"} />
<Box mx={2}></Box>
- <GrPowerReset title={"Retry"} />
+ <GrDocumentDownload title={"plan"} />
</Flex>
);
};
@@ -69,28 +68,23 @@ export const ProgressCell: (props: any) => React.ReactNode
= (props: any) => {
const columns: Column<any>[] = [
{
- Header: "UUID",
- accessor: "uuid",
+ Header: "Job ID",
+ accessor: "job_id",
Cell: LinkCell,
},
{
- Header: "Query",
- accessor: "query",
+ Header: "Status",
+ accessor: "job_status",
},
{
- Header: "Status",
- accessor: "status",
+ Header: "Number of Stages",
+ accessor: "num_stages",
},
{
Header: "Progress",
- accessor: "progress",
+ accessor: "percent_complete",
Cell: ProgressCell,
},
- {
- Header: "Started",
- accessor: "started",
- Cell: DateCell,
- },
{
Header: "Actions",
accessor: "",
@@ -105,7 +99,6 @@ const getSkeletion = () => (
<Skeleton height={5} />
<Skeleton height={5} />
<Skeleton height={5} />
- <Skeleton height={5} />
</>
);
@@ -114,15 +107,8 @@ export const QueriesList:
React.FunctionComponent<QueriesListProps> = ({
}) => {
const isLoaded = typeof queries !== "undefined";
- //TODO: Remove blur once queries api is ready
return (
- <VStack
- flex={1}
- p={4}
- w={"100%"}
- alignItems={"flex-start"}
- filter="blur(3px)"
- >
+ <VStack flex={1} p={4} w={"100%"} alignItems={"flex-start"}>
<Text mb={4}>Queries</Text>
<Stack w={"100%"} flex={1}>
{isLoaded ? (
