comphead commented on code in PR #7420:
URL: https://github.com/apache/datafusion-comet/pull/7420#discussion_r4460104688
##########
native/native-executors/src/bin/comet-scheduler.rs:
##########
@@ -20,10 +20,26 @@
//!
//! Configured through environment variables, with Ballista's defaults
otherwise:
//! - `COMET_SCHEDULER_BIND_HOST` (default `127.0.0.1`)
+//! - `COMET_SCHEDULER_EXTERNAL_HOST`: the host executors reach this scheduler
at, which they
+//! report task status to (default `localhost`)
//! - `COMET_SCHEDULER_PORT`: Ballista's gRPC port, for executors (default
`50050`)
//! - `COMET_SCHEDULER_FLIGHT_PORT`: the Flight port, for drivers (default
`50060`)
//! - `COMET_SCHEDULER_RESULT_RETENTION_SECONDS`: how long a query's results
stay readable after
//! the query ends, whether or not a driver has read them (default `300`).
`0` keeps them.
+//! - `COMET_GRPC_MAX_MESSAGE_SIZE`: the largest message, in bytes, of
Ballista's gRPC service
+//! between the scheduler and its executors, such as a task with many
partitions of a scan
+//! (default: Ballista's 16 MiB). Set it on the executors as well, for their
side of the same
+//! messages. The Flight service for drivers has no limit. Shuffle reads
between executors keep
+//! the limit of the query's `ballista.client.grpc_max_message_size` (16 MiB
by default), which
+//! this does not change.
+//! - `COMET_SCHEDULER_EVENT_LOG_DIR`: a directory to write one event log per
query to, as
+//! Ballista's scheduler does: the query's plan, a record for each task, and
each stage's rows
Review Comment:
Do the stage rows and compute time hold for Comet stages? In Ballista
55.0.0, `task_row_counts` (`scheduler/src/api/dto_build.rs`) sums
`elapsed_compute` over the operators that report metrics, and takes rows from
the `output_rows` of the root and the leaves. `CometFragmentExec` has no
`metrics()`, `ShuffleWriterExec` reports no `elapsed_compute` of its own, and
the leaves of a Comet stage are the one-row-per-partition descriptor and the
shuffle readers. So I expect compute time near zero and input rows counting
descriptor rows. I have not run it. Could you check `elapsed_compute_nanos` and
`input_rows` in the `TaskEnd` records the new test writes? If they are off, the
doc could name what the log does give for Comet stages, or `CometFragmentExec`
could report baseline metrics in a follow-up.
##########
native/native-executors/src/bin/comet-scheduler.rs:
##########
@@ -43,17 +59,30 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
init_logging();
let mut config = SchedulerConfig {
override_physical_codec: Some(Arc::new(CometPhysicalCodec::default())),
+ // Compiling the event-log writer compiles the REST API in with it, on
the gRPC port.
+ disable_rest_api: true,
..Default::default()
};
if let Some(host) = env_var("COMET_SCHEDULER_BIND_HOST")? {
config.bind_host = host;
}
+ if let Some(host) = env_var("COMET_SCHEDULER_EXTERNAL_HOST")? {
+ config.external_host = host;
+ }
+ if let Some(size) = env_var("COMET_GRPC_MAX_MESSAGE_SIZE")? {
+ config.grpc_server_max_decoding_message_size = size;
+ config.grpc_server_max_encoding_message_size = size;
+ config.grpc_client_max_message_size = size;
+ }
if let Some(port) = env_var("COMET_SCHEDULER_PORT")? {
config.bind_port = port;
}
if let Some(seconds) =
env_var("COMET_SCHEDULER_RESULT_RETENTION_SECONDS")? {
config.finished_job_data_clean_up_interval_seconds = seconds;
}
+ if let Some(dir) = env_var::<String>("COMET_SCHEDULER_EVENT_LOG_DIR")? {
+ config.event_log_dir = Some(dir);
+ }
Review Comment:
Ballista creates `event_log_dir` on a background task, and when that fails,
`run` in `ballista/history/src/writer.rs` logs a warning and returns. The
scheduler then starts normally and every later event logs `dropping event ...
(channel full or closed)`. So an unwritable directory gives no logs and no
startup error, while the other malformed settings fail fast through `env_var`.
Could the binary call `std::fs::create_dir_all` on it here and return the error?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
