CN105808334A - MapReduce short job optimization system and method based on resource reuse - Google Patents

Abstract

The invention discloses a MapReduce short job optimization system and method based on resource reuse; the system includes: a master node, a first-level slave node and several second-level slave nodes, wherein the master node is connected to the first-level slave node, and the first-level slave node Connect with several second-level slave nodes; deploy resource manager and first-level task scheduler on the master node; deploy application manager, task performance evaluator and second-level task scheduler on the first-level slave node, wherein two The first-level task scheduler is connected to the task performance evaluator, and the second-level task scheduler is also connected to the master node; a node manager is deployed on each of the second-level slave nodes. It optimizes the running performance of short jobs from the aspect of improving the effective utilization of resources, reduces the frequency of resource allocation and recycling, uses the time of resource allocation and recycling to run short jobs, and improves the performance of executing short jobs by reducing the time for jobs to wait for resources.

Description

A system and method for optimizing MapReduce short jobs based on resource reuse

technical field

The invention relates to a resource reuse-based MapReduce short job optimization system and method.

Background technique

Industries such as the Internet, finance, and media are facing the challenge of processing large-scale data sets, but conventional data processing tools and computing models cannot meet their requirements. The MapReduce model proposed by Google provides an effective solution for it, and Hadoop is an open source implementation of MapReduce. Hadoop decomposes the submitted job into smaller-grained Map tasks and Reduce tasks. These tasks run in parallel on multiple nodes in the cluster, thus greatly reducing the running time of the job. Hadoop hides the details of parallel computing—distributing data to computing nodes, rerunning failed tasks, etc., allowing users to focus on specific business logic processing. Moreover, Hadoop provides good linear expansion, data redundancy, and high fault tolerance of computing, which make Hadoop a mainstream computing framework for running data-intensive and computing-intensive applications. The success of Hadoop in the industry has prompted the academic community to pay close attention to Hadoop and make suggestions for improving its design and implementation.

Hadoop's initial design goal is to process large-scale jobs in parallel in a large number of computing nodes. However, in actual production, Hadoop is often used to process smaller-scale short jobs. Short jobs refer to jobs whose completion time is less than the set threshold, which is generally set by the user. Small-scale short jobs differ from large-scale jobs in many aspects, such as the size of the input data set, the number of tasks that the job decomposes, the amount of resources required by the task, the completion time of the task, and the user's expectation on the completion time of the job, etc. Since Hadoop does not consider the characteristics of short jobs, the performance of short jobs running in Hadoop is relatively inefficient.

There are many factors that affect job running performance, among which the configuration of cluster nodes, job scheduling algorithm and cluster load are the three key factors. When Hadoop schedules tasks, it assumes that the nodes that make up the cluster are isomorphic, that is, the hardware configurations of each node, such as CPU, memory, and disk, are the same. However, as the company's business expands and the scale of the cluster gradually expands, the configuration of new nodes is significantly higher than that of previous nodes. Therefore, the same task is executed on different nodes, and the completion time of the task varies greatly. In the case of high cluster load, all tasks decomposed by the job cannot immediately apply for the resources to run the tasks, and some tasks enter the queue waiting for resources. After the resource-acquired task is executed, the occupied resource is released, and Hadoop selects the appropriate task from the waiting queue according to the scheduling algorithm and allocates the available resources to the selected task. Therefore, if there are many tasks decomposed by the job, the tasks decomposed by the same job may be executed in multiple rounds to complete. For example, in TaoBao's Hadoop cluster, more than 30% of the Map tasks run for more than two rounds. Therefore, the load situation of the cluster has a decisive influence on the response time and completion time of the job.

In recent years, optimizing the execution performance of MapReduce jobs has gradually become a research hotspot. A lot of research work has improved job execution efficiency from multiple aspects, such as Hadoop framework, job scheduling algorithm, job running process, and hardware accelerators.

Based on the job history data, Piranha summarizes the characteristics of small jobs—few tasks and low failure rate. In the existing technology, the execution process of small jobs is optimized based on the characteristics of a small number of tasks. For example, the Map task and the Reduce task are started at the same time, and the intermediate results of the Map task are directly written to the Reduce task end. Since the failure rate of small jobs is less than 1%, a simple fault tolerance mechanism is adopted - if a job fails to run, Piranha re-executes the entire job instead of the failed task. In the prior art, the batch processing job is decomposed into a large number of tiny tasks, each task only processes 8M, and the task can complete the calculation within a few seconds. Due to the small amount of data input by the task and the short running time, the task does not have the problems of data skew and lagging tasks, and at the same time solves the problem of interactive jobs waiting for resources for a long time. Tenzing provides a set of job process pools to reduce the turnaround time of MapReduce jobs. After a new job is submitted, Tenzing's scheduler selects an idle process from the job process pool to run the submitted job. Using the process pool reduces the cost of starting new jobs, but Tenzing has two disadvantages: one is that the number of reserved processes exceeds actual needs, which wastes cluster resources; the other is that the Tenzing scheduler can only use certain specific process, compromising the nature of task-local computing.

In the prior art, the shortcomings of running short jobs in the Hadoop computing framework are analyzed, and it is proposed to improve the running efficiency of short jobs from the Hadoop framework. The paper optimizes from three aspects: 1) Change the setup task and cleanup task to run on the master node to avoid changing the job status through the heartbeat message, and the job completion time is directly shortened by one heartbeat cycle; 2) The task allocation is changed from " The "pull" mode is changed to the "push" mode to reduce the delay of task assignment; 3) The control message between the master node and the slave node is independent from the current heartbeat message mechanism, and the instant delivery mechanism is adopted.

Spark is also a computing framework for processing large-scale data sets, with an engine for running DAG graphs. The difference between Spark and Hadoop is that Spark is based on memory computing. Spark stores the intermediate results of jobs in memory instead of local disk or HDFS. Subsequent jobs in the DAG graph directly read input data from memory for calculation. The goal of SpongeFiles is to alleviate the problem of data skew, but it is very effective in reducing job runtime. In the existing technology, distributed memory is introduced, and the output results of the Map task are written to the distributed memory first, reducing the time consumed for writing data to the local disk and the time for reading the intermediate results of the Map task in the shuffle stage. This method has an obvious effect on the optimization of the job with large output data of Map task, but has no obvious effect on the job with only Map task and the job with small output data of Map task. In the prior art, job execution efficiency is improved by reducing disk IO.

Sparrow is designed to provide low-latency task scheduling. Sparrow runs some task processes in each node for a long time, and the long-running task processes execute new tasks, reducing the cost of frequent startup of task processes. The number of long-running task processes is statically set by the user or automatically adjusted by the resource manager according to the load of the cluster. Quincy is a task-level scheduler, similar to Sparrow. In order to calculate the optimal scheduling sequence, Quincy maps the scheduling problem into a graph, and considers the location of data, fairness, and hunger when scheduling. Compared to Sparrow, Quincy takes longer to calculate the scheduling sequence.

