andygrove commented on code in PR #7420:
URL: https://github.com/apache/datafusion-comet/pull/7420#discussion_r4460687183
##########
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:
I ran the cluster test with a two-stage query (a range stage hash-shuffled
into a stage that reads it) and read the `TaskEnd` records:
| Stage | `input_rows` | `output_rows` | `elapsed_compute_nanos` |
|---|---|---|---|
| Range stage, reads no other stage | 0 | 1000 | 0 |
| Reads the range stage's shuffle | 1000 | 1000 | 3328833 |
Compute time is as you expected: 0 for a stage with no shuffle input, and
otherwise only the `ShuffleReaderExec`'s poll time. Input rows don't count the
descriptor, whose `DataSourceExec` reports no `output_rows`, but they count
only rows read from other stages, so a stage that reads files or makes its own
rows logs 0. Output rows and the task times are right, as is the stage
summary's `elapsed_compute`, which runs from the first task's start to the last
task's end.
In dd77ce75d the doc says this: Ballista sees a Comet block as one operator
with no metrics, so the plan shows a block only by its size, input rows count
only rows read from other stages, and a task's compute time covers only that
reading. I opened #7465 for `CometFragmentExec` to report its block's metrics,
which #7373 needs for the Spark UI as well.
##########
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:
Done in dd77ce75d. The scheduler now calls `create_dir_all` and returns the
error, naming `COMET_SCHEDULER_EVENT_LOG_DIR`. `create_dir_all` succeeds on a
directory that exists but isn't writable, such as a volume mounted with another
owner, and Ballista then only warns for each job, so the scheduler also creates
and removes a file there. `refuses_an_event_log_dir_it_cannot_write` sets the
variable to a path inside a regular file and checks that the scheduler exits
with an error naming the variable. That fails for root too, which CI's
container runs as, and the test fails without the check. By hand, an existing
read-only directory gives `Permission denied (os error 13)`.
--
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]
