[
https://issues.apache.org/jira/browse/DRILL-5975?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16258585#comment-16258585
]
Paul Rogers commented on DRILL-5975:
------------------------------------
The design proposed may work for simple queries, but it is not sufficiently
general for large queries. Let's start with some background. Here is Drill's
current design. As you point out, there are many opportunities for improvement.
But, this is what we have today.
This is an important topic, so I've taken the liberty of explaining our current
understanding in some detail.
h4. Current Model
Drill's current resource model is based on symmetry: it assumes queries arrive
randomly at each Drillbit. That is, that each user connects to a random
Foreman, that each user has similar load, and that the number of users is
larger than the number of Drillbits, so that each Foreman sees roughly equal
load.
Next, Drill assumes that users want maximum performance of each query.
Therefore, it strives for maximum parallelization across both the CPUs in each
node and all nodes in the cluster.
To do this, Drill assumes nodes are symmetrical: that all have the same number
of available CPUs and memory. Drill also assumes that Drill is the cluster is
dedicated to Drill and so Drill attempts to saturate all CPUs on all hosts.
Next, Drill assumes that all queries are fast, and so that query 1 will
complete roughly before query 2 starts. This means that resource sharing is
sequential: query 1 uses all memory and CPUs, then query 2 does so.
The above model may not be ideal, but it is the "simplest thing that could
work." It has gotten Drill this far, but it does clearly have limits.
h4. Limitations of the Current Model
Problems occur, obviously, when the real world does not agree with Drill's
assumptions. Typical problems:
* Drill does not "own" its cluster, and instead shared resources (such as CPU)
with other workloads.
* Queries are not short, and instead of running sequentially, they end up
running in parallel.
* Queries need more resources (CPU or memory) than the user has provided.
h4. CPU Overload
While our recent work focused on memory (with spilling and queue-based memory
allocation), you correctly point out that we need to turn our attention to CPU.
Let's look just a bit deep at the cause of CPU usage. Drill makes two
assumptions:
* Create as many "major fragments" (work units) as possible for each query.
(Shown as different colors in the query plan visualization UI.)
* Create as many "minor fragments" (slices) as possible for each major
fragment. (Shown as numbered items in the query plan UI tables.)
* By default, create a number of minor fragments equal to 70% of the number of
CPUs. (If your machine has 20 cores, Drill will create 14 slices per major
fragment.)
* Every minor fragment is implemented as a Java thread.
A bit of math shows why this is a problem. Assume a typical query with, say, 5
major fragments. Assume 24 CPUs. We will create a number of threads equal to:
24 CPUs * 70% * 5 major fragments = 87 threads
If all these threads were busy at the same time, we'd overload our CPUs by a
factor of 3, causing intense context switching (which pretty much defeats any
attempts to optimize for internal CPU caches.) Larger queries can easily
oversubscribe CPUs by 10 or more times.
Note that each of these threads wants to make use of unlimited memory. It does
not take too many such queries before Drill thrashes the CPU and runs out of
memory.
The lesson is that the workload exceeds the available resources, then the
"assume infinite resources" model no longer works. Some form of throttling is
needed. Let's discuss that.
Suppose that Drill can make a query faster by using all CPUs. Then, there is,
by design, no room in the cluster for another query. If we are already using
300% of CPU, then adding another query simply causes more thrashing, puts more
pressure on memory, and causes both queries to slow down. In an ideal case,
both queries will take twice as long as if they ran separately. In extreme
cases, the slow-down is sub-linear once the OS starts wasting time thrashing
threads (as shown by the percent system time in, say, the "top" command.)
In this (simplified) model, we are better off running one query to completion,
then starting the second. Both make full use of the cluster. Total run time is
the same. Plus, memory pressure is halved.
In general, some queries need all resources, but many do not. In our own
testing, we see that, with TPC-H queries, there is some advantage to running
parallel queries up to maybe three or four concurrent queries. After that, we
just see a linear slow-down. (We've pushed the system to 20, 30 or more queries
-- it is like your local freeway at rush hour; everything becomes vey slow.)
h4. Throttling
The first challenge is to accept that every cluster has limits. Once Drill
saturates CPUs, there is nothing more to give; things will just get slower. No
one likes this truth, but the physics are pretty clear.
This leads to the idea of throttling queries as a whole. To maximize query
performance, allow a limited number of queries into the system. Hold additional
queries until resources are available. The goal would be to keep CPU
utilization at some target (say 90%). Too few concurrent queries and CPU is
wasted. Too many and the CPUs are overused, leading to excessive context
switching and memory pressure.
In the end, a throttling model maximizes query throughput for a given cluster
size. (Of course, we should work to improve Drill efficiency so that each query
requires less CPU. Even so, in any given release, there is some CPU cost per
query that we must manage.)
Once one starts rationing resources (which is what throttling is), users want
some say. The Boss is more important than a worker-bee and so should move to
the head of the line. ("Per-user prioritization.") Dashboard queries should run
before batch reports. ("Application prioritization.") Marketing paid for only
1/3 of the cluster and so should only be able to use that much. ("Resource
pools.") And so on. Impala, as noted, does a pretty good job here.
h4. Minor Fragment Scheduling
The proposal in this query is to schedule at the fragment level. This seems
like a good idea, but the devil is in the details.
* Deadlock can occur if fragment A depends on B and C, but fragment C is
blocked waiting for A to complete. Good DAG dependency analysis will help here.
* Drill gets its speed from in-memory processing. Spilling data to disk between
stages simply reinvents Hive. So, while Drill should do spilling when
absolutely necessary, we wish to minimize extra disk writes.
* Drill is a big data engine: it is designed to deal with large data volumes.
Typically, insufficient memory is available to buffer the entire data set.
(Hence the need for spilling in operations such as sort.) Instead, Drill
attempts to "stream" data from data source, through the DAG, and out to the
client.
* Increased memory pressure as fragment A holds onto buffers while paused.
Ideally, fragments with large memory would have a higher priority to finish so
that they can release memory. But, they may feed into another memory-hungry
operator.
Where fragment-level throttling would be helpful is if a single query were so
large that, by itself, it would create far too many threads. In this case, it
might be a good idea to break the query into "stages" which run sequentially,
with intermediate results materialized to disk. (That is, be Hive-like for
large queries.)
h4. Open vs. Closed Loop Scheduling
Let me introduce one other concept. Impala has tried many ways to control load.
One attempt was to use a "closed-loop" controller: the Impala state store
attempted to monitor actual cluster load. Impala daemons (equivalent to a
Foreman) released queries based on this perceived load. The problems were
exactly what control theory predicts: extreme oscillation. Queries use
resources over time. Monitor load early and it looks like the cluster is idle,
so admit more queries. Eventually, all these queries need peak resources
queries are blocked. But, by now, excessive resources are in use and
out-of-memory errors occur. Queries die, load drops, and the whole cycle starts
again.
Impala moved to the classic YARN-like "open loop" design: queries are assigned
resources and must (presumably) live within that "resource budget." We wish to
learn from Impala and start with the classic open-loop, resource allocation
scheduling model.
h4. Summary
All of this is a way of suggesting that the best next step in throttling is to
implement a better query-level scheduler. Some goals:
* Better estimate query resource usage to do better resource planning. (Better
than the crude large/small model that exists today.)
* Revise each operator so that it lives within a defined memory budget. (That
is, add spilling or other techniques where needed.)
* Find a solution to schedule threads; perhaps limiting slices per major
fragment.
* Find a solution to the major-fragment count problem.
Once query-level throttling works, then we can look at opportunities to
optimize at the slice (minor-fragment) level.
> Resource utilization
> --------------------
>
> Key: DRILL-5975
> URL: https://issues.apache.org/jira/browse/DRILL-5975
> Project: Apache Drill
> Issue Type: New Feature
> Affects Versions: 2.0.0
> Reporter: weijie.tong
> Assignee: weijie.tong
>
> h1. Motivation
> Now the resource utilization radio of Drill's cluster is not too good. Most
> of the cluster resource is wasted. We can not afford too much concurrent
> queries. Once the system accepted more queries with a not high cpu load, the
> query which originally is very quick will become slower and slower.
> The reason is Drill does not supply a scheduler . It just assume all the
> nodes have enough calculation resource. Once a query comes, it will schedule
> the related fragments to random nodes not caring about the node's load. Some
> nodes will suffer more cpu context switch to satisfy the coming query. The
> profound causes to this is that the runtime minor fragments construct a
> runtime tree whose nodes spread different drillbits. The runtime tree is a
> memory pipeline that is all the nodes will stay alone the whole lifecycle of
> a query by sending out data to upper nodes successively, even though some
> node could run quickly and quit immediately.What's more the runtime tree is
> constructed before actual running. The schedule target to Drill will become
> the whole runtime tree nodes.
> h1. Design
> It will be hard to schedule the runtime tree nodes as a whole. So I try to
> solve this by breaking the runtime cascade nodes. The graph below describes
> the initial design.
> !https://raw.githubusercontent.com/wiki/weijietong/drill/images/design.png!
> [graph
> link|https://raw.githubusercontent.com/wiki/weijietong/drill/images/design.png]
> Every Drillbit instance will have a RecordBatchManager which will accept all
> the RecordBatchs written by the senders of local different MinorFragments.
> The RecordBatchManager will hold the RecordBatchs in memory firstly then disk
> storage . Once the first RecordBatch of a MinorFragment sender of one query
> occurs , it will notice the FragmentScheduler. The FragmentScheduler is
> instanced by the Foreman.It holds the whole PlanFragment execution graph.It
> will allocate a new corresponding FragmentExecutor to run the generated
> RecordBatch. The allocated FragmentExecutor will then notify the
> corresponding FragmentManager to indicate that I am ready to receive the
> data. Then the FragmentManger will send out the RecordBatch one by one to the
> corresponding FragmentExecutor's receiver like what the current Sender does
> by throttling the data stream.
> What we can gain from this design is :
> a. The computation leaf node does not to wait for the consumer's speed to end
> its life to release the resource.
> b. The sending data logic will be isolated from the computation nodes and
> shared by different FragmentManagers.
> c. We can schedule the MajorFragments according to Drillbit's actual resource
> capacity at runtime.
> d. Drill's pipeline data processing characteristic is also retained.
> h1. Plan
> This will be a large PR ,so I plan to divide it into some small ones.
> a. to implement the RecordManager.
> b. to implement a simple random FragmentScheduler and the whole event flow.
> c. to implement a primitive FragmentScheduler which may reference the Sparrow
> project.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