Hadoop's job scheduling algorithm has a significant impact on the running time of jobs. The FIFO scheduling algorithm schedules jobs in order of their submission time. Because FIFO does not take into account the differences between jobs, the performance of small and interactive jobs is relatively inefficient. The FAIR scheduling algorithm ensures that jobs submitted by users share cluster resources fairly, ensures that short jobs are completed within a reasonable time, and avoids job starvation problems. However, the FAIR scheduling algorithm does not consider the heterogeneity of the cluster and the time constraints of the job. HFSP evaluates the size of the job when the job is running, and sets the small job as a high-priority job to ensure that the small job is completed in the shortest time. The priority of jobs is dynamically adjusted over time to prevent job starvation.

Contents of the invention

The object of the present invention is to solve the above-mentioned problems, provide a kind of MapReduce short job optimization system and method based on resource reuse, it optimizes the running performance of short jobs from the aspect of improving the effective utilization of resources, reduces the frequency of resource allocation and recovery, and allocates resources and The reclaimed time is used to run short jobs, which improves the performance of short jobs by reducing the time for jobs to wait for resources.

In order to achieve the above object, the present invention adopts the following technical solutions:

A MapReduce short job optimization system based on resource reuse, including: a master node, a first-level slave node and several second-level slave nodes, wherein the master node is connected to the first-level slave node, and the first-level slave node is connected to several second-level slave nodes connect;

A resource manager and a first-level task scheduler are deployed on the master node;

An application manager, a task performance evaluator, and a secondary task scheduler are deployed on the first-level slave node, wherein the secondary task scheduler is connected to the task performance evaluator, and the secondary task scheduler is also connected to the master node;

A node manager is deployed on each of the secondary slave nodes.

The resource manager is responsible for the allocation and monitoring of global resources, and the starting and monitoring of the application manager.

The first-level task scheduler is used to schedule the job queue according to the priority of the tasks, the resources required by the tasks, and the time sequence of task submission.

The application manager decomposes the job into a Map task and a Reduce task, and at the same time applies for resources for the Map task and the Reduce task, cooperates with the node manager to run, and is also used to monitor the tasks. The application manager is the control unit for job running, and each job corresponds to an application manager.

The task performance evaluator predicts the completion time of running tasks and unscheduled tasks based on the task performance model.

The secondary task scheduler judges whether the task being executed is a short task according to the prediction result of the task performance evaluator, and selects a task from the unscheduled task queue.

If the task being executed is a short task, the secondary task scheduler selects a new short task from the unscheduled task queue, and the new short task reuses the resources that are about to be released by the currently executing short task; if the executing task is not For short tasks, the resources occupied are released directly after the execution of the task being executed is completed.

When the secondary task scheduler selects unscheduled tasks, it needs to consider the locality of tasks, the running time of tasks, the fairness of resources and the heterogeneity of clusters.

The node manager is responsible for monitoring the amount of resources used by the task to prevent the amount of resources used during the running of the task from exceeding the amount of resources requested by the task.

A method for optimizing short MapReduce jobs based on resource reuse, comprising the following steps:

Step (1): The application manager applies for resources from the resource manager through a heartbeat message;

Step (2): The resource manager allocates idle resources to the application manager that applies for resources, and the application manager obtains the applied resources;

Step (3): The application manager allocates the requested resources to unscheduled tasks, and then the application manager notifies the corresponding node manager to start the task process;

Step (4): The node manager starts the task process to run the task, and the task sends a heartbeat message to the application manager at every set heartbeat period during the running process, and the heartbeat message includes the progress of the task, task statistics and Task health status; task statistics refer to the amount of processed data, output quantity, consumed time, overflow write IO rate; task health status refers to whether the process of executing the task is abnormal;

Step (5): The application manager receives the heartbeat message, and predicts the running time of the task based on the task completion time prediction model in the task performance evaluator. If the task running time is less than or equal to the task completion time set by the user, the current task is a short task; otherwise the current task is a long task;

If the current task is a short task, the secondary task scheduler selects a new task from the unscheduled task queue; if the current task is a long task, the heartbeat message is ignored;

Step (6): The application manager notifies the task process of the selected new task, and the task process continues to run the new task after executing the running task;

Step (7): If the task process does not receive a new task after executing the running task, the task process exits and releases the occupied resources, and the node manager notifies the resource manager of the released resources through a heartbeat message.

The step of sending a heartbeat message to the associated application manager at every set heartbeat period during the operation of the task in step (4) is as follows:

Step (401): judge whether the task progress exceeds the set threshold, if the task progress exceeds the set threshold, calculate the statistical data of the current task, send a statistical heartbeat message, and turn to step (403); otherwise turn to step (402); Task statistics include the number of tasks processed, the elapsed running time and the amount of output data;

Step (402): if the progress of the task does not exceed the set threshold, send a task health heartbeat message; enter step (404);

Step (403): the task receives the heartbeat message from the application manager; enter step (404);

Step (404): Judging whether the task process has received a new task, if the task process has received a new task, go to step (405); otherwise go to (406);

Step (405): read the input data of the new task to the current node; after the execution of the current task is completed, run the newly received task;

Step (406): If the task process does not receive a new task, after the execution of the current task is completed, the task process releases the resource occupied by the task.

Described step (5) predicts the running time of task based on the task completion time prediction model in the task performance evaluator:

If the running process of the task can be divided into multiple sub-phases, the completion time of the task is closely related to the time consumed by the task in the sub-phase, so the task completion time prediction model is established according to the time consumed by the task in the sub-phase;

The running process of the task is divided into multiple sub-phases. If there is no overlap between the sub-phases, the completion time of the task is the sum of the time of each sub-phase;

If there is overlap between the sub-stages, the overlapping part should be removed;

If the running process of task i is decomposed into n sub-stages, the amount of data processed by each sub-stage is represented by vector s, s=[s ₁ , s ₁ ,...,s _n ], and the rate of data processed by each sub-stage is represented by vector r , r=[r ₁ , r ₁ , . . . , r _n ].

The completion time T _i of task i is:

$\mathrm{<span\; class="notranslate">\; T</span>}<span\; class="notranslate">\; =</span>\underset{\mathrm{<span\; class="notranslate">\; i</span>}<span\; class="notranslate">\; =</span>\mathrm{<span\; class="notranslate">\; 1</span>}}{\overset{\mathrm{<span\; class="notranslate">\; no</span>}}{<span\; class="notranslate">\; \&Sigma;</span>}}\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}}<span\; class="notranslate">\; +</span>\mathrm{<span\; class="notranslate">\; \&alpha;</span>}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 1</span>}<span\; class="notranslate">\; )</span>$

Among them, α is the time to start the task, which changes within a fixed range and is a constant.

The running process of the Map task is decomposed into four sub-phases, namely, the phase of reading input data, the phase of running map operations, the phase of overflowing output data, and the phase of merging intermediate result files. Each phase is represented by read, map, spill, and combine.

Calculate the completion time T _i of Map task i according to formula (1):

