Scheduler的任务
相关概念:Distributed Plan
在Frontend中已经把SQL转换成了single node plan,然后又将其切分成了distributed plan。代码见Planner::createPlan 和DistributedPlanner::createPlanFragments .
比如左图的single node plan会切成为右图的distributed plan(图片来源自impala官方PPT:Query Compilation in Impala )。在impala-shell中set explain_level=3
后用explain可以看到distributed plan。
上图的Distributed Plan中,一个带颜色的圈是一个Plan Fragment。Plan Fragment间的箭头是Fragment间的数据流动,Plan Fragment内矩形间的箭头是Fragment内各个Plan Node的数据流动。上图其实省略了Distributed Plan中每个非叶子结点里的ExchangeNode,其是用来接收其它Plan Fragment的数据的。一个Plan Fragment会有一个或多个instance,运行在不同impalad上。
相关概念:Scan Range
在distributed plan中,每个scan node已经设置好要处理的scan ranges,即有哪些输入文件要处理,每个有哪些block。从Frontend传来的scan range对应一个block,由于一个block有多个replica,因此一个scan range会有多个location。了解scan range的概念非常重要,下面是相关的thrift定义:
struct TScanRange {
1 : optional THdfsFileSplit hdfs_file_split
2 : optional THBaseKeyRange hbase_key_range
}
struct THdfsFileSplit {
1 : required string file_name
2 : required i64 offset
3 : required i64 length
4 : required i64 partition_id
5 : required i64 file_length
6 : required CatalogObjects.THdfsCompression file_compression
7 : required i64 mtime
}
struct THBaseKeyRange {
1 : optional string startKey
2 : optional string stopKey
}
一个ScanRange要么是HDFS文件上的一部分(用file_name,offset和length等来表示),要么是HBase一片连续的rowKey(用[startKey, endKey)表示)。
struct TScanRangeLocation {
1 : required i32 host_idx;
2 : optional i32 volume_id = -1
3 : optional bool is_cached = false
}
struct TScanRangeLocations {
1 : required PlanNodes.TScanRange scan_range
2 : list <TScanRangeLocation> locations
}
TScanRangeLocations
就记录了Scan Range的位置列表,这个类型在TQueryExecRequest
中会被用到。TQueryExecRequest
的成员per_node_scan_ranges
记录了每个ScanNode要处理的一系列ScanRange,每个ScanRange用TScanRangeLocations
来表示。因此per_node_scan_ranges
的类型是 map<Types.TPlanNodeId, list<Planner.TScanRangeLocations>>
.
TQueryExecRequest是Frontend传给Backend的执行计划,Backend会生成 fragment 的执行计划发送给其它 impalad 去执行,这其中就包含了Scheduler的工作。impalad之间通信的数据结构定义在ImpalaInternalService.thrift中,关于ScanRange使用的是TScanRangeParams,此时已经确定了Scan任务要放在哪台机器上去执行(就是接收信息的impalad),因此不再需要location信息里的host_idx。
struct TScanRangeParams {
1 : required PlanNodes.TScanRange scan_range
2 : optional i32 volume_id = -1
3 : optional bool is_cached = false
4 : optional bool is_remote
}
Scheduler的任务
我们知道HDFS是有多备份的,一个block经常会有多于一个复本,因此在HDFS上的位置也会有多于一个。那么读取的时候到底在哪个复本上执行,这就是Scheduler要做的事情。另外其它的上层plan fragment也需要确定在哪些机器上执行。
代码注释如是说:
Given a list of resources and locations returns a list of hosts on which to execute plan fragments requiring those resources.
这其实很像MapReduce里对mapper和reducer的调度。
Scheduler基本接口
virtual Status Init() = 0;
该接口初始化本Scheduler,返回时表示已获取了做调试决策所需要的所有资源。
定义BackendList类型表示一列Backend描述符。
typedef std::vector<TBackendDescriptor> BackendList;
virtual Status GetBackends(const std::vector<TNetworkAddress>& data_locations, BackendList* backends) = 0;
该接口输入是一组 host/port 二元组表示的data location,输出是一组运行在那些机器上或附近(比如同机架)的Backend列表。
virtual Status GetBackend(const TNetworkAddress& data_location, TBackendDescriptor* backend) = 0;
该接口输入是一个 host/port 二元组表示的 data location,输出是一个运行在该机器上或附近的Backend。
virtual bool HasLocalBackend(const TNetworkAddress& data_location) = 0;
返回是否有Backend运行在资源所在的机器上。
virtual void GetAllKnownBackends(BackendList* backends) = 0;
返回该scheduler知道的所有Backend
virtual Status Schedule(Coordinator* coord, QuerySchedule* schedule) = 0;
该接口最主要的输入是QueryScheduler::request(),即一个TQueryExecRequest对象,描述了query的执行计划。参数里的Coordinator只提供一些全局的信息,比如user name等。TQueryExecRequest的per_node_scan_ranges记录了要处理的scan ranges。该接口将这些scan range分配到对应的机器上去,分配信息(即调度信息)记录在QuerySchedule的fragment_exec_params_中。
如果开启了资源管理,即impala运行在Yarn上时,该接口也会向resource manager(通过Llama)申请资源来执行query。该接口在资源申请被同意或拒绝前是阻塞的。
virtual Status Release(QuerySchedule* schedule) = 0;
释放资源
下面的几个接口主要是Impala on Yarn要用到的回调函数,字面意思比较直观
virtual void HandlePreemptedReservation(const TUniqueId& reservation_id) = 0;
virtual void HandlePreemptedResource(const TUniqueId& client_resource_id) = 0;
virtual void HandleLostResource(const TUniqueId& client_resource_id) = 0;
QuerySchedule
在读懂SimpleScheduler的代码前需要先过一下QuerySchedule这个类,SimpleScheduler的任务是把QueryScheduler对象补充完整。
一个QueryScheduler包含了coordinator生成fragment执行请求从而启动query执行所需要的所有信息。Coordinator根据QueryScheduler生成FragmentExecParams,发送给各个impalad去执行fragment instance。
如果开启了资源管理(即Impala on Yarn),则QueryScheduler中还会包含需要申请的资源和已经获得的资源。
query-schedule.h中首先定义了两个类型:
typedef std ::map <TPlanNodeId, std ::vector <TScanRangeParams> > PerNodeScanRanges;
typedef boost::unordered_map <TNetworkAddress, PerNodeScanRanges> FragmentScanRangeAssignment;
PerNodeScanRanges是ScanNode到ScanRange列表的映射,FragmentScanRangeAssignment是impalad到PerNodeScanRanges的映射。
每个Scan Node可能有多个scan range,分配到不同impalad上去执行。因此这里有个impalad到其任务的映射,即它要处理哪些scan node的哪些scan range。
QuerySchedule有几个重要的成员变量:
Name
Type
Description
plan_node_to_fragment_idx_
vector<int32_t>
记录每个PlanNode id到Fragment id的映射
plan_node_to_plan_node_idx_
vector<int32_t>
将PlanNode id映射到其分配在的Fragment的PlanNode列表里的下标。PlanNode id是在single node plan生成的时候就给定的,后续转换成distributed plan后,每个PlanNode从属于一个PlanFragment,每个PlanFragment里有自己的PlanNode列表,这里就需要一个下标。
fragment_exec_params_
vector<FragmentExecParams>
本query的所有PlanFragment的执行请求参数
unique_hosts_
boost::unordered_set<TNetworkAddress>
本query将在哪些机器上执行(除本机外)
其中plan_node_to_fragment_idx_和plan_node_to_plan_node_idx_是在进入SimpleScheduler::Schedule前就生成好的了。Schedule函数最主要的任务是填充fragment_exec_params_和unique_hosts等成员变量。fragment_exec_params是一个FragmentExecParams数组,记录了每个fragment的执行请求,用来发送给其它impalad去运行fragment instance。FragmentExecParams的定义如下:
struct FragmentExecParams {
std ::vector <TNetworkAddress> hosts;
std ::vector <TUniqueId> instance_ids;
std ::vector <TPlanFragmentDestination> destinations;
std ::map <PlanNodeId, int > per_exch_num_senders;
FragmentScanRangeAssignment scan_range_assignment;
int sender_id_base;
};
这里的sender_id_base需要再解释一下。除了根fragment以外的每个fragment都有另一个fragment作为输出目标,除了叶子fragment外的所有fragment都有若干输入fragment。因此除了根fragment以外的每个fragment都需要扮演sender的角色,除了叶子fragment外的所有fragment也都要扮演receiver的角色。每一个fragment都有一个或多个instance,receiver会为自己的各个sender (instance)分配id用来区分,id从0开始。为了方便,我们给同一个fragment的不同instance分配连续的id,这样只需要记住起始id和该fragment的instance数目就够了。成员变量sender_id_base记录的就是自己作为sender时,由receiver分配的起始id。
SimpleScheduler接口实现
目前Impala on Yarn还不是很流行,我们可以先跳过跟resource management相关的代码,这样看起来会更容易些。最主要的逻辑还是Schedule接口:
Status SimpleScheduler::Schedule(Coordinator* coord, QuerySchedule* schedule)
主要做三件事情:
ComputeScanRangeAssignment(schedule->request(), schedule));
填充各个FragmentExecParams对象中的scan_range_assignment
ComputeFragmentHosts(schedule->request(), schedule);
填充各个FragmentExecParams对象中的hosts,即计算每个fragment instance在哪个impalad上执行
ComputeFragmentExecParams(schedule->request(), schedule);
填充各个FragmentExecParams对象中剩下的内容,即destinations、per_exch_num_senders和sender_id_base
调用完这三个函数后,QuerySchedule的fragment_exec_params_就被填充完整了。下面分别介绍这三部分。
ComputeScanRangeAssignment
Status SimpleScheduler::ComputeScanRangeAssignment(const TQueryExecRequest& exec _request,
QuerySchedule* schedule)
正如函数名所述,该函数主要处理scan range的分配,即每个scan range的data host的选择。这里主要是hdfs上的scan range需要处理,因为每个block有多个replica,有多个DataNode可供选择,而hbase上的region只归一个RegionServer(即这里的data host)管理。
exec_request.per_node_scan_ranges是一个map,记录了每个scan node对应的所有scan ranges。该函数的主体是一个for循环,每次处理一个scan node。
for (entry = exec_request.per_node_scan_ranges.begin();
entry != exec_request.per_node_scan_ranges.end(); ++entry) {
const TPlanNodeId node_id = entry->first;
int fragment_idx = schedule->GetFragmentIdx(node_id);
const TPlanFragment& fragment = exec_request.fragments[fragment_idx];
bool exec_at_coord = (fragment.partition.type == TPartitionType::UNPARTITIONED);
const TPlanNode& node = fragment.plan.nodes[schedule->GetNodeIdx(node_id)];
DCHECK_EQ(node.node_id, node_id);
const TReplicaPreference::type * node_replica_preference = node.__isset.hdfs_scan_node
&& node.hdfs_scan_node.__isset.replica_preference
&node.hdfs_scan_node.replica_preference : NULL;
bool node_random_replica = node.__isset.hdfs_scan_node &&
node.hdfs_scan_node.__isset.random_replica &&
node.hdfs_scan_node.random_replica;
FragmentScanRangeAssignment* assignment =
&(*schedule->exec_params())[fragment_idx].scan_range_assignment;
RETURN_IF_ERROR(ComputeScanRangeAssignment(
node_id, node_replica_preference, node_random_replica, entry->second,
exec_request.host_list, exec_at_coord, schedule->query_options(), assignment));
schedule->AddScanRanges(entry->second.size());
}
ComputeScanRangeAssignment
Status SimpleScheduler::ComputeScanRangeAssignment(
PlanNodeId node_id, const TReplicaPreference::type* node_replica_preference,
bool node_random_replica, const vector <TScanRangeLocations> & locations,
const vector <TNetworkAddress> & host_list, bool exec_at_coord,
const TQueryOptions& query_options, FragmentScanRangeAssignment* assignment)
逐个过下参数:
node_id:要处理的scan node的id
node_replica_preference:replica的默认选择策略,是个枚举类例,取值可以是CACHE_LOCAL、DISK_LOCAL等,没有的话就是NULL
node_random_replica:对于条件相同的replica,是否随机选择。否的话就直接选第一个。
locations:各个scan range(也即block)的location列表(replica列表)
host_list:复本所在的host列表,即DataNode列表
exec_at_coord:该scan node是否在Coordinator处运行
query_options:各种查询参数,详见TQueryOption定义
assignment:调度结果存储在该对象中
函数的主体是两层的for循环,第一层为每个scan range,第二层为该scan range的各个location,也即replica。
代码中比较难懂的局部变量是base_distance,所谓的distance不过是CACHE_LOCAL, CACHE_RACK, DISK_LOCAL, DISK_RACK, REMOTE中的一种,表示impalad和它要读取的数据的距离。base_distance是一个基准线,就是说距离比这个还小(优)的我们就当成一样的来对待了。
另一个局部变量random_non_cached_tiebreak意义与参数node_random_replica相同,就是条件都一样的replica中是随机挑一个还是直接选第一个。
除去profile和打log的代码,函数主体的伪代码如下:
assigned_bytes_per_host是一个map ,记录各个DataNode被分配的数据量,即有多少数据要从该DataNode读取
foreach ScanRange in ScanNode
min_distance 记录replica到impalad的最小距离,初始值为REMOTE
min_assigned_bytes,该变量帮助我们找到被分配任务最少的DataNode。每个replica从属于一个DataNode,如果有几个replica的distance相等,则比较它们所在DataNode所要读取的数据量。
这几个局部变量记录最终的选择结果:data_host、volume_id、is_cached、remote_read
num_equivalent_replicas记录各方面条件都相同的replica数目
foreach Location in ScanRange
replica_host 是replica所在的DataNode
计算memory_distance,如果优于base_distance则折合成base_distance
assigned_bytes 是replica所在DataNode当前被分配的数据量
bool 变量found_new_replica表示是否选择该replica
if (memory_distance < min_distance) {
min_distance = memory_distance;
num_equivalent_replicas = 1 ;
found_new_replica = true ;
} else if (memory_distance == min_distance) {
bool cached_replica = memory_distance == TReplicaPreference::CACHE_LOCAL;
if (assigned_bytes < min_assigned_bytes) {
num_equivalent_replicas = 1 ;
found_new_replica = true ;
} else if (assigned_bytes == min_assigned_bytes &&
(random_non_cached_tiebreak || cached_replica)) {
++num_equivalent_replicas;
const int r = rand();
found_new_replica = (r % num_equivalent_replicas == 0 );
}
}
if (found_new_replica) {
min_assigned_bytes = assigned_bytes;
data_host = &replica_host;
volume_id = location.volume_id;
is_cached = location.is_cached;
remote_read = min_distance == TReplicaPreference::REMOTE;
}
}
更新remote_bytes、remote_hosts、local_bytes、cached_bytes、assigned_bytes_per_host
为data_host找一个backend host(即exec_hostport),如果该DataNode所在机器上就有impalad,则选择该impalad,
否则以round robin(轮循)的方式在impalad列表中挑一个。另外,如果函数参数中exec_at_coord为true ,
则该scan node只在Coordinator所在impalad上执行,也就是当前机器。
PerNodeScanRanges* scan_ranges =
FindOrInsert(assignment, exec_hostport, PerNodeScanRanges());
vector <TScanRangeParams> * scan_range_params_list =
FindOrInsert(scan_ranges, node_id, vector <TScanRangeParams> ());
TScanRangeParams scan_range_params;
scan_range_params.scan_range = scan_range_locations.scan_range;
scan_range_params.__set_volume_id(volume_id);
scan_range_params.__set_is_cached(is_cached);
scan_range_params.__set_is_remote(remote_read);
scan_range_params_list->push_back(scan_range_params);
}
回顾一下,assignment是一个映射,记录每个impalad的scan任务,存在一个PerNodeScanRanges对象中。每个PerNodeScanRanges对象也是一个映射,记录一个scan node在该impalad要读的数据,用vector表示。每个TScanRangeParams表示一个读任务,如果is_remote为false,则在所在impalad机器上的DataNode处读取,此时is_cached表示数据是否已缓存在该impalad中,volume_id记录了数据在哪块磁盘。如果is_remote为true,则根据PlanNodes.TScanRange去获取数据源来读取,使用hdfs或hbase的api。
ComputeFragmentHosts
void SimpleScheduler::ComputeFragmentHosts(const TQueryExecRequest& exec_request,
QuerySchedule* schedule) {
上一个Compute函数进行了replica的选取,接下来要给出每个fragment instance在哪个impalad上去执行。
这个函数的输出是schedule->exec_params()中各个FragmentExecParams的hosts部分,最后还会更新下schedule里的unique_hosts_,即整个query涉及到的其它impalad(除去本机)
vector <FragmentExecParams> * fragment_exec_params = schedule->exec_params();
TNetworkAddress coord = MakeNetworkAddress(FLAGS_hostname, FLAGS_be_port);
DCHECK_EQ(fragment_exec_params->size(), exec_request.fragments.size());
vector <TPlanNodeType::type> scan_node_types;
scan_node_types.push_back(TPlanNodeType::HDFS_SCAN_NODE);
scan_node_types.push_back(TPlanNodeType::HBASE_SCAN_NODE);
scan_node_types.push_back(TPlanNodeType::DATA_SOURCE_NODE);
for (int i = exec_request.fragments.size() - 1 ; i >= 0 ; --i) {
const TPlanFragment& fragment = exec_request.fragments[i];
FragmentExecParams& params = (*fragment_exec_params)[i];
if (fragment.partition.type == TPartitionType::UNPARTITIONED) {
params.hosts.push_back(coord);
continue ;
}
if (ContainsNode(fragment.plan, TPlanNodeType::UNION_NODE)) {
vector <TPlanNodeId> scan_nodes;
FindNodes(fragment.plan, scan_node_types, &scan_nodes);
vector <TPlanNodeId> exch_nodes;
FindNodes(fragment.plan,
vector <TPlanNodeType::type> (1 , TPlanNodeType::EXCHANGE_NODE),
&exch_nodes);
vector <TNetworkAddress> scan_hosts;
for (int j = 0 ; j < scan_nodes.size(); ++j) {
GetScanHosts(scan_nodes[j], exec_request, params, &scan_hosts);
}
unordered_set <TNetworkAddress> hosts(scan_hosts.begin(), scan_hosts.end());
for (int j = 0 ; j < exch_nodes.size(); ++j) {
int input_fragment_idx = FindSenderFragment(exch_nodes[j], i, exec_request);
const vector <TNetworkAddress> & input_fragment_hosts =
(*fragment_exec_params)[input_fragment_idx].hosts;
hosts.insert(input_fragment_hosts.begin(), input_fragment_hosts.end());
}
DCHECK(!hosts.empty()) << "no hosts for fragment " << i << " with a UnionNode" ;
params.hosts.assign(hosts.begin(), hosts.end());
continue ;
}
PlanNodeId leftmost_scan_id = FindLeftmostNode(fragment.plan, scan_node_types);
if (leftmost_scan_id == g_ImpalaInternalService_constants.INVALID_PLAN_NODE_ID) {
int input_fragment_idx = FindLeftmostInputFragment(i, exec_request);
params.hosts = (*fragment_exec_params)[input_fragment_idx].hosts;
continue ;
}
GetScanHosts(leftmost_scan_id, exec_request, params, ¶ms.hosts);
}
unordered_set <TNetworkAddress> unique_hosts;
BOOST_FOREACH(const FragmentExecParams& exec_params, *fragment_exec_params) {
unique_hosts.insert(exec_params.hosts.begin(), exec_params.hosts.end());
}
schedule->SetUniqueHosts(unique_hosts);
}
ComputeFragmentExecParams
void SimpleScheduler::ComputeFragmentExecParams(const TQueryExecRequest& exec_request,
QuerySchedule* schedule)
这是SimpleSchedule::Schedule里调用的最后一个Compute函数,用来把各个FragmentExecParams剩下的内容填补完。
函数主体有两部分:先为各个fragment instance分配id,都有了id之后就可以指明各个instance的输入输出,把它们串成一个有向无环图。
这段代码不难了,有以下两点需要解释:
根fragment没有输出,其它的每个fragment都会输出到另一个fragment中的exchange node去,因此需要把目标fragment的各个instance id记录下来,也就是记录本fragment要有几个sender来发送数据,各发到哪些instance去。另外,每个exchange node也要知道有多少个instance会给自己发数据,即sender数目。
每个instance的id是一个128位整数(由两个64位整数表示,类型为TUniqueId),其中的高64位与query id的高64位一样,低64位是query id的低64位加上其是第几个instance。
总结
读懂SimpleScheduler的代码对后续阅读Impala更底层的代码很有帮助。在这里可以弄明白很多基本概念,如Plan Fragment Instance、Scan Range、Exchange Node等。
SimpleScheduler的调度策略很直接,把包含Scan Node的Plan Fragment Instance优先调度到数据所在的机器上去运行,同时会考虑均摊对HDFS DataNode的负载。如果数据所在机器上没有impalad,则用round robin的方式选一个远程的impalad。另外上层的Plan Fragment会尽量在其依赖的Plan Fragment所在机器上去运行。