Zoltán Borók-Nagy created IMPALA-15688:
------------------------------------------
Summary: Scan-range scheduling ignores host speed: slow executors
set the finish time of scan-heavy queries
Key: IMPALA-15688
URL: https://issues.apache.org/jira/browse/IMPALA-15688
Project: IMPALA
Issue Type: Improvement
Components: Distributed Exec, Backend
Reporter: Zoltán Borók-Nagy
The scheduler gives each scan range to the local replica host with the fewest
assigned bytes (SelectExecutorFromCandidates, scheduler.cc:1775-1804 and :2004,
master c580d36e61). Host speed is not considered, and a coordinator that is
also an executor gets a full share. At runtime an idle instance only takes
ranges from its own host's queue (ScanRangeQueueMt,
hdfs-scan-node-base.cc:257-266, IMPALA-9655), so no host can help a slower one.
*Example: TPC-DS SF10000, 12 executors, HDFS replication 2, MT_DOP=8,
PLANNER=CALCITE, master, best of 3.* Over all 96-instance scan fragments, every
host read within 1% of the median bytes, but two hosts used 1.095x the median
CPU time per byte, likely due to other tenants (lowest in all 8 later
memory-copy tests; a compute-only test was within 1.2% on all hosts). One of
them finished the largest scan fragment last in 46 of 98 queries.
* q75: 4.29-4.52 GiB of store_sales per host; the slow hosts used 71.2 and 69.2
CPU-s (median 54.1) and finished at 33.9 and 33.2 s (median 24.8 s).
* q97: they finished at 39.9 and 40.3 s (median 33.4 s).
* q76 (I/O-bound): 19.3-19.5 GiB per host finished in 20.1-30.5 s (scanner I/O
wait 5.7 s on the fastest host, 51-82 s elsewhere).
Summed over 98 queries, the last host finished the largest scan fragment 189 s
after the median host. The slow hosts changed over days, so fixed weights do
not fit: a prototype giving the coordinator host a weight of 0.9 (with other
pending changes present) saved 5 s on 8 queries (627.7 -> 622.6 s), within
noise.
*Proposal:*
* Work stealing: an executor with an empty queue takes unstarted ranges of a
slower host whose other replica it holds, so reads stay local. It also helps
with runtime-filter skew (IMPALA-13645).
* Optionally weight assigned bytes by recent per-host scan throughput.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