${\mathrm{<span\; class="notranslate">\; T</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}<span\; class="notranslate">\; =</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; r</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; d</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; r</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; d</span>}}}<span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; p</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; p</span>}}}<span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}}<span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; c</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; b</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; c</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; b</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}}<span\; class="notranslate">\; +</span>\mathrm{<span\; class="notranslate">\; \&alpha;</span>}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 2</span>}<span\; class="notranslate">\; )</span>$

For short tasks, the running time of the intermediate result file merging stage is very small, which is regarded as a constant.

The number of processes processed in the phase of reading input data and the phase of map operation is the same, which is the amount of input data s task of the _task .

Since there is a partial overlap between the map operation phase and the output data overflow phase, the amount of data in the overflow phase is the output after the map operation, that is, the amount of the last overflow.

Assume the amount of input data processed, For the amount of output data, the cache size set by the s _buffer configuration item, is the amount of data written in the last overflow, formula (2) evolves into:

${\mathrm{<span\; class="notranslate">\; T</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}<span\; class="notranslate">\; =</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; k</span>}}<span\; class="notranslate">\; (</span>\frac{\mathrm{<span\; class="notranslate">\; 1</span>}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; r</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; d</span>}}}<span\; class="notranslate">\; +</span>\frac{\mathrm{<span\; class="notranslate">\; 1</span>}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; p</span>}}}<span\; class="notranslate">\; )</span><span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}^{\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; t</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}}<span\; class="notranslate">\; +</span>\mathrm{<span\; class="notranslate">\; \&beta;</span>}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 3</span>}<span\; class="notranslate">\; )</span>$

${\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}^{\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; t</span>}}<span\; class="notranslate">\; =</span><span\; class="notranslate">\; (</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; k</span>}}<span\; class="notranslate">\; *</span>{\mathrm{<span\; class="notranslate">\; ratio</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}<span\; class="notranslate">\; )</span>{\mathrm{<span\; class="notranslate">\; \%s</span>}}_{\mathrm{<span\; class="notranslate">\; b</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; f</span>}\mathrm{<span\; class="notranslate">\; f</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; r</span>}}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 4</span>}<span\; class="notranslate">\; )</span>$

${\mathrm{<span\; class="notranslate">\; ratio</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}<span\; class="notranslate">\; =</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}^{\mathrm{<span\; class="notranslate">\; d</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}<span\; class="notranslate">\; /</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}^{\mathrm{<span\; class="notranslate">\; d</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 5</span>}<span\; class="notranslate">\; )</span>$

Among them, s _task is the input data size of the task, β is a constant, r _read , r _map , and r _spill are obtained from the statistical heartbeat message sent by the task end. is the amount of data processed in stage i, is the elapsed time in stage i.

When the application manager receives the statistical heartbeat message sent by the task end, it calculates the completion time of the current task according to formula (3).

If task k is running on node w, when predicting the completion time of unscheduled task m running on node w, the processing rate of each sub-stage of task m is

${\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}^{\mathrm{<span\; class="notranslate">\; m</span>}}<span\; class="notranslate">\; =</span>{\mathrm{<span\; class="notranslate">\; \&eta;</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}^{\mathrm{<span\; class="notranslate">\; k</span>}}<span\; class="notranslate">\; ,</span>\mathrm{<span\; class="notranslate">\; i</span>}<span\; class="notranslate">\; \&Element;</span><span\; class="notranslate">\; \{</span>\mathrm{<span\; class="notranslate">\; r</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; d</span>}<span\; class="notranslate">\; ,</span>\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; p</span>}<span\; class="notranslate">\; ,</span>\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}<span\; class="notranslate">\; \}</span><span\; class="notranslate">\; ,</span>$

