andygrove commented on PR #2657:
URL:
https://github.com/apache/datafusion-ballista/pull/2657#issuecomment-6441258727
**Benchmark of the pipelined mode at TPC-H SF1000: 15 of 22 queries run and
are much faster, 7 fail with `ConnectionReset` between executors**
I ran TPC-H at SF1000 (Parquet, zstd, Hive-partitioned, S3) on Kubernetes at
`34a1151a`: 32 executors with 8 concurrent tasks each (target partitions 256),
`--task-scheduling-policy push-staged` on the scheduler and executors, 2
iterations per query. The comparison is the same build and cluster shape with
`ballista.execution.mode` left at `blocking` (pull-staged).
## Speed
Pipelined, for the queries that completed (mean of 2 iterations, seconds):
| Query | blocking | pipelined | change |
|---|---|---|---|
| q1 | 15.7 | 13.5 | -14% |
| q3 | 19.9 | 8.7 | -56% |
| q4 | 12.6 | 5.7 | -55% |
| q6 | 1.7 | 2.0 | +23% |
| q7 | 36.7 | 17.2 | -53% |
| q9 | 34.0 | 23.4 | -31% |
| q10 | 37.4 | 25.6 | -31% |
| q12 | 10.8 | 3.6 | -66% |
| q13 | 10.9 | 5.7 | -48% |
| q14 | 6.1 | 1.6 | -74% |
| q15 | 11.2 | 6.4 | -43% |
| q17 | 23.6 | 14.6 | -38% |
| q18 | 57.1 | 34.8 | -39% |
| q19 | 10.6 | 7.2 | -33% |
| q22 | 10.8 | 5.8 | -46% |
q6 is a 2 s query, so I'd read that as noise. The other 14 are 14 to 74%
faster.
## Failures
q2, q5, q8, q11, q16, q20 and q21 failed. Each failed job reports a task
failing because a `FetchPipelinedPartition` stream was reset, for example:
```
Pipelined job 038BX5GTR0000 failed because task 14 of stage 7 failed: Task
failed due to runtime
execution error: GrpcActionError("failed to stream pipelined partition
PartitionId { job_id:
JobId("038BX5GTR0000"), stage_id: 6, partition_id: 113 } of task 25: code:
'Unknown error',
message: "transport error", source: tonic::transport::Error(Transport,
hyper::Error(Io,
Kind(ConnectionReset)))"). Pipelined jobs do not retry tasks
```
The failing stage, task and partition differ from job to job, so it is not
one bad partition. The failing queries tend to be the ones with the most joins
and stages (q5, q8, q21).
What the executor logs show for q2 (job `038BX5GTR0000`):
- All 32 executors have `RESTARTS 0`, so no executor was killed.
- Memory is not close to the limit. Executor logs show about 315 MiB
allocated against a 45,875 MiB pool during the failure.
- The first error is on the reading executor (stage 7 task 14, 13.8 s after
the job launched). The other 33 failures within 0.1 s are the same error
propagating through downstream tasks and the job being cancelled.
- The producing executor logged nothing about the stream. It has no `WARN`
or `ERROR` lines, and its stage 6 task had finished about 10 s earlier. So the
producer did not report a failure, and the reset is visible only on the reading
side.
## What I don't know
- I don't know why the connection resets. My guess, untested, is the fan-in:
each reader drains all 256 upstream partitions concurrently, so a stage 7 task
opens a stream to every stage 6 producer at once, and with 8 tasks per executor
over 32 executors that is a large all-to-all mesh. I'll test it by rerunning
the failed queries on fewer, smaller executors, and with `hyper`/`h2` debug
logging to find which side sends the RST.
- I don't know whether the queries that passed were just lucky.
## Suggestions
- If a reset is plausible under load, retrying only a stream open (before
any batch was consumed) would keep one transient failure from failing the whole
job.
- Log on the producing side when a `FetchPipelinedPartition` stream ends
abnormally, so a reset can be attributed.
- Say in `pipelined-execution.md` what cluster width this is expected to
work at.
Logs and the result JSON are available if useful.
--
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]