${\mathrm{<span\; class="notranslate">\; ratio</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}^{\mathrm{<span\; class="notranslate">\; m</span>}}<span\; class="notranslate">\; =</span>{\mathrm{<span\; class="notranslate">\; ratio</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}^{\mathrm{<span\; class="notranslate">\; k</span>}}<span\; class="notranslate">\; .</span>$

The step that the secondary task scheduler of described step (5) selects new task from unscheduled task queue comprises:

Step (51):

According to the locality of tasks, unscheduled tasks are divided into node local tasks, rack local tasks, and other rack tasks;

If the input data of the task is on the node that processes the task, the task is called a node-local task;

If the input data of the task is in the same rack as the node processing the task, the task is called a rack-local task;

If the input data of the task is not in the same rack as the node processing the task, the task is called another rack task;

Then select tasks in the order of node local tasks, rack local tasks, and other rack tasks, and give priority to the task with the smallest running time in each priority;

Step (52): Select the optimal task according to the principle of heterogeneity.

The step of described step (51) is:

Step (511): the application manager receives the statistical heartbeat message sent by the task process, and turns to step (512);

Step (512): the task performance evaluator predicts the completion time of sending the heartbeat task, and turns to step (513);

Step (513): If the completion time of sending the heartbeat task is less than the set time, then calculate the running time of unscheduled tasks on all nodes running the current job, and turn to step (514);

Step (514): Add unscheduled tasks to node local task queues, rack local task queues and other rack task queues respectively according to locality, and go to step (515); locality refers to whether the input data of the task is related to the processing node Same, whether in a rack or not;

Step (515): sort the node local task queue, the rack local task queue and other rack task queues according to the running time of the task; go to step (516);

Step (516): According to the local priority of the task, determine whether the unscheduled task belongs to the optimal task, if it is the optimal task, delete the currently selected unscheduled task from the unscheduled list, and return to the task; otherwise, determine the next Unscheduled tasks until all unscheduled tasks are checked.

The step of described step (52) is:

Step (521): Calculate the running time of the input task on the full node running the current job, go to step (522);

Step (522): Obtain the node with the shortest running time of the calculation input task, go to step (523);

Step (523): Judging whether the obtained node is the same as the input node, if they are the same, the returned task is the optimal task; otherwise, go to step (524);

Step (524): Calculate the income from transferring the task from the input node to the node with the shortest running time. If the income exceeds the set threshold, the returned input task is the optimal task; otherwise, the returned input task is not the optimal task.

The locality of the task means that: if the input data of the task is on the same node as the processing node, it is called a local task; if it is in the same rack, it is called a rack local task; if it is not in a rack, it is called another rack task. It is the relationship between the input data of the task and the processing node.

The task transfer operation benefit refers to: if the task runs on other nodes, the running time is shorter than that on the original node. This shortened portion of time is called the benefit.

The optimal task refers to the task with the shortest running time on the current node.

The beneficial effect of the present invention is to optimize Hadoop and improve the operating efficiency of short jobs. First, the present invention describes the problems existing in the processing of short jobs by analyzing the execution process of jobs in Hadoop. Then, according to the characteristics of tasks running for multiple rounds under high load conditions, a short job optimization mechanism based on resource reuse is proposed to reduce the waste of resources in the process of allocation and recycling by reusing the resources released by the executed tasks. In the case of a high cluster load, the Map task accounts for much more than the Reduce task in the tasks running multiple rounds, so the present invention only optimizes the Map task. Experimental results show that the short job optimization method based on resource reuse proposed by the present invention can effectively reduce the running time of short jobs and significantly improve the utilization rate of cluster resources.

Description of drawings

Figure 1 is a long task running timing diagram;

Figure 2 is the short job calculation framework;

FIG. 3 is a sequence diagram of short task execution flow;

Figure 4(a) shows the error variation of the running task predicted by the task performance model when the process of the task takes different values;

Figure 4(b) shows the relationship between task running time and error;

Figure 5(a) is the running time of the word frequency statistics job;

Figure 5(b) shows the job running time transformed by HiveSQL;

Figure 5(c) shows the optimized job running time for short jobs;

Figure 6 shows the CPU utilization of computing nodes;

Figure 7 shows the memory utilization of computing nodes.

detailed description

The present invention will be further described below in conjunction with the accompanying drawings and embodiments.

Hadoop is a parallel computing platform for processing large-scale data sets, which has the advantages of good scalability, high fault tolerance, and easy programming. Although its original design goal is to process large-scale jobs in parallel in a large number of computing nodes, in actual production, Hadoop is often used to process small-scale short jobs. Since Hadoop does not consider the characteristics of short jobs, the execution of short jobs in Hadoop is relatively inefficient. In view of the above challenges, the present invention first describes the problems existing in the short job processing process by analyzing the job execution process in Hadoop. Then, according to the characteristics of tasks running for multiple rounds under high load conditions, a short job optimization mechanism based on resource reuse is proposed to reduce the waste of resources in the process of allocation and recycling by reusing the resources released by the executed tasks. Experimental results show that the short job optimization method based on resource reuse proposed by the present invention can effectively reduce the running time of short jobs and significantly improve the utilization rate of cluster resources.

The present invention optimizes short jobs based on Hadoop2.x version. This section first analyzes Hadoop's computing framework and task execution process, and then describes the problems existing in Hadoop's execution of short jobs.

Hadoop adopts a master-slave structure, which consists of a master node and multiple slave nodes, as shown in Figure 2. The resource manager (ResourceManager) runs on the master node and is responsible for the allocation and monitoring of global resources, as well as the startup and monitoring of the application manager. The application manager (ApplicationMaster) runs on the slave nodes, and its responsibilities are to decompose the job into Map tasks and Reduce tasks, apply for resources for the Map tasks and Reduce tasks, and cooperate with the node manager to run and monitor tasks. The application manager is the control unit for job running, and each job corresponds to an application manager. The node manager (NodeManager) also runs on the slave node and is responsible for monitoring the amount of resources (such as memory, CPU, etc.) used by the task to prevent the amount of resources used during the running of the task from exceeding the amount of resources requested by the task.

Figure 1 describes the process of the application manager applying for resources for a task and running the task. ① The application manager divides the job into Map tasks and Reduce tasks, and applies for resources from the resource manager through heartbeat messages. ②The resource manager selects the appropriate task from the task queue applying for resources according to a certain scheduling algorithm, and allocates the idle resources to the selected task, and the application manager obtains the applied resource in the next heartbeat message. ③ The application manager allocates the requested resources to appropriate tasks, and then notifies the node manager of the node where the resources are located to start the task. ④The node manager starts the task process to run the task, and the task reports the progress and health status of the task to the application manager to which it belongs at intervals of heartbeat cycles during the running process. ⑤ Release the resources occupied by the task after the task is completed, and the node manager will return the released resources to the resource manager through the heartbeat message.

From the execution process of the above tasks, it is known that the tasks with short running time and tasks with long running time follow the above process, but the execution of tasks with short running time according to the above process has the following problems:

1) The task startup cost is high. It takes at least one heartbeat period (3 seconds by default) for the application manager to apply for resources for the task, and it takes 1 to 2 seconds from the time the application manager notifies the node manager to run the task to the completion of the task initialization. In Taobao's cluster, Map tasks with a running time of less than 10 seconds account for more than 50%, and in Yahoo! The average completion time of a large number of Map tasks in the ad hoc query and data analysis cluster is about 20 seconds, so the task startup time accounts for more than 20% (5/25) of the total time. When a task applies for resources, if the resource manager has no available resources, the task needs to queue up for available resources, which will further increase the task startup cost. The phenomenon of tasks queuing and waiting for available resources often occurs in the cluster.

2) Serious waste of resources. The completed tasks release the occupied resources, and the node manager returns the released resources to the resource manager at intervals of one heartbeat cycle. The resource manager allocates idle resources to tasks waiting for resources, and the application manager to which the tasks belong can only obtain the requested resources at intervals of one heartbeat cycle. It takes 1-2 seconds from the start of the task process to the completion of the task initialization, so it takes 7-8 seconds for the released resources to be reused again, and the heartbeat period is 3 seconds. Therefore, the frequent execution of short-running tasks by the cluster will result in a serious waste of cluster resources.

According to the above analysis, the current process of Hadoop executing tasks is not suitable for executing tasks with short running time. In order to describe the problem conveniently, we introduce a new concept—short task—for short-running tasks.

Definition 1. Let the completion time of task i be T _i , and the user set the task completion time as T _shortTask . If the task completion time satisfies T _i ≤ T _shortTesk , task i is called a short task.

A short job is composed of a large number of short tasks, and running a large number of short tasks reduces the execution performance of the short job. According to the characteristics of multiple rounds of task running under high load conditions, the present invention proposes a short job optimization mechanism based on resource reuse—reducing the waste of resources in the process of allocation and recovery by reusing the resources released by the executed tasks. Tasks that reuse resources are run ahead of time, reducing job runtime.

1 Short job computing framework for resource reuse

This section describes the short job computing framework for resource reuse and the task execution process of short jobs. Task operation requires a certain amount of resources, including memory, CPU, network, disk storage space, task process, etc. These resources can be used by unscheduled tasks. Since the resource quantities required by tasks of different jobs are different, the present invention only considers resource reuse between Map tasks of the same job.

1.1 Short job computing framework

In the framework of Hadoop, the application manager is a job control center, responsible for applying for resources for tasks, and cooperating with node managers to execute and monitor tasks. The present invention adds two components in the application manager——TaskPerformanceEstimator (TaskPerformanceEstimator) and a secondary task scheduler (Sub-scheduler), the frame of which is shown in FIG. 2 .

The task performance estimator predicts the completion time of two types of tasks—running tasks and unscheduled tasks—based on the task performance model. Because the mechanism of resource reuse is only suitable for short tasks, the definition of short tasks requires the running time of the task, which is unknown before the task is completed. The running time of unscheduled tasks on the set node is the key basis for selecting tasks by the secondary task scheduler, and its calculation results directly affect the scheduling order of tasks. Since a task can be divided into sub-phases, the task performance model is built based on the time consumed by the sub-phases of the task. During the execution of the task, statistical data is collected continuously, and the time consumed by the task in each sub-stage is calculated.

The responsibility of the secondary task scheduler is to judge whether the task being executed is a short task according to the prediction result of the task performance evaluator, and to select a task from the unscheduled task queue. If the task being executed is a short task, the secondary task scheduler selects an appropriate task from the unscheduled task queue to reuse the resources that the short task will release. When selecting unscheduled tasks, the secondary task scheduler comprehensively considers the locality of tasks, the running time of tasks, the fairness of resources and the heterogeneity of clusters.

1.2 Task execution process of short jobs

Figure 3 describes the task execution flow of the short job. ① The application manager applies for resources from the resource manager through the heartbeat message. ②The resource manager allocates idle resources to the task of applying for resources, and the application manager obtains the applied resources at the next heartbeat message. ③The application manager allocates the requested resources to the appropriate tasks, and then notifies the corresponding node manager to start the tasks. ④ The node manager starts the task process to run the task, and the task reports the task progress, task statistics and health status to the application manager to which it belongs at intervals of heartbeat cycles during the running process. ⑤ When the application manager receives the heartbeat message, if the current task is a short task, the secondary task scheduler selects a suitable task from the unscheduled task queue. ⑥The application manager notifies the task process of the selected task, and the task process continues to run the newly received task after executing the running task. ⑦ If the task process does not receive a new task after executing the running task, the task process exits and releases the occupied resources, and the node manager notifies the resource manager of the released resources through the heartbeat message.

The short job computing framework does not affect the execution of long tasks, but long tasks still follow the process described in Figure 1, and the process described in Figure 3 is only applicable to short tasks.

2 Short job optimization implementation of resource reuse

According to the design of the short job computing framework, it should be implemented separately on the application manager side and the task process. The implementation of the application manager is based on the statistical data collected by the task process. This section first introduces the implementation of the heartbeat message of the task process, and then introduces the implementation of the task performance model and the second-level task scheduler on the application manager.

2.1 Task process heartbeat message

The communication between the task and the application manager is based on the heartbeat message, and the content of the heartbeat message includes information such as the progress of the task and the health status of the task. The task performance model predicts the completion time of the task based on the statistics of the task running process - the number of processed inputs, the amount of output data, the task output rate, the rate of reading input data, the rate of map operation and the overflow of output data speed and other information. Because the sending frequency of normal heartbeat messages is relatively high, in order to alleviate the pressure of the application manager, the present invention adds a statistical heartbeat message between the task and the application manager, responsible for sending statistical data to the application manager. Statistical heartbeat messages are sent only when the progress of a task exceeds a set threshold.

Algorithm 1. sendHeartbeat algorithm.

Input: the minimum task progress for sending statistical heartbeat messages, set by the user;

the progress of the current task;

Output: heartbeat message

Step 101: If the task progress exceeds the set threshold, calculate the statistical data of the current task (the number of tasks processed, the elapsed running time, the amount of output data, etc.), send a statistical heartbeat message, and go to step 103; otherwise, go to step 102;

Step 102: The progress of the task does not exceed the set threshold, and a task health heartbeat message is sent;

Step 103: the task receives a heartbeat message from the application manager;

Step 104: If the task process receives a new task, go to step 105; otherwise, go to step 106;

Step 105: Read the input data of the new task to the current node. After the current task is executed, run the newly received task;

Step 106: If the task process does not receive a new task, the task process releases the resources occupied by the task after the execution of the current task is completed.

Algorithm 1 describes the process of the task process sending heartbeat messages, curTask is the currently running task, and newTask is the newly received task. When the progress of the task exceeds the set threshold, calculate the number of tasks processed, the amount of output data, the read and write rate and other statistical data, and then send a statistical heartbeat message to the application manager. When the task process receives the heartbeat message fed back by the application manager, it checks whether it has received a new task. If the task process receives a new task, it reads the input data of the new task to the current node in advance. The operation of reading the input data of the new task is executed in parallel with the running task to avoid reading data when the new task is executed. After the running task completes, continue with the newly received task. If the task process does not receive a new task, the running task will release the occupied resources after running.

2.2 Task Completion Time Prediction Model

If the running process of the task can be divided into multiple sub-phases, the completion time of the task is closely related to the time consumed by the task in the sub-phases, so the present invention establishes a task performance model according to the time consumed by the task in the sub-phases. The running process of the task can be divided into multiple sub-phases. If there is no overlap between the sub-phases, the completion time of the task is the sum of the time of each sub-phase; if there is overlap between the sub-phases, the overlapping part should be removed. If the running process of task i is decomposed into n sub-stages, the amount of data processed by each sub-stage is represented by vector s, s=[s ₁ , s ₁ ,...,s _n ], and the rate of data processed by each sub-stage is represented by vector r , r=[r ₁ , r ₁ , . . . , r _n ]. The completion time T _i of task i is:

${\mathrm{<span\; class="notranslate">\; T</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}<span\; class="notranslate">\; =</span>\underset{\mathrm{<span\; class="notranslate">\; i</span>}<span\; class="notranslate">\; =</span>\mathrm{<span\; class="notranslate">\; 1</span>}}{\overset{\mathrm{<span\; class="notranslate">\; no</span>}}{<span\; class="notranslate">\; \&Sigma;</span>}}\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}}<span\; class="notranslate">\; +</span>\mathrm{<span\; class="notranslate">\; \&alpha;</span>}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 1</span>}<span\; class="notranslate">\; )</span>$

Among them, α is the time to start the task, and this value changes within a fixed range, which can be regarded as a constant.

The running process of the Map task is decomposed into four sub-phases, namely, the phase of reading input data, the phase of running map operations, the phase of overflowing output data, and the phase of merging intermediate result files. Each phase is represented by read, map, spill, and combine. Calculate the completion time T _i of Map task i according to formula (1):

${\mathrm{<span\; class="notranslate">\; T</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}<span\; class="notranslate">\; =</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; r</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; d</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; r</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; d</span>}}}<span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; p</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; p</span>}}}<span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}}<span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; c</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; b</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; c</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; b</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}}<span\; class="notranslate">\; +</span>\mathrm{<span\; class="notranslate">\; \&alpha;</span>}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 2</span>}<span\; class="notranslate">\; )</span>$

For short tasks, the running time of the intermediate result file merging stage is very small and can be regarded as a constant. The number of processes processed in the phase of reading input data and the phase of map operation is the same, which is the amount of input data s task of the _task . Since there is a partial overlap between the map operation phase and the output data overflow phase, the amount of data in the overflow phase is the amount output after the map operation, that is, the amount of the last overflow. Assume the amount of input data processed, For the amount of output data, the cache size set by the s _buffer configuration item, is the amount of data written in the last overflow, formula (2) evolves into:

${\mathrm{<span\; class="notranslate">\; T</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}}<span\; class="notranslate">\; =</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; k</span>}<span\; class="notranslate">\; (</span>}\frac{\mathrm{<span\; class="notranslate">\; 1</span>}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; r</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; d</span>}}}<span\; class="notranslate">\; +</span>\frac{\mathrm{<span\; class="notranslate">\; 1</span>}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; m</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; p</span>}}}<span\; class="notranslate">\; )</span><span\; class="notranslate">\; +</span>\frac{{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}^{\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; t</span>}}}{{\mathrm{<span\; class="notranslate">\; r</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}}<span\; class="notranslate">\; +</span>\mathrm{<span\; class="notranslate">\; \&beta;</span>}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 3</span>}<span\; class="notranslate">\; )</span>$

${\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; l</span>}}^{\mathrm{<span\; class="notranslate">\; l</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; t</span>}}<span\; class="notranslate">\; =</span><span\; class="notranslate">\; (</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; a</span>}\mathrm{<span\; class="notranslate">\; the\; s</span>}\mathrm{<span\; class="notranslate">\; k</span>}}<span\; class="notranslate">\; *</span>{\mathrm{<span\; class="notranslate">\; ratio</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}<span\; class="notranslate">\; )</span>{\mathrm{<span\; class="notranslate">\; \%s</span>}}_{\mathrm{<span\; class="notranslate">\; b</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; f</span>}\mathrm{<span\; class="notranslate">\; f</span>}\mathrm{<span\; class="notranslate">\; e</span>}\mathrm{<span\; class="notranslate">\; r</span>}}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 4</span>}<span\; class="notranslate">\; )</span>$

${\mathrm{<span\; class="notranslate">\; ratio</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}<span\; class="notranslate">\; =</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}^{\mathrm{<span\; class="notranslate">\; d</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}<span\; class="notranslate">\; /</span>{\mathrm{<span\; class="notranslate">\; the\; s</span>}}_{\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; p</span>}\mathrm{<span\; class="notranslate">\; u</span>}\mathrm{<span\; class="notranslate">\; t</span>}}^{\mathrm{<span\; class="notranslate">\; d</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; e</span>}}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 5</span>}<span\; class="notranslate">\; )</span>$

Among them, s _task is the input data size of the task, β is a constant, r _read , r _map , and r _spill are obtained from the statistical heartbeat message sent by the task end. is the amount of data processed in stage i, is the elapsed time in stage i.

When the application manager receives the statistical heartbeat message sent by the task end, it can calculate the completion time of the current task according to formula (3). If task k is running on node w, when predicting the completion time of unscheduled task m running on node w, the processing rate of each sub-stage of task m is $r_i^m=r_i^k,i\&Element;\{read,map,\mathrm{the\; s}pill\},{\mathrm{ratio}}_{output}^m={\mathrm{ratio}}_{output}^k.$

2.3 Secondary Task Scheduler

After receiving the statistical heartbeat message sent by the task, the application manager predicts the running time of the task based on the task performance model. If the task sending the statistical heartbeat message is a short task, select an appropriate task from the unscheduled task queue. This section describes the process of selecting unscheduled tasks, and comprehensively considers three aspects of task locality, cluster heterogeneity and resource fairness when selecting tasks.

Problem 1. Locality of tasks. The running time of a task is the sum of the time to read the input data and the time to calculate the input data, so reducing the time to read the input data can shorten the running time of the task. In the cluster, the IO rate of the local disk is higher than the transmission rate of the network, and the network bandwidth within the same rack is larger than the network bandwidth between racks, so when assigning tasks, try to assign tasks to nodes close to the input data . This strategy can not only reduce the time of data copy, but also relieve the load of the network. Therefore, the secondary task scheduler selects tasks in the order of node-local tasks, rack-local tasks, other rack tasks.

Definition 2. Task transfer operation income. Let the running time of task i on node a be The running time at node b is The benefit of transferring task i from running on node a to running on node b is like is a positive value, indicating that the running time of task i transferred from node a to node b is shortened; The degree of income for task transfer operation, the larger the value, the higher the income. like A negative value indicates that the task running time is prolonged.

Definition 3. The optimal task to run on a certain node. Task i running on node a satisfies (6) or (7), then task i is called the optimal task running on node a.

${\mathrm{<span\; class="notranslate">\; T</span>}}_{\mathrm{<span\; class="notranslate">\; a</span>}}^{\mathrm{<span\; class="notranslate">\; i</span>}}<span\; class="notranslate">\; =</span><span\; class="notranslate">\; =</span>{\mathrm{<span\; class="notranslate">\; T</span>}}_{\mathrm{<span\; class="notranslate">\; m</span>}}^{\mathrm{<span\; class="notranslate">\; i</span>}}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 6</span>}<span\; class="notranslate">\; )</span>$

${\mathrm{<span\; class="notranslate">\; gain</span>}}_{\mathrm{<span\; class="notranslate">\; a</span>}}^{\mathrm{<span\; class="notranslate">\; m</span>}}<span\; class="notranslate">\; <</span>{\mathrm{<span\; class="notranslate">\; gain</span>}}_{\mathrm{<span\; class="notranslate">\; c</span>}\mathrm{<span\; class="notranslate">\; o</span>}\mathrm{<span\; class="notranslate">\; no</span>}\mathrm{<span\; class="notranslate">\; f</span>}\mathrm{<span\; class="notranslate">\; i</span>}\mathrm{<span\; class="notranslate">\; g</span>}}<span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; -</span><span\; class="notranslate">\; (</span>\mathrm{<span\; class="notranslate">\; 7</span>}<span\; class="notranslate">\; )</span>$

in, X is the set of nodes running the current job, and m is the node running task i for the shortest time. (6) Indicates that the running time of the task at node a is the minimum, and (7) indicates that the revenue of task i running on the node with the shortest running time is less than the set revenue threshold.

Algorithm 2. assignTask Algorithm

Input: set of unassigned tasks;

Output: tasks to be scheduled;

Step 201: the application manager receives the statistical heartbeat message sent by the task process, and proceeds to step 202;

Step 202: The task performance evaluator predicts the completion time of the sending heartbeat task, then go to step 203;

Step 203: If the completion time of sending the heartbeat task is less than the set time, calculate the running time of unscheduled tasks on all nodes running the current job, and go to step 204;

Step 204: Add the unscheduled tasks to the node local task queue, the rack task queue and other task queues respectively according to the locality, and go to step 205;

Step 205: sort the node local task queue, rack task queue and other task queues according to the running time of the task; go to step 206

Step 206: According to the local priority of the task, judge whether the unscheduled task belongs to the optimal task. If it is the optimal task, delete it from the unscheduled list and return the task; otherwise, judge the next unscheduled task until all unscheduled tasks are checked. Scheduling tasks.

Problem 2. Heterogeneity of clusters. The heterogeneity of the cluster refers to the inconsistency of the hardware configuration of the cluster nodes. The hardware mainly refers to CPU, memory, and disk. The heterogeneity of the cluster has an important impact on the execution efficiency of tasks. When the same task is executed on a high-performance node compared with a low-performance node, the running time of the former is significantly different from that of the latter. When selecting unscheduled tasks, the secondary task scheduler checks whether the selected task is the optimal task for a certain node. If the selected task is the optimal task, execute the task; otherwise, skip the task and re-select the task to be executed.

Question 3. Fairness of resources. Resource fairness refers to the fair sharing of cluster resources by each job. Since the cluster is heterogeneous, the computing capabilities of each node vary greatly. It is necessary to avoid a job from occupying high-performance or low-performance node resources for a long time. In order to share resources fairly, the maximum resource reuse time is set by the user, and the resource reuse time is the same. is the maximum reuse time of resources, is the average running time of tasks on node x, and the resource reuse times are The average running time of tasks on each node is different, so the times of resource reuse in each node are different, and the times of resource reuse in nodes with high performance are more.

The secondary task scheduler selects unscheduled tasks in two steps. The first step is to divide unscheduled tasks into node-local tasks, rack-local tasks, and other-rack tasks by task locality. Then select tasks according to the order of local priority, and select the task with the smallest running time in each priority priority, see Algorithm 2. The second step is to select the optimal task according to the principle of heterogeneity, see Algorithm 3.

In Algorithm 2, k is the node running task i, The resource reuse times for task i, dask _node is the node local task collection, Task _rack is the rack local task collection, Task _offRack is the other rack task collection, It is the maximum resource reuse time, set by the user. Algorithm 2 describes the process that the application manager selects unscheduled tasks after receiving a heartbeat message. First determine whether the task sending the heartbeat message is a short task, and whether the resource reuse times exceed the maximum value. Then unscheduled tasks are divided into three groups: node local tasks, rack local tasks, and other rack tasks, and finally select tasks according to local priority.

Algorithm 3: selectOptimalTask algorithm.

Input: the set of nodes running the current job;

The minimum profit set by the user;

selected tasks;

the node running the current task;

Output: Whether the selected task is the optimal task.

Step 301: Calculate the running time of the input task on all nodes running the current job, go to step 302;

Step 302: Obtain the node with the shortest running time of the calculation input task, go to step 303;

Step 303: Determine whether the obtained node is the same as the input node, if they are the same, the returned task is the optimal task; otherwise, go to step 304;

Step 304: Calculate the revenue of transferring the task from the input node to the node with the shortest running time. If the revenue exceeds the set threshold, the returned input task is the optimal task; otherwise, the returned input task is not the optimal task.

Algorithm 3 describes whether the selected task is the optimal task for the current node. First, calculate the running time of the selected task on other nodes running the current job, and judge whether the running time of the selected task on the current node is the minimum. If the running time of the selected task on node m is the minimum, calculate the income of the task executed on node m. If the income of the task exceeds the set threshold, the selected task is skipped; otherwise, the selected task is executed on the current node.

3 Experiment and Analysis

The present invention realizes the optimization of short jobs based on Hadoop2.2.0, the version before optimization (Apache Hadoop) is called AH, and the version after optimization is called SJH. We evaluate the optimization effect of short jobs by comparing SJH with AH.

The experimental cluster consists of 1 master node and 8 computing nodes. The main configuration of the nodes is shown in Table 1. Configuration 1 is used for the master node and four computing nodes, and configuration 2 is used for the remaining four computing nodes. The nodes are distributed in two racks, and the nodes are connected by gigabit network. The data block size of HDFS is 64M, and the number of copies of the data block is set to 3. The cluster uses the FAIR scheduler provided by Hadoop. Running the Map task requires 1G memory and 1 CPU resource. The Reduce task and the application manager require 1.5G memory and 1 CPU resource respectively.

Table 1 Cluster node configuration information

The experiment uses two test data sets, the first data set is generated by the randomtextwriter provided by Hadoop, and the second data set is the user's electricity consumption data collected by a provincial electric power terminal collection system. randomtextwriter is used to generate a data set with a set amount of data, and the content of the data set consists of random words. The experiment uses word frequency statistics (wordcount), Hive's single-table conditional query and Terasort as benchmark programs. Hive is a data warehouse built on top of HDFS, and the SQL-like provided by Hive is converted into Hadoop jobs. After Hive's single-table conditional query statement is converted into a Hadoop job, there are only Map tasks. Word frequency statistics are used to count the frequency of each word in the input data set, and Terasort sorts the input data set in lexicographical order.

3.1 Accuracy of task performance model

The present invention first tests the accuracy of the task performance model in predicting running tasks and non-running tasks. Use the word frequency statistics program as the benchmark test program, use dataset 1, the data size is 2.5G, and the cluster uses 40 Map tasks to process the dataset. Experiments use relative error to describe accuracy, e is the size of the error, is the predicted completion time of the task, and T ⁱ is the actual completion time of the task. When the progress p _min of the task is 40%, 60%, 70%, 80%, respectively, the error of predicting the completion time of the running task.

Figure 4(a) depicts how the task performance model predicts the error variation of a running task when the task's progress takes different values. It can be seen from Figure 4(a) that the error decreases with the increase of progress, and the error tends to be stable. When the progress of the task exceeds 60%, the error is within 20%; when the progress of the task exceeds 70%, the error is about 10%. Figure 4(b) depicts the relationship between task running time and error. The error increases with the running time of the task, and when the completion time of the task reaches 80 seconds, the error exceeds 20%. And the error in predicting non-running tasks is higher than the error in predicting running tasks. The task performance model uses the read rate, map operation processing rate, and overflow write rate in a short period of time to calculate the completion time of the task, without considering the change of the rate over time. Therefore, the prediction error of the task performance model will increase with the increase of task time. From Figure 4(b), we know that the prediction error is less than 15% when the task completion time is less than 30 seconds, and the average completion time of short tasks is 20 seconds, so the prediction model meets the actual needs.

3.2 Job Completion Time

We test the short-job optimization effect when short-jobs and long-jobs are mixed, and the test data are shown in Table 2. The experiment uses a shell script to poll and submit the benchmark test program. One job is submitted per second, and all jobs are submitted within 30 seconds. The total number of containers provided by the cluster is 96, so the Map task of each job needs multiple rounds to be scheduled and completed.

Figure 5(a) and Figure 5(b) show the impact of short job optimization on job completion time. The running time of the word frequency statistics job is shortened by 6%-22%, and the running time of the HiveSQL conversion job is shortened by 6%-27%. These two jobs are short jobs. Hive single-table conditional query only has Map tasks, and its optimization effect is better than word frequency statistics jobs. From Figure 5(c), it can be seen that short job optimization has no significant impact on long jobs, and the maximum lengthening time of jobs is 5%. Short-task optimization makes short-job tasks seize resources, resulting in a partial extension of long-job completion time.

Table 2 job test data

3.3 Resource utilization

This section focuses on the CPU and memory utilization of computing nodes, and the benchmark programs and data sets are shown in Table 2. The resource utilization is collected using Ganglia, and the final result is the average value of each resource during the calculation process. Figure 6 shows the CPU utilization of computing nodes, and the optimized CPU utilization increases by 4%-13% on average. In the benchmark test program, word frequency statistics and HiveSQL conversion tasks are computationally intensive. The increase in CPU utilization of nodes N1, N2, N6, and N7 is greater than that of N3, N4, N5, and N8, because the former are nodes with high performance, and the latter are nodes with low performance. Figure 7 describes the change of computing node memory, and the utilization rate of optimized memory is increased by 2.6%-6.18%. The increase in memory utilization is not large, because only terasort's output data is relatively large in the benchmark program. The experimental test results show that the optimized short job method proposed by the present invention has a significant effect in improving the effective utilization of cluster resources.

4 Summary

Since Hadoop does not consider the characteristics of short jobs, the execution of short jobs in Hadoop is relatively inefficient. In view of the above challenges, the present invention first describes the problems existing in the short job processing process by analyzing the job execution process in Hadoop. Then, according to the characteristics of tasks running for multiple rounds under high load conditions, a short job optimization mechanism based on resource reuse is proposed to reduce the waste of resources in the process of allocation and recycling by reusing the resources released by the executed tasks. The present invention reduces the running time of short jobs by optimizing the Map task execution process. Future work will improve the execution efficiency of short jobs from the aspects of Reduce task execution process and task scheduling.

Although the specific implementation of the present invention has been described above in conjunction with the accompanying drawings, it does not limit the protection scope of the present invention. Those skilled in the art should understand that on the basis of the technical solution of the present invention, those skilled in the art do not need to pay creative work Various modifications or variations that can be made are still within the protection scope of the present invention.

Claims (10)

1. A MapReduce short operation optimization system based on resource reuse is characterized by comprising: the system comprises a main node, a primary slave node and a plurality of secondary slave nodes, wherein the main node is connected with the primary slave node, and the primary slave node is connected with the plurality of secondary slave nodes; a resource manager and a primary task scheduler are deployed on the main node; the primary slave node is provided with an application manager, a task performance evaluator and a secondary task scheduler, wherein the secondary task scheduler is connected with the task performance evaluator and is also connected with the primary node; each secondary slave node is provided with a node manager;

the task performance evaluator predicts the completion time of the running task and the unscheduled task;

and the secondary task scheduler judges whether the task being executed belongs to a short task or not according to the prediction result of the task performance evaluator and selects the task from the unscheduled task queue.

2. The MapReduce short job optimization system based on resource reuse as recited in claim 1,

if the task being executed is a short task, the secondary task scheduler selects a new short task from the unscheduled task queue, and the new short task reuses the resources to be released by the short task being executed; and if the task being executed is not the short task, directly releasing occupied resources after the task being executed is executed.

3. The MapReduce short job optimization system based on resource reuse as recited in claim 1,

when the secondary task scheduler selects the unscheduled task, the locality of the task, the running time of the task, the fairness of the resource and the heterogeneity of the cluster need to be considered.

4. The MapReduce short job optimization system based on resource reuse as recited in claim 1,

the resource manager is responsible for the distribution and monitoring of global resources and the starting and monitoring of the application manager;

the application manager decomposes the operation into a Map task and a Reduce task, applies resources for the Map task and the Reduce task, operates in cooperation with the node manager and is also used for monitoring the tasks; the application manager is a control unit for operation of the jobs, and each job corresponds to one application manager;

the node manager is responsible for monitoring the resource amount used by the task and preventing the resource amount used in the task running process from exceeding the resource amount applied by the task.

5. A MapReduce short operation optimization method based on resource reuse is characterized by comprising the following steps:

step (1): the application manager applies for resources from the resource manager through the heartbeat message;

step (2): the resource manager allocates idle resources to the application manager applying for the resources, and the application manager acquires the applied resources;

and (3): the application manager distributes the applied resources to the unscheduled tasks, and then informs the corresponding node manager to start a task process;

and (4): the node manager starts a task process to run a task, and the task sends heartbeat messages to the application manager at intervals of a set heartbeat period in the running process;

and (5): the application manager receives the heartbeat message, predicts the running time of the task based on a task completion time prediction model in the task performance evaluator, and if the running time of the task is less than or equal to the task completion time set by the user, the current task is a short task; otherwise, the current task is a long task; if the current task is a short task, the secondary task scheduler selects a new task from an unscheduled task queue; if the current task is a long task, the heartbeat message is ignored;

and (6): the application manager informs the task process of the selected new task, and the task process continues to run the new task after finishing executing the running task;

and (7): and if the task process does not receive a new task after executing the running task, the task process exits and releases the occupied resources, and the node manager informs the released resources to the resource manager through heartbeat messages.

6. The method as set forth in claim 5, wherein,

the step (4) of sending heartbeat messages to the application manager at intervals of a set heartbeat cycle in the running process of the task is as follows:

a step (401): judging whether the task progress exceeds a set threshold, if so, calculating statistical data of the current task, sending a statistical heartbeat message, and turning to the step (403); otherwise go to step (402); the statistical data of the current task comprises the processed quantity, the running time and the output data volume of the task;

step (402): sending a healthy heartbeat message of the task if the progress of the task does not exceed a set threshold; entering a step (404);

step (403): a task receives a heartbeat message of an application manager; entering a step (404);

a step (404): judging whether the task process receives a new task, and if the task process receives the new task, turning to the step (405); otherwise go to (406);

step (405): reading input data of the new task to a current node; after the current task is executed, running a newly received task;

step (406): and if the task process does not receive the new task, the task process releases the resources occupied by the task after the current task is executed.

7. The method as recited in claim 5, wherein said step (5) of the secondary task scheduler selecting a new task from the unscheduled task queue comprises:

step (51):

dividing the unscheduled tasks into node local tasks, rack local tasks and other rack tasks according to the locality of the tasks;

selecting tasks according to the sequence of the node local tasks, the rack local tasks and other rack tasks, and preferentially selecting the task with the minimum running time in each priority level;

step (52): and selecting an optimal task according to the isomerism principle.

8. The method as set forth in claim 7, wherein,

the step (51) comprises the following steps:

step (511): the application manager receives the statistical heartbeat message sent by the task process, and then the step (512) is carried out;

step (512): the task performance evaluator predicts the completion time of sending the heartbeat task, and then the step (513) is carried out;

step (513): if the completion time of sending the heartbeat task is less than the set time, calculating the running time of the unscheduled task on all nodes running the current operation, and turning to the step (514);

step (514): adding unscheduled tasks into the node local task queue, the rack local task queue and other rack task queues according to locality, and turning to step (515); the locality refers to whether input data of a task is the same as a processing node or not and whether the input data is in a rack or not;

step (515): sequencing the node local task queue, the rack local task queue and other rack task queues according to the running time of the tasks; turning to step (516);

step (516): judging whether the unscheduled task belongs to the optimal task or not according to the local priority of the task, if so, deleting the currently selected unscheduled task from the unscheduled list, and returning to the task; otherwise, judging the next unscheduled task until all unscheduled tasks are checked.

9. The method as set forth in claim 7, wherein,

the step (52) comprises the following steps:

step (521): calculating the running time of the input task on the whole node running the current job, and turning to the step (522);

step (522): acquiring a node with the shortest running time of the calculation input task, and turning to the step (523);

step (523): judging whether the acquired node is the same as the input node or not, if so, judging that the returned task is the optimal task; otherwise go to step (524);

step (524): calculating the running profit of the task transferred from the input node to the node with the shortest running task time, and if the profit exceeds a set threshold value, returning that the input task is the optimal task; otherwise, the return input task is not the optimal task.

10. The method as set forth in claim 7, wherein,

if the input data of the task is on the node for processing the task, the task is called as a node local task;

if the input data of the task and the nodes for processing the task are in the same frame, the task is called a frame local task;

if the input data of a task is not in the same rack as the node processing the task, the task is called other rack task.