CN114490027A - Distributed job adjustment method, master node, system, physical machine and storage medium - Google Patents

Abstract

The embodiment of the application provides a distributed job adjustment method, a main node, a system, a physical machine and a storage medium, wherein the method comprises the following steps: acquiring a job submitted by a user; generating an execution plan of the job, wherein a plurality of execution stages of the execution plan comprise an upstream execution stage and a direct downstream execution stage; in the process of executing the operation, acquiring statistical information of output data in an upstream execution stage; configuring the direct downstream execution stage according to the statistical information. The embodiment of the application can dynamically adjust the configuration of the downstream execution stage based on the output result of the upstream execution stage, for example, configure the concurrency of the downstream execution stage, allocate a data partition, select a subsequent execution path, and the like. Furthermore, the embodiment of the application can optimize the description of the execution plan in the deep learning field and reasonably optimize the resource allocation. The method and the device can improve the performance of the distributed system.

Description

Distributed job adjustment method, master node, system, physical machine and storage medium

This application is a divisional application of an application with an application number of 202110950182.9, an application date of August 18, 2021, and an invention title of "distributed job adjustment method, master node, system, physical machine and storage medium".

technical field

The embodiments of the present application relate to the field of distributed technologies, and in particular, to a distributed job adjustment method, a master node, a system, a physical machine, and a storage medium.

Background technique

A distributed system is a system composed of multiple physical machines interconnected through communication lines, and has the characteristics of distribution, autonomy, parallelism, and globality. Using a distributed system to execute jobs submitted by users can improve the efficiency of job execution through the distributed computing capabilities of the distributed system.

A distributed system mainly includes a master node and a worker node. Jobs submitted by users can be executed by the master node. The master node can configure the generated execution plan. Based on the configuration of the execution plan, the master node can schedule worker nodes and resources to implement the specific execution of the job during the execution of the job. Based on the wide application of distributed systems, those skilled in the art have been working to improve the performance of distributed systems.

SUMMARY OF THE INVENTION

In view of this, the embodiments of the present application provide a distributed job adjustment method, a master node, a system, a physical machine, and a storage medium, so as to accurately and reasonably dynamically configure an execution plan during job execution, so as to realize dynamic adjustment of job configuration , thereby improving the performance of the distributed system.

To achieve the above purpose, the embodiments of the present application provide the following technical solutions.

In a first aspect, an embodiment of the present application provides a distributed job adjustment method, including:

Get jobs submitted by users;

generating an execution plan for the job, the execution plan including a plurality of execution stages, the plurality of execution stages including an upstream execution stage and an execution stage directly downstream of the upstream execution stage;

During the execution of the job, obtain the statistical information of the output data of the upstream execution stage;

According to the statistical information, the direct downstream execution stage is configured so that the direct downstream execution stage executes the job based on the configuration result.

In a second aspect, an embodiment of the present application provides a distributed job adjustment method, including:

Get deep learning assignments;

An execution plan for the deep learning job is generated, the execution plan includes multiple execution stages, and the multiple execution stages include: a worker execution stage and a resource optimization execution stage; the worker execution stage is used to calculate deep learning parameters the gradient of ;

During the execution of the deep learning job, schedule resource optimization nodes corresponding to the resource optimization execution stage, and determine historically used resource information that matches the current execution state of the deep learning job through the resource optimization nodes;

The resource information is configured for the worker execution phase through a resource optimization node.

In a third aspect, an embodiment of the present application provides a master node, where the master node is configured to execute the distributed job adjustment method described in the first aspect or the second aspect.

In a fourth aspect, an embodiment of the present application provides a distributed system, where the distributed system includes a master node and a plurality of worker nodes, where the master node is the master node described in the third aspect.

In a fifth aspect, an embodiment of the present application provides a physical machine, where the physical machine includes at least one memory and at least one processor, the memory stores one or more computer-executable instructions, and the processor calls the one or more A plurality of computer-executable instructions are used to execute the distributed job adjustment method according to the first aspect or the second aspect.

In a sixth aspect, an embodiment of the present application provides a storage medium, where the storage medium stores one or more computer-executable instructions, and when the one or more computer-executable instructions are executed, the above-mentioned first aspect or the second The distributed job adjustment method described in the aspect.

In the distributed job adjustment method provided by the embodiment of the present application, after the master node obtains the job submitted by the user, it can generate an execution plan of the job, the execution plan includes multiple execution stages, and the multiple execution stages include an upstream execution stage and The immediate downstream execution stage of the upstream execution stage. During the job execution process, the master node can obtain the statistical information of the output data of the upstream execution stage, so as to configure the direct downstream execution stage according to the statistical information, so that the direct downstream execution stage executes the job based on the configuration result. Because the embodiments of the present application can dynamically adjust the configuration of the downstream execution stage based on the data output result of the upstream execution stage during the job execution process, so that the configuration of the downstream execution stage can be adapted to the actual execution result of the upstream execution stage, so that the downstream execution stage can be adapted to the actual execution result of the upstream execution stage. The configuration of concurrency and resources in the execution phase can match the specific execution situation of the job, and improve the rationality and accuracy of the configuration of the execution plan. It can be seen that the distributed job adjustment method provided by the embodiment of the present application can dynamically adjust the configuration of the execution plan during the execution of the job, realize the dynamic adjustment effect of the job, and make the configuration of the execution plan fit the specific execution situation of the job. Furthermore, the specific execution of the job is realized based on the dynamically configured execution plan, which can make the distributed system complete the job execution more reasonably and efficiently, and significantly improve the performance of the distributed system to execute the job.

Description of drawings

In order to more clearly illustrate the embodiments of the present application or the technical solutions in the prior art, the following briefly introduces the accompanying drawings required for the description of the embodiments or the prior art. Obviously, the drawings in the following description are only It is an embodiment of the present application. For those of ordinary skill in the art, other drawings can also be obtained according to the provided drawings without any creative effort.

FIG. 1A is a schematic structural diagram of a distributed system.

FIG. 1B is a schematic diagram of an execution plan.

Figure 1C is a schematic diagram of a DAG.

FIG. 1D is a schematic diagram of mapping between a logical map and a physical map.

FIG. 2 is a flowchart of a distributed job adjustment method provided by an embodiment of the present application.

FIG. 3A is a schematic diagram of processing data distribution in the directly downstream stage.

FIG. 3B is another flowchart of the distributed job adjustment method provided by the embodiment of the present application.

Figure 3C is an example diagram of Partitions assigned to directly downstream stages.

FIG. 4A is an example diagram of data shuffling.

FIG. 4B is still another flowchart of the distributed job adjustment method provided by the embodiment of the present application.

FIG. 4C is an example diagram of partition splitting according to an embodiment of the present application.

FIG. 5A is an example diagram of a Join process.

FIG. 5B is another example diagram of the Join process.

FIG. 5C is another flowchart of the distributed job adjustment method provided by the embodiment of the present application.

FIG. 5D is another example diagram of the Join process.

Figure 5E shows yet another example diagram of the Join process.

FIG. 5F is an example diagram of a union operation further shown on the basis of FIG. 5D .

FIG. 5G is an example diagram of a union operation further shown on the basis of FIG. 5E .

FIG. 6A is an example diagram of Sort Merge Join.

FIG. 6B is an example diagram of Broadcast Join.

FIG. 6C is still another flowchart of the distributed job adjustment method provided by the embodiment of the present application.

FIG. 6D is a schematic diagram of an execution plan carrying multiple execution paths.

FIG. 7A is a flowchart of generating an execution plan with multiple execution paths.

Figures 7B, 7C, 7D and 7E illustrate example diagrams of a process of converting a physical plan into an execution plan.

FIG. 7F is an example diagram of a complete execution plan after selecting an execution path.

FIG. 7G is another example diagram of a complete execution plan after an execution path is selected.

FIG. 8A is an example diagram of a PS stage and a Worker stage connected in parallel.

FIG. 8B is still another flowchart of the distributed job adjustment method provided by the embodiment of the present application.

FIG. 8C is a diagram showing an example of resource adjustment of the Worker node by the Resource Optimization node.

FIG. 9 is a structural block diagram of a physical machine.

Detailed ways

The technical solutions in the embodiments of the present application will be clearly and completely described below with reference to the drawings in the embodiments of the present application. Obviously, the described embodiments are only a part of the embodiments of the present application, but not all of the embodiments. Based on the embodiments in the present application, all other embodiments obtained by those of ordinary skill in the art without creative efforts shall fall within the protection scope of the present application.

A distributed job can be understood as submitting a job to a distributed system for execution. FIG. 1A exemplarily shows a schematic structural diagram of a distributed system. As shown in FIG. 1A , the distributed system may include: a master node 110 and a plurality of worker nodes 120 . The master node 110 and the worker node 120 may be regarded as computing nodes in a distributed system, the computing nodes may be hosted on physical machines with data computing capabilities, and one physical machine may host one or more computing nodes.

In a distributed system, the master node 110 is a computing node for management and control. For example, the master node 110 can manage the worker nodes 120, coordinate the corresponding concurrency and resources of the job in each execution stage of the execution plan, and so on. In some aspects, the master node 110 acts as a central management and control node in the distributed system, also referred to as the execution engine of the distributed system. The worker node 120 is a computing node that specifically executes a job in a distributed system, and can be managed and coordinated by the master node 110 to execute the job.

When the distributed system executes the job, the job can be submitted by the user to the cluster resource manager through the terminal, and the cluster resource manager pulls up the master node 110 . Therefore, the master node 110 can analyze the job and generate an execution plan. The execution plan describes the process of the job data from the original source table, after a series of data flow, execution, and changes, to finally produce output. FIG. 1B exemplarily shows a schematic diagram of an execution plan. As shown in FIG. 1B , the execution plan may include: a plurality of stages (execution stages) having a hierarchical relationship. In some embodiments, there may be a tree-like hierarchy between stages. A stage can contain one or more tasks. For each stage, the master node 110 can schedule multiple worker nodes to execute tasks of the stage in parallel by configuring the number of worker nodes (concurrency), used resources, etc., so as to implement job execution in the distributed system.

In some embodiments, jobs are typically submitted by terminals to the distributed system on a request basis. In one example, the job submitted by the terminal includes a query statement for querying a database, such as a SQL (Structured QueryLanguage, structured query language) statement.

In some further embodiments, the execution plan can be described by DAG (Directed Acyclic Graph). A DAG includes a plurality of vertices and connecting edges between the vertices. Figure 1C exemplarily shows a schematic diagram of a DAG. It should be noted that the actual number of vertices, levels, and connecting edges of the DAG may be more complicated than those in FIG. 1C , which is only a simple DAG example for the purpose of easy understanding. As shown in FIG. 1C , the DAG may include four vertices V1 to V4 , and connecting edges 11 , 12 , 13 and 14 . The connecting edge 11 connects the vertices V1 and V2, the connecting edge 12 connects the vertices V1 and V3, the connecting edge 13 connects the vertices V2 and V4, and the connecting edge 14 connects the vertices V3 and V4.

A vertex in a DAG can represent a separate stage in the execution plan. Connecting edges between vertices can be directed connecting edges, which represent relationships between vertices. Based on the relationship that the connecting edges point to, the connecting edges connected by vertices may be input connecting edges of vertices (input connecting edges point to vertices), or may be output connecting edges of vertices (output connecting edges point from vertices to other vertices). For example, in Figure 1C, the connecting edge 12 points to V3, which is the input connecting edge of V3; the connecting edge 14 is output by V3 and is the output connecting edge of V3; and the connecting edge 12 is output by V1, so the connecting edge 12 is also used as the output of V1. Connection edge; connection edge 14 inputs V4, so connection edge 14 in turn acts as an input connection edge for V4.

Among the two vertices connected by a connecting edge, the vertex that connects the output of the connecting edge is called the immediate upstream vertex of the other vertex, and the vertex that connects the input of the connecting edge is called the immediate downstream vertex of the other vertex. For example, in FIG. 1C, connecting edge 12 connects V1 and V3, V1 outputs connecting edge 12 and connecting edge 12 inputs V3, then V1 can be referred to as the immediate upstream vertex of V3, and V3 can be referred to as the immediate downstream vertex of V1. A vertex may have one or more directly upstream vertices and one or more directly downstream vertices. It should be noted that, in addition to the direct upstream vertex, a vertex may also have an indirect upstream vertex. The indirect upstream vertex is not directly connected to the vertex, but is located at the upper level of the vertex and is connected to the vertex through one or more vertices. vertices are connected. For example, in Fig. 1C, V1 is in the upper layer of V4, and V1 is connected to V4 through V2 or V3, so V1 can be called the indirect upstream vertex of V4. Obviously, in addition to the direct downstream vertex, a vertex may also have indirect downstream vertex. The indirect downstream vertex is not directly connected to the vertex, but is located in the lower layer of the vertex and is connected to the vertex through one or more vertices. connect. For example, in Fig. 1C, V4 is in the lower layer of V1, and is connected to V1 through V2 or V3, so V4 can be called an indirect downstream vertex of V1. An upstream vertex of a vertex may include a direct upstream vertex and an indirect downstream vertex, and a downstream vertex of a vertex may include a direct downstream vertex and an indirect downstream vertex.

The execution of a vertex may depend on the immediate upstream vertex, that is, there is an execution dependency between the vertex and the immediate upstream vertex, and the vertex needs to be executed after the immediate upstream vertex is executed; Vertices are executed in parallel.

In further embodiments, the DAG may have two levels of representation: a logical graph and a physical graph. The logic diagram can be considered as a natural extension of the execution plan, which describes the data execution flow that the user wants to implement for the job. The physical map reflects the physical attributes of the execution plan mapped to the distributed system by each stage, and describes the physical attributes such as the concurrency of each stage of the execution plan at the execution level, the resources used by the worker nodes, and the data transmission method.

FIG. 1D exemplarily shows a schematic diagram of a mapping between a logical map and a physical map. For the convenience of illustration, FIG. 1D only illustrates that the execution plan has 4 stages. As shown in Figure 1D, the logic diagram describes the 4 vertices (vertices V0, V1, V2, and V3) of the execution plan and the relationship of each vertex (for example, vertex V0 points to vertex V2, and vertex V1 and vertex V2 point to vertex V3), A vertex corresponds to a stage of the execution plan. The logic diagram can reflect the data execution flow of the execution plan. After the logical graph is mapped to a physical graph, the physical graph can describe the concurrency that needs to be configured for each stage, the resources used by the worker nodes of each stage (such as CPU resources, memory resources, etc.), data transmission methods, and other physical attributes. For example, combined with the example in Figure 1D, the physical diagram describes that vertex V0 needs to be configured with 3 worker nodes (with a concurrency of 3), and vertices V1, V2, and V3 need to be configured with 2 worker nodes (with a concurrency of 2). That is, a physical graph can express the physical properties of vertices and connecting edges in a DAG. Through the physical attributes of the vertices and connecting edges described by the physical graph, the master node can schedule worker nodes and resources for each stage, so that tasks in the stage can be executed in parallel by multiple worker nodes, realizing the execution of jobs in a distributed system.

The configuration execution plan referred to in the embodiment of the present application may include the logic of the configuration execution plan and the physical properties of the configuration execution plan. The logic of configuring the execution plan can be considered as configuring the execution plan at the logical layer, for example, configuring the execution flow of the execution plan. Configuring the physical properties of the execution plan can be considered as configuring the execution plan at the physical layer, for example, configuring the physical properties such as the concurrency, resources, and data transmission methods of each stage of the execution plan.

If the execution plan is configured before the specific execution of the job, and after the configuration of the execution plan cannot be adjusted during the specific execution of the job, such an execution plan is called a static execution plan. That is, a static execution plan is configured before job execution and cannot be adjusted during job execution. However, it is impossible to accurately estimate the actual resources required by each stage of the execution plan and the reasonable execution path of the execution plan before the specific execution of the job, which undoubtedly makes the configuration of the static execution plan difficult to be reasonable and accurate. performance.

Based on this, the embodiments of the present application provide a new execution plan configuration solution, which can dynamically adjust the configuration of the downstream stage based on the data output results of the upstream stage during the job execution process, so that the configuration of the downstream stage can be adapted to the upstream stage. Therefore, the configuration of the concurrency and resources of the downstream stage can match the specific execution situation of the job, and improve the rationality and accuracy of the configuration of the execution plan.

As an optional implementation, FIG. 2 exemplarily shows a flowchart of the distributed job adjustment method provided by the embodiment of the present application. The method flow may be implemented by the master node. Referring to FIG. 2 , the method flow may include the following steps.

In step S210, the job submitted by the user is acquired.

In step S211, an execution plan of the job is generated, and the execution plan includes a plurality of stages, and the plurality of stages include an upstream stage and a stage directly downstream of the upstream stage.

After the terminal submits the job to the distributed system, the master node in the distributed system can parse the job and generate an execution plan of the job, and the execution plan can be described by the DAG. The execution plan may include multiple stages, and the multiple stages may include an upstream stage and a stage directly downstream of the upstream stage. In some embodiments, any stage with a downstream stage in the plurality of stages can be regarded as the upstream stage, and during the specific execution process of the job, the output data of the upstream stage can be input directly to the downstream stage for processing. The output data of an upstream stage can be input to one or more directly downstream stages, and a direct downstream stage may also input the output data of one or more upstream stages; that is, an upstream stage can correspond to one or more directly downstream stages, a A directly downstream stage can also correspond to one or more upstream stages. In some embodiments, the upstream stage may be referred to as the previous stage of the directly downstream stage, and the directly downstream stage may be referred to as the next stage of the upstream stage.

In an example, as shown in Figure 1C, V1 is the upstream stage of V2 and V3, and V2 and V3 are the direct downstream stages of V1. During the execution of the job, the output data of V1 can be input to V2 and V3 for processing; and V2 and V3 also serve as the upstream stage of V4, V4 serves as the direct downstream stage of V2 and V3, and the output data of V2 and V3 can be input to V4 for processing. One aspect that the embodiments of this application focus on is how much concurrency, resources, etc., need to be configured for the direct downstream stage to process the output data of the upstream stage; for example, how much concurrency, resources, etc. need to be configured for V2 and V3 respectively to process the output data of V1, How much concurrency and resources need to be configured for V4 to process the output data of V2 and V3.

In step S212, during the execution of the job, the Statistics (statistical information) of the output data of the upstream stage is acquired.

During job execution, after the execution of any stage is completed, the master node can collect the Statistics of the output data of the worker nodes of the stage. Based on this, after an upstream stage of the execution plan is executed, the master node can collect the statistics of the output data of the upstream stage. In some embodiments, the Statistics of the output data of the upstream stage may include any of the following: the data volume of the output data (for example, the data volume of the output data before and after data compression respectively), the Partition of the output data (data partition) The data volume distribution information, the number of Records (serialized data records) of each Partition in the output data, etc.

In step S213, the directly downstream stage is configured according to the Statistics, so that the directly downstream stage executes a job based on the configuration result.

Based on the statistics of the output data of the upstream stage, the master node can configure the direct downstream stage, so that the configuration of the direct downstream stage can be adapted to the actual execution result of the upstream stage, so that the concurrency, resources and other configurations of the downstream stage can be posted. The specific execution situation of the cooperative business; further, the direct downstream stage can execute the job based on the configuration result, so that the corresponding task of the direct downstream stage can be executed reasonably and efficiently.

During job execution, in the manner provided by the embodiments of the present application, the directly downstream stages of the upstream stage are dynamically configured, so that the execution plan can be dynamically adjusted during the job execution process, and the configuration of the execution plan can be adapted to the job's configuration. specific implementation. That is, during the job execution process, the embodiments of the present application can dynamically adjust the distributed job based on the output data and statistical information of the upstream stage, so as to realize the dynamic adjustment effect of the job. Further, the specific execution of the job is realized based on the dynamically configured execution plan, which enables the distributed system to complete the job execution more reasonably and efficiently, and significantly improves the performance of the distributed system to execute the job.

In some embodiments, based on the statistics of the output data of the upstream stage, configuring the directly downstream stage may include any of the following:

According to the statistical information, configure the degree of concurrency for the direct downstream stage;

Assign a Partition to the directly downstream stage according to the statistical information; for example, assign the Partition output by the upstream stage to the worker node of the directly downstream stage, or, in the Join (connection) scenario, split the Partition output by the upstream stage and assign it Give the direct downstream stage to perform the Join operation on the worker node of the direct downstream stage;

According to the statistical information, the direct downstream stage for subsequent execution is selected; in the execution plan, the downstream of the upstream stage can be configured with multiple candidate execution paths, so that during the specific execution of the job, based on the execution result of the upstream stage, from multiple execution paths The actual execution path is selected in the execution path to make the execution logic of the execution plan more accurate and reasonable; in this case, the stages in an execution path can include at least the direct downstream stage of the upstream stage; The selection of multiple execution paths enables dynamic adjustment of execution logic and selection of directly downstream stages for subsequent execution.

For the above-mentioned various situations of configuring the direct downstream stage based on statistical information, the following sections will describe them respectively, and will not be expanded here.

In the distributed job adjustment method provided by the embodiment of the present application, after acquiring the job submitted by the user, the master node can generate an execution plan of the job, the execution plan includes multiple stages, and the multiple stages include an upstream stage and the upstream stage. The stage directly downstream of the stage. During the job execution process, the master node can obtain the statistical information of the output data of the upstream stage, so as to configure the directly downstream stage according to the statistical information. Because the embodiment of the present application can dynamically adjust the configuration of the downstream stage based on the data output result of the upstream stage during the job execution process, so that the configuration of the downstream stage can be adapted to the actual execution result of the upstream stage, so that the concurrency of the downstream stage can be improved. The allocation of resources, etc. can match the specific implementation of the business, and improve the rationality and accuracy of the allocation of the implementation plan. It can be seen that the distributed job adjustment method provided by the embodiment of the present application can dynamically adjust the configuration of the execution plan during the execution of the job, realize the dynamic adjustment effect of the job configuration, and make the configuration of the execution plan fit the specific execution of the job. Then, the specific execution of the job is realized based on the dynamically configured execution plan, which can make the distributed system complete the job execution more reasonably and efficiently, and significantly improve the performance of the distributed system to execute the job.

The following describes the implementation scheme for configuring the concurrency of the direct downstream stage based on the statistical information of the output data of the upstream stage.

For static execution plans, the concurrency of each stage of the execution plan can be determined by the master node through estimation rules. For example, after the job is submitted, the master node can use the estimation rule to configure the concurrency of each stage of the execution plan according to the total amount of source data of the job (which can be regarded as configuring the concurrency of each vertex in the DAG), or, based on the user Specify the concurrency of different types of stages, and configure the concurrency of each stage of the execution plan. However, the source data processed by distributed jobs is complex and diverse, and it is often difficult for the master node to rely on the estimation rules to configure the concurrency of different jobs, which leads to inaccurate configuration of the concurrency of each stage of the execution plan. For example, for a stage that processes a small amount of data, if a large concurrency is statically configured, it will lead to a waste of computing resources in the distributed system; while for a stage that processes a large amount of data, if a small concurrency is statically configured , which will prolong the execution time of the stage, and even bring about various errors such as memory usage exceeding the limit, resulting in the failure of job execution. Therefore, in the implementation of statically configured concurrency, in order to avoid the possibility that the stage is configured with a low concurrency and it is difficult to process massive data, the stage often needs to be configured with a high concurrency, which leads to more computations during the actual execution of the job. Waste of resources. For example, for a Map (mapping)-Reduce (reduction) job, even if the upstream Map stage only generates 1KB of output data during the actual operation, the downstream Reduce stage will still have a higher degree of concurrency due to static configuration. , and a higher number of worker nodes are scheduled to process the 1KB data, which undoubtedly leads to unnecessary waste of computing resources.

Based on this, in the process of job execution, it is particularly necessary to configure the concurrency for the direct downstream stage of the upstream stage based on the output data results of the upstream stage. In some embodiments, for the directly downstream stage of the upstream stage, the master node may allocate the partitions (data partitions) in the output data of the upstream stage to the work nodes of the directly downstream stage according to the principle of the number of partitions being even. For example, each worker node of the directly downstream stage will process multiple consecutive Partitions output by the upstream stage, and each worker node processes the same number of Partitions. This method can be called the Even Reduction strategy based on the number of Partitions, which can achieve ideal results when the data volume of each Partition is uniform; however, in the actual environment, the data distribution characteristics are diverse And it often has uneven characteristics. For non-uniform data (the amount of data in each Partition is not uniform), the above method may lead to a data skew problem in a single worker node of the direct downstream stage, that is, the amount of data processed by a single worker node is much larger than that of other nodes. worker nodes, which further leads to the long tail problem of worker nodes, causing the job execution time to be stretched unnecessarily. For example, FIG. 3A exemplarily shows a schematic diagram of processing data distribution in a directly downstream stage. As shown in Figure 3A, the data volume of each Partition output by the upstream stage is not uniform. The value in the Partition output by the upstream stage can represent the data volume of the corresponding Partition. If you simply merge multiple specified number of Partitions to the direct downstream For a single worker node of the stage, although the number of Partitions processed by each worker node is the same (for example, each worker node of the directly downstream stage in Figure 3A processes 2 Partitions), this may cause the worker nodes of the directly downstream stage to process data in the process. Uneven distribution in terms of volume: On the one hand, the data processed by some worker nodes is unreasonably small (there may even be worker nodes that have no data to process at all); Large Partition, which further exacerbates the data skew problem, causing the running time of the worker nodes to be elongated and forming a long tail.

In addition, since the number of Partitions may be positively related to the time-consuming of the actual calculation as a whole, but for some data types (such as integers), the data compression rate is relatively large, and it is possible that after the output data is compressed, the data file Severely compressed. In this case, a small number of Partitions may correspond to a large number of data records (that is, a large number of data records are recorded in a small number of Partitions), so the direct downstream stage concurrency is simply based on the number of Partitions in the upstream stage. degree configuration, which may bring greater uncertainty in actual operations.

Based on the above situation, the embodiment of the present application provides a solution for dynamically adjusting the concurrency of the direct downstream stage based on the statistical information of the output data of the upstream stage during the job execution process, and guarantees the The amount of data processing tends to be balanced. As an optional implementation, FIG. 3B shows another flowchart of the distributed job adjustment method provided by the embodiment of the present application. The method flow may be implemented by the master node. Referring to FIG. 3B , the method flow may include the following steps.

In step S310, the job submitted by the user is acquired.

In step S311, an execution plan of the job is generated, and the execution plan includes a plurality of stages, and the plurality of stages includes an upstream stage and a stage directly downstream of the upstream stage.

In step S312, the concurrency is initially configured for at least the stage that is executed first among the multiple stages.

The master node can initially configure the concurrency of the execution plan. In some embodiments, the master node can use estimation rules to initially configure the concurrency of the execution plan according to the total amount of source data of the job; The concurrency of the execution plan is initially configured.

In this embodiment of the present application, the master node can configure the concurrency of the direct downstream stage based on the amount of data output by the upstream stage; based on this, since the first executed stage in the execution plan does not have an upstream stage, the master node is in the process of monitoring the execution plan. When initializing the configuration of concurrency, the concurrency should be initially configured for at least the stage executed first, so that the stage executed first can execute the job through the concurrency of the initial configuration; and for the stage that is not executed first in the execution plan, because it is not executed first The stages generally have upstream stages, so the stage that is not executed first can be used as the direct downstream stage of one or more upstream stages, so as to configure the concurrency based on the statistics of the output data of the upstream stage.

Since the stage that is not executed first in the execution plan can dynamically adjust the degree of concurrency during job execution, when initializing and configuring the degree of concurrency, the embodiment of the present application does not necessarily need to initialize and configure the degree of concurrency for the stage that is not executed first. Of course, the embodiments of the present application may also support initializing and configuring the concurrency for stages that are not executed first, which is not limited in the embodiments of the present application. In an example, as shown in FIG. 1C , V1, as the first stage to be executed, needs to initialize the configuration concurrency, and V2, V3, and V4 all have upstream stages. This embodiment of the present application does not limit the need for V2, V3, and V4. Initialize the configuration concurrency.

In step S313, during the job execution process, the data volume distribution information of the output data of the upstream stage is determined, where the data volume distribution information includes the data volume of the multiple data partitions corresponding to the output data.

During job execution, the master node can schedule worker nodes to process the input data of the upstream stage according to the concurrency of the upstream stage, so that the worker nodes scheduled by the upstream stage can generate the output data of the upstream stage after processing the input data. The output data of the upstream stage can be divided into multiple data partitions (Partitions). The data volume of the data partition of the output data can be used as the data volume distribution information of the output data, that is, the data volume distribution information of the output data can represent the data volume distributed in each data partition of the output data, and the data volume distribution information can be carried in the output data. in the statistics of the data.

In this embodiment of the present application, the data volume distribution information of the output data of the upstream stage may be collected by the master node.

In some embodiments, the upstream stage referred to in step S313 may be the stage that is executed first, or the stage that is not executed first. It should be noted that if the upstream stage is the stage that is executed first, the upstream stage can schedule worker nodes to process data based on the concurrency of the initial configuration; if the upstream stage is the stage that is not executed first, the upstream stage can be based on the embodiments of this application. The provided solution dynamically adjusts the concurrency, and then schedules worker nodes to process data.

In step S314, according to the ideal data size (Ideal DataSize) corresponding to the work node of the directly downstream stage and the data size distribution information, the processing data size allocated to the work node of the directly downstream stage is determined.

After the master node collects the data volume distribution information of the output data of the upstream stage, the master node can allocate the processing data volume to the worker nodes of the directly downstream stage based on the data volume distribution information. This embodiment of the present application can automatically configure the concurrency of the direct downstream stage after the worker nodes of the direct downstream stage complete the allocation and processing of the data amount.

In some embodiments, the embodiments of the present application may preset the ideal data volume corresponding to the worker nodes of the directly downstream stage. If the processing data volume of the worker nodes exceeds the ideal data volume, the processing load of the worker nodes will be exceeded, causing the worker nodes to execute The probability of failure increases, so when allocating processing data to the worker nodes of the directly downstream stage, the data volume of the processing data should be close to but not exceeding the ideal data volume.

Based on the data volume distribution information that expresses the data volume of the data partition of the upstream stage, when the master node allocates the processing data volume to the worker nodes of the directly downstream stage, it can determine the direct data volume based on the ideal data volume and the data volume distribution information. The amount of processing data allocated by each worker node of the downstream stage, such that the amount of processing data allocated to the worker nodes does not exceed the ideal amount of data.

In some embodiments, the master node can allocate the continuous data partition output by the upstream stage and the total data volume does not exceed the ideal data volume to a worker node of the directly downstream stage, so that the worker node of the directly downstream stage can be allocated to the continuous data partition. and the total data volume does not exceed the ideal data volume. In some further embodiments, the master node can ensure the processing of the allocation of each direct downstream stage on the basis of allocating multiple data partitions with a total data volume that does not exceed the ideal data volume to one worker node of the directly downstream stage The amount of data is as balanced as possible.

In one example, FIG. 3C shows an example diagram of data partitions (Partitions) assigned to directly downstream stages. As shown in Figure 3C, there are multiple Partitions output by the upstream stage, and the value in the box representing the Partition in Figure 3C is the data volume of the Partition; it can be seen that when the master node allocates the processing data volume for the direct downstream stage, it is Allocate continuous data partitions with a total data volume that does not exceed the ideal data volume to a worker node of the directly downstream stage, and ensure that the processing data volume of each worker node tends to be distributed evenly. For example, the 4 worker nodes of the directly downstream stage are allocated 15, 16, 19 and 10 of the processed data respectively. It can be seen that after the processing data volume allocated to the worker nodes of the direct downstream stage, the master node can complete the configuration of the concurrency of the direct downstream stage. For example, after each worker node in Figure 3C completes the allocation of processing data volume, the master node can The node can determine that 4 worker nodes (with a concurrency of 4) are needed to process the output data of the upstream stage. By configuring concurrency in the manner provided by this embodiment of the present application, it is possible to avoid the situation that although the processing data volume of a certain worker node of the direct downstream stage does not exceed the ideal data volume, it is too high or too low compared with other worker nodes, thus making The amount of data processed by each worker node can be balanced and reasonable.

In step S315, the concurrency of the directly downstream stage is configured according to the number of worker nodes in the directly downstream stage that are allocated the amount of processing data.

After the host node determines the amount of processed data allocated by the worker nodes of the direct downstream stage, it can complete the concurrency configuration of the direct downstream stage based on the number of worker nodes that are allocated the amount of processed data.

In the distributed job adjustment method provided by the embodiment of the present application, after the execution plan of the job is generated, the concurrency degree is initially configured for at least the stage executed first in the execution plan, so that the job can be actually executed. During job execution, the master node can determine the data volume distribution information of the output data of the upstream stage of the execution plan, and the data volume distribution information can express the data volume of multiple data partitions corresponding to the output data; thus, based on the work of the directly downstream stage The ideal data volume corresponding to the node, and the data volume distribution information, determine the processing data volume allocated to the worker nodes of the direct downstream stage; further, the master node can configure the direct The concurrency of the downstream stage enables each worker node in the direct downstream stage to execute the job with a data volume that does not exceed the ideal data volume, reducing the execution failure caused by the excessive processing data volume of a single worker node in the direct downstream stage. probability. In the embodiment of the present application, during the job execution process, based on the data volume data volume distribution information of the output data of the upstream stage, and the ideal data volume corresponding to the work node of the directly downstream stage, the processing data volume can be allocated to the work node of the directly downstream stage, And realize the automatic configuration of the concurrency of the direct downstream stage, to achieve the effect of dynamic adaptive adjustment of the concurrency of the direct downstream stage based on the output data results of the upstream stage, and improve the accuracy and rationality of the concurrency configuration of the direct downstream stage. , which greatly reduces the waste of computing resources caused by unreasonable concurrency configuration, and can significantly improve the performance of distributed systems.

The solution provided by the embodiment of the present application may be called a Fair-Parallelism (evenly similar) strategy based on the amount of Partition data. As an optional implementation process, take the upstream stage to implement data shuffle (in distributed job execution, the process of calculating the hash value of the data record according to a specific Hash function and sending it to the corresponding worker node) as an example, the master node can obtain the upstream stage. After the data shuffle, the worker node outputs the partition data volume distribution information (the data volume of each partition). The master node can allocate the amount of processing data to the worker nodes of the directly downstream stage according to Partition aggregation, for example, by allocating multiple Partitions that are consecutive and whose total data volume does not exceed the ideal amount of data to a worker node of the directly downstream stage. In the above allocation process, if the total amount of data of multiple Partitions allocated to a worker node reaches the ideal data volume, the subsequent consecutive Partitions whose total data volume does not exceed the ideal data volume are automatically allocated to the next worker node. And so on, until all Partitions output by the upstream stage are allocated. The actual concurrency of the direct downstream stage can be automatically determined after all the partitions in the upstream stage are allocated, that is, the concurrency of the direct downstream stage is determined based on the number of worker nodes allocated to process the data in the direct downstream stage. In this embodiment of the present application, the partition output by the upstream stage can be distributed as evenly as possible to each work node of the direct downstream stage, so as to achieve a more accurate and reasonable configuration of concurrency for the direct downstream stage, and avoid the long tail problem of a single work node. .

It should be noted that, after configuring the concurrency for the direct downstream stage in step S315, if the direct downstream stage also inputs other downstream stages through the connection edge (that is, the direct downstream stage also has downstream stages), then step S315 currently configures the direct downstream of the concurrency. A stage can become an upstream stage, and based on the solution provided by the embodiments of the present application, the concurrency adjustment for subsequent downstream stages continues until each stage of the execution plan completes the adjustment of the concurrency during the job execution process.

In some further embodiments, the embodiments of the present application can be further improved through technology to reduce the situation of data skew in the working nodes of the directly downstream stage, so as to meet various data characteristics requirements of job execution, specifically:

When the data volume of multiple consecutive Partitions is very small, the sum of the data volume of multiple Partitions may be less than the ideal data volume. In this extreme case, a large number of Partitions may be allocated to a single worker node for processing. However, if the worker nodes of the directly downstream stage operate a large number of Partitions sequentially, this may lead to a fallback in read and write performance. Based on this, the embodiment of the present application can set an upper limit of the number of Partitions allowed to be allocated by a single worker node, so as to control the upper limit of the number of Partitions added to the worker node on the premise that the amount of data processed by the worker node is zero.

When the running time of a worker node has less correlation with the size of the processed data, but is related to other characteristics of the data (such as the number of data records, etc.), if the worker nodes are allocated processing data according to the size of the partition data, it may be Allocate a partition with a small amount of data but a large number of data records to a single worker node, which will prolong the running time of the worker node and cause a bottleneck in job execution; in addition, it is necessary to consider the unit data records involved in the worker node itself to calculate The complexity, which may be related to the number of operators on the worker nodes, as well as the characteristics of the operators and other factors, also need to be taken into account when calculating concurrency. Based on this, in addition to using the data size of the Partition as a benchmark to allocate the processing data volume of the worker nodes, the embodiment of the present application can further combine the number of records of the Partition, the number of operators of the worker nodes, and the complexity of the operators, etc. The amount of processing data allocated to the worker nodes is readjusted, and the results with higher concurrency obtained in these dimensions are selected to complete the configuration of the final concurrency of the direct downstream stage. Based on this, in an optional implementation of step S315, the embodiment of the present application may further according to the number of work nodes allocated to process the amount of data in the direct downstream stage, the number of records of the Partition allocated to the work node, the number of operators of the work node, As well as the operator complexity, the final concurrency configuration of the direct downstream stage is completed.

When dynamically adjusting the concurrency of job execution, the embodiments of the present application can combine the data characteristics of multiple dimensions to calculate the execution results statistically, and provide more optimization directions for the dynamic execution mode of the distributed system engine; the embodiments of the present application are applied to In the actual production environment, it can bring an order of magnitude reduction to the overall execution concurrency of production jobs, significantly improve the computing efficiency of work nodes in a distributed system, and greatly avoid the waste of computing resources and frequent scheduling of work nodes. consumption. Compared with the simple and direct Even-Reduction strategy, the Fair-Parallelism strategy proposed in the embodiment of the present application can effectively avoid the serious data skew that may be introduced under the Even-Reduction strategy, so that the amount of data processed by all working nodes in a distributed job is increased. Distribute as evenly as possible to avoid prominent long tails or short boards, which can prevent long tails from becoming a bottleneck in running time, and can also prevent worker nodes from frequently processing small data and causing waste of resources caused by repeatedly starting worker nodes. Further, when merging Partitions with a small amount of data, the upper limit of the number of consecutively merged Partitions is limited, so as to prevent excessive merging of Partitions from causing degradation of data reading performance of downstream stages. Further, according to the detailed data information (including the amount of data and the number of Records) generated by each work node during the operation of the statistical job, the embodiment of the present application can be combined with information such as the number of Records, the number of operators, and the complexity of operators to make a More balanced and perfect concurrency adjustment. It can be seen that the embodiments of the present application can significantly improve the performance of the distributed system.

Based on the statistical information of the output data of the upstream stage, the embodiments of the present application provide a solution for dynamically merging multiple Partitions. At the same time, for a Partition with a relatively large amount of data, the embodiment of the present application also provides a partitioning solution for the Partition. The specific implementation scheme is described in detail below.

For data orchestration and shuffle, effective data orchestration and shuffle are important prerequisites for the horizontal expansion of distributed systems. For the execution stage of Map-Reduce, efficient data shuffle has always been one of the important performance indicators of the execution performance of distributed jobs. However, full-connection data shuffle (full-shuffle) can only be executed efficiently in the ideal scenario where the data is evenly distributed. In actual production operations, the distribution of data is often uneven, and the skewed characteristics of data may be further amplified in the full-shuffle mode, resulting in a significant increase in the running time of individual worker nodes and serious long tails.

It should be noted that the concept of shuffle comes from the calculation model of Map-Reduce. Although the modern job execution framework has evolved to use a more general DAG-based description, the operation mode of Map-Reduce is still the sub-graph in the DAG. Important description. In many more complex DAG topologies, the data flow on each connection edge can also be described by various shuffle models; for example, in a distributed execution framework, an important physical attribute of the connection edge of a DAG is the transmission of data on the edge. The transmission of data on the edge can not only use full-shuffle, but also allow the introduction of a more dynamic and intelligent data arrangement, and use this to solve the problems faced by full-shuffle in many practical scenarios.

Taking dynamic partition writing as an example, for a typical distributed data warehouse system, partition tables are widely used. There are usually two ways to write data into the partition table: static partition writing and dynamic partition writing; when the value that needs to be written to the partition can be specified in advance, it is relatively simple to directly use static writing to the specified partition; for the partition value It cannot be judged in advance, especially when the data produced by a query is distributed in multiple partitions, the dynamic partition (Dynamic Partition) is used to write, that is, the value of the data written to the Partition will be written according to the specific output during the job running process. data to obtain. For example, in the following SQL (Structured Query Language) statement, the data is written out as the partition column name designated as country, but which partitions will be written to, and the corresponding partition values are obtained during operation:

INSERT OVERWRITE TABLE Partitioned_sales Partition(country);

SELECT country, item, price from sales.

In a distributed system, due to the uncertainty of the number of partitions, the diverse and uneven characteristics of data distribution, it has always been a challenging problem to achieve general efficient dynamic partition writing. In this regard, the distributed system needs to be able to avoid the serious long tail of the job, and at the same time not bring a serious burden to the storage system due to the large number of small files.

In some embodiments, the master node can implement dynamic partitioning through a single Map stage execution plan. In this implementation, the execution plan generated by the master node contains only one Map stage, which can execute tasks through one or more worker nodes. After each worker node of the map stage reads the data, it can write out the file according to the partition value (such as country in the above example). That is, if a worker node processes data corresponding to different country values, it will generate corresponding files with different paths. The execution plan of a Map stage is simple and intuitive, but it will bring various hidden dangers to the actual large-scale distributed system, the most prominent of which is the fragmentation of small files. In the execution plan of a Map stage, assuming that the concurrency of the Map stage is M, the value of country in the user data may be N. In the case of random distribution of data, because each worker node of the Map stage outputs independently, each partition may eventually write out M files. Therefore, this execution method will eventually generate M*N data files, and there may be a large number of small files among them. The existence of these large numbers of fragmented small files will have a greater negative impact on a distributed system:

For the storage of distributed systems, the management of a large number of small files needs to consume huge Meta (meta) information; in extreme cases, even the main node of the entire distributed system will be destroyed, making the entire distributed system unavailable; In terms of storage efficiency, the existence of fragmented small files will also bring about a poor storage compression ratio and take up more storage space;

After the data is generated, it always needs to be processed. After the upstream stage produces small files, the calculation consumption of the downstream stage will increase, and the actual data reading efficiency will also be lower;

For the Map stage, since it is necessary to write out M*N small files, it is necessary to create an M*N write service (writer), and each writer reserves a certain Memory Buffer (memory cache) for Encoding (compiling), Compress (compression), etc.; and if the Buffer is too large, the memory used by the Map stage will be too large, and if the Buffer is too small, the effects of Encoding and Compress will be poor.

In other embodiments, the master node may implement dynamic partitioning based on a fully connected Reshuffle. In order to avoid the above-mentioned small file problem, the master node can perform a fully connected Reshuffle according to the PartitionKey (key) after generating the input data in an execution stage, that is, aggregate the data of the same Partition to a worker node and then write it out. . In this way, it is ensured that only one partition file is generated for each Partition's partition value. However, this mandatory restriction, while reducing the number of files, will also bring about another negative effect: data skew.

It should be noted that for the data generated in the execution phase, the data distribution is various, and the data distribution of the dynamic partition cannot be obtained before the Query is executed. For example, in the example of the above SQL statement, if the user data of a certain platform is partitioned in different countries, the data in the local partition will undoubtedly be much larger than other partitions, which leads to the work node corresponding to the data in the local partition. A serious long tail occurs, causing the running time of the entire job to be greatly elongated. For heavily skewed data, this long tail may bring hundreds or even thousands of times of job delay, and also have a very bad impact on the resource utilization of the entire distributed system.

In order to alleviate this problem, the implementation of Reshuffle can actually be optimized by introducing an additional Randomshuffle Key (random shuffle key). For example, through a Reshuffle key with a value of [0,9], the data is randomly scattered into 10 partitions to reduce the skewness of the data. However, such a solution still has certain problems: in the case of serious data skew, even if the data is divided into 10 parts, the data after the division may still have a skew problem, and may even change from a long tail of data to 10 A slightly lighter data long-tail situation; for data that is not skewed or a partition with a small amount of data itself, if it is also divided into 10 parts, it will bring an increase in the number of final files (10*N); distributed systems force Adding Random Key's shuffle method may destroy the idempotency of data. In a distributed system, if the worker node reruns, there is a risk of data correctness.

Based on the above situation, the embodiment of the present application performs intelligent dynamic data arrangement according to the data distribution situation generated in real time by the upstream stage (for example, the Map stage), and solves the problem of small files while ensuring the balance of the written data, thereby overcoming the above two kinds of problems. Disadvantages of the program.

First, use Figure 4A to briefly describe the shuffle method (for ease of understanding, the details of performing shuffle according to Reduce concurrent modulo are omitted here). Figure 4A shows an example diagram of data shuffle. As shown in Figure 4A, in Figure 4A, the same linear box represents the data of the same Partition Key in the shuffle data generated in the Map stage; The solid-line boxes respectively represent the data of the same Partition Key. The value in the box represents the corresponding amount of data. After a shuffle, the data of the same Partition Key will be handed over to the same worker node in the Reduce phase for processing. As can be seen from Figure 4A, the amount of data processed by worker node R#0 in the Reduce phase is 8, the amount of data processed by worker node R#1 is 2, the amount of data processed by worker node R#2 is 43, and the amount of data processed by worker node R# 3 The amount of data processed is 16; the data processed by worker node R#2 has serious data skew, which leads to a very serious long tail when worker node R#2 executes tasks in the Reduce phase.

For scenarios such as dynamic partition insertion, the data of the same partition only needs to be guaranteed to be written to the same directory, not to the same data file. Therefore, during the shuffle process, it is not necessary to ensure that all data with the same Partition Key are handed over to the same reduce worker node for processing. Based on this feature, the embodiments of the present application can automatically split Partitions with a large amount of data generated in the Map phase, and then hand them over to multiple work nodes in the Reduce phase for processing.

As an optional implementation, FIG. 4B shows yet another flowchart of the distributed job adjustment method provided by the embodiment of the present application. The method flow may be implemented by the master node. Referring to FIG. 4B , the method flow may include the following steps.

In step S410, the job submitted by the user is acquired.

In step S411, an execution plan of the job is generated, and the execution plan includes a plurality of stages, and the plurality of stages include an upstream stage and a stage directly downstream of the upstream stage.

In some embodiments, the upstream stage is such as the Map stage. Direct downstream stages such as Reduce stage.

In step S412, during the execution of the job, the Statistics of the output data of the upstream stage is obtained, where the Statistics includes: data volume distribution information of the Partition of the output data.

In some embodiments, the worker node(s) of the upstream stage may report the Statistics of the output data to the master node during the execution of the task and after the execution ends. The Statistics may include: data volume distribution information of a Partition of the output data, where the data volume distribution information may indicate the data volume of each Partition of the output data. In some further embodiments, the Statistics may further include: the data volume of the output data (for example, the data volume of the output data before and after data compression), the number of records of each Partition in the output data, and the like.

In step S413, according to the ideal data amount, the partitions whose data amount is larger than the ideal data amount in the output data of the upstream stage are divided.

In this embodiment of the present application, the ideal amount of data for the partition of the work node of the directly downstream stage can be preset. It should be noted that when different stages of the execution stage are used as direct downstream stages, the ideal amount of data set can be different. After obtaining the Statistics of the output data of the upstream stage, the master node can determine the data volume of each Partition in the output data. Therefore, based on the ideal data volume of the Partition by the working node of the direct downstream stage, it is judged whether there is a Partition whose data volume is greater than the ideal data volume in the output data; for a Partition whose data volume in the output data is greater than the ideal data volume, the embodiments of the present application may The partition is split, and the data volume of the split partition is not larger than the ideal data volume.

In some embodiments, the embodiments of the present application may, according to the ideal data volume, split the partitions whose data volume is greater than the ideal data volume in the output data, so that the data volume of the split Partitions is not greater than the ideal data volume and is close to the ideal data volume. Evenly distributed.

In step S414, the split Partition is allocated to the directly downstream stage, and one split Partition is configured to be executed by a worker node of the directly downstream stage.

After the partition of the output data is split, the embodiment of the present application can assign the split partition to the directly downstream stage, and a split partition is configured to be executed by a work node of the directly downstream stage, so as to realize Configure the data to be processed for the immediate downstream stage.

In some further embodiments, if there are Partitions whose data volume is less than the ideal data volume in the output data, in this embodiment of the present application, at least two Partitions whose total data volume is not greater than the ideal data volume in the output data may be merged. The data volume of each Partition in the two Partitions is smaller than the ideal data volume, so that the merged Partition is allocated to the directly downstream stage, and a merged Partition is configured to be executed by a worker node of the directly downstream stage to realize the direct downstream stage. Efficient use of computing resources.

In this embodiment of the present application, by acquiring the statistics of the output data of the upstream stage during the job execution process, based on the data amount of each partition of the output data indicated in the statistics, the partitions whose data amount is greater than the ideal data amount are divided, ensuring that The data volume of the partitioned partition is not larger than the ideal data volume. Then, the split Partition is assigned to the directly downstream stage, and a split Partition is configured to be executed by a worker node of the directly downstream stage, which realizes the configuration of the data to be processed for the worker node of the directly downstream stage. This embodiment of the present application can dynamically adjust the Partition configured to the directly downstream stage according to the statistics of the output data of the upstream stage, so as to ensure that the data volume of a Partition processed by each worker node in the directly downstream stage does not exceed the ideal data volume, so that the The data processed by the working nodes of the direct downstream stage can be distributed evenly, which reduces the data skew of the direct downstream stage, and avoids the running time caused by the individual working nodes of the direct downstream stage needing to process the partition with a large amount of data. A greatly elongated long tail problem. It can be seen that the embodiment of the present application can make the data processed by the working nodes of the directly downstream stage tend to be evenly distributed, reduce the data skew of the directly downstream stage and the long tail problem of the working nodes, and significantly improve the performance of the distributed system.

In some embodiments, for the Map-Reduce stage, Map can be considered as an upstream stage, and Reduce can be considered as a direct downstream stage of Map. Each worker node in the Map stage can report the Statistics of the output shuffle data to the execution engine (the execution engine can be the master node) during and after the operation; for example, there may be one or more worker nodes in the Map stage To execute the task, each worker node can report the Statistics of the output shuffle data to the execution engine during and after the execution of the task. In some embodiments, the statistics of the shuffled data are, for example, the data volume of the shuffled data before and after compression, the data volume of each Partition of the shuffled data, and the number of Records included in the Partition, and the like.

In some embodiments, the embodiments of the present application may define an ideal amount of data corresponding to one Partition in the Reduce phase. For any Partition output in the Map stage, if the data volume of the Partition is greater than the ideal data volume, the embodiment of the present application can split the Partition according to the ideal data volume, and try to ensure the data volume processed by each worker node in the Reduce phase as much as possible. is uniform; if the data volume of the Partition is less than the ideal data volume, the embodiment of the present application may combine multiple Partitions with the data volume less than the ideal data volume, and ensure that the data volume of the merged Partition is not greater than the ideal data volume.

In an implementation example, FIG. 4C shows an example diagram of partition splitting in an embodiment of the present application. As shown in Figure 4C, the 4 shuffle data output in the Map stage can be divided into 4 Partitions (P#0, P#1, P#2 and P#3), for example, the data aggregation of the same linear expression in the 4 shuffle data The data corresponding to the thin dashed line in the shuffle data is aggregated into P#0, the data corresponding to the thin solid line is aggregated into P#1, the data corresponding to the thick dashed line is aggregated into P#2, and the data corresponding to the thick solid line is aggregated into a Partition. Aggregated into P#3. As shown in FIG. 4C , the data sizes of the four Partitions are 8, 2, 43, and 16 in sequence. Assuming that the ideal data volume corresponding to a Partition that defines the Reduce phase is 10, the data volume of P#0 and P#1 are both less than 10, and the combined data volume of P#0 and P#1 does not exceed the ideal data volume of 10 , so P#0 and P#1 can be combined into a Partition and assigned to a worker node R#0 in the Reduce phase for processing. Since the data volume 43 of P#2 is greater than the ideal data volume 10, then P#2 can be split into multiple Partitions as evenly as possible according to the ideal data volume, and the data volume of each partition is not greater than the ideal data volume , as shown in Figure 4C, P#2 can be divided into 5 Partitions as much as possible according to the ideal amount of data, and allocated to the 5 working nodes R#1 to R#5 in the Reduce stage for processing, where R#1 To R#4, each partition with a data volume of 9 is allocated, and R#5 is allocated a partition with a data volume of 7. Similarly, since the data volume of P#3 is larger than the ideal data volume, it is equally divided into two partitions and assigned to the two worker nodes R#6 and R#7 in the Reduce phase for processing.

The above-mentioned mechanism for splitting Partitions with a larger than ideal amount of data in the Map stage and merging Partitions with a smaller amount of data in the Map stage may be called an Adaptive shuffle mechanism. In some further embodiments, based on the Adaptive shuffle mechanism, the number of files finally written in the Reduce phase will mainly depend on the size of the input partition data volume and the size configuration of the ideal data volume. Assuming that the ideal number of partitions (Ideal Parallelism) in the input Reduce phase is defined as: the total amount of shuffled data divided by the ideal amount of data; then after the Adaptive shuffle mechanism of the embodiment of the present application, the maximum number of files finally generated in the Reduce phase is Ideal Parallelism Add N. Therefore, if you want to ensure that at least one data file exists on each Partition with output data, only the Partition with a larger data volume than the ideal data volume will increase the number of files after splitting. However, the file sizes generated by this split are all on the order of Partition, rather than fragmented small files. Based on this, if the size of the ideal data volume is reasonably configured, no matter how many files are generated, it is reasonable for a distributed system; that is, as long as the file size is appropriate, massive data can be stored by a little more files.

That is to say, based on the Adaptive shuffle mechanism, the amount of data processed by each worker node in the Reduce phase will not exceed the given ideal amount of data, and each Partition smaller than the ideal amount of data can be output by a worker node in the Reduce phase a file. This fundamentally solves the problem of generating a large number of small files and possible data skew, and can better solve the dilemma faced in dynamic partition scenarios. In addition, it should be noted that using the Adaptive shuffle mechanism can avoid the need to add an additional Random shuffle Key to reduce data skew. Since the Adaptive shuffle mechanism provided by the embodiments of this application is deterministic and reentrant in the whole process, in an unstable distributed system environment, it can fundamentally ensure that the output data is correct when various retries occur. sex.

The embodiment of the present application realizes adaptive data arrangement by combining the statistical information collection capability during job running provided by the execution engine (master node) and the dynamic adjustment capability of the multi-job execution graph and shuffle mode; it can be based on the output of the upstream stage. The data characteristics of the data allow for intelligent allocation and arrangement of output data, including automatic splitting of slanted data partitions, and merging of multiple small partitions with a large amount of data, which fundamentally solves the problem of data shuffle during the process of data shuffling. It may bring about data skew and long-tail problems of worker nodes, while avoiding the data fragmentation problem in other non-shuffle solutions, which can significantly improve the performance of distributed systems.

In addition to the general optimization of Partition merging and splitting, the embodiments of the present application perform specific optimizations for specific scenarios. Taking the Join scenario as an example, the implementation scheme of the Join operation based on the output data of the upstream stage is introduced.

In the job execution process of a distributed system, the Join operation is one of the most common and complex operations. Since it involves the calculation and processing of multi-channel data, in addition to the challenges that generally need to be solved in distributed systems, the interaction of different channels of data on the Join operator also generates more data processing scenarios. However, uneven data distribution will lead to data skew and long tail problems in Join operations, which have become common problems in distributed systems and have not been systematically solved.

FIG. 5A exemplarily shows an example diagram of the Join process of distributed SQL. As shown in Figure 5A, the two input data provided by the upstream stage to the downstream stage (M1 and M2 shown in Figure 5A) are joined in the downstream stage, wherein the two input data of M1 and M2 can be written according to Partition, And according to the partition number, it is shuffled to different worker nodes of J3 to realize data join. Among them, the intermediate data will be arranged according to the partition and stored on the physical medium.

Figure 5A shows that the partition data distribution of the upstream stage is relatively uniform, but in the actual query (query) and data processing, the distribution of the partition data of the upstream stage is likely to be skewed. FIG. 5B shows another example diagram of the Join process of distributed SQL. As shown in Figure 5B, Partition1 (Partition with partition number 1) in the input data provided by M1 to J3 has serious data skew, and at the same time, Partition1 in the input data provided by M2 to J3 has a slight data skew. In this case, when J3 performs the Join operation on Partition1, there will be a serious long tail on the J3 worker node, and the job may even fail due to memory overrun.

To solve the above problem, FIG. 5C shows another flowchart of the distributed job adjustment method provided by the embodiment of the present application. The method flow may be implemented by the master node. Referring to FIG. 5C , the method flow may include the following steps.

In step S510, the job submitted by the user is acquired.

In step S511, an execution plan of the job is generated, the execution plan includes multiple stages, the multiple stages include a Join stage and a direct upstream stage of the Join stage, and the direct upstream stage provides multiple input data to the Join stage ; Among them, one input data includes multiple Partitions.

In the case of Join operation, the Join stage, as the direct downstream stage of a stage, can input multiple input data provided by the stage, and one input data can include multiple Partitions.

In step S512, during the job execution process, if there is a target Partition with skewed data in any input data, the target Partition is split into multiple Partitions, and the multiple sub-Partitions are allocated to multiple jobs of the Join stage node.

During the job execution process, the master node can obtain the statistics of the output data of any stage. Based on the statistics, if any input data provided by the upstream stage directly to the Join stage has a data-sloped Partition (for the convenience of explanation, there is a data-sloped partition). The Partition may be referred to as a target Partition), and in this embodiment of the present application, the target Partition may be split into multiple Partitions, and the multiple Partitions may be allocated to multiple work nodes of the Join stage.

In some embodiments, the embodiments of the present application can determine whether the data volume of the Partition is greater than the first data volume threshold for any partition in the input data, and if the data volume of the Partition is greater than the first data volume threshold, it is determined that the Partition exists The target Partition for the data skew.

In some embodiments, when the target Partition is split, the embodiments of the present application may split the target Partition based on the second data volume threshold, so that the data volume of the multiple sub-Partitions after the split of the target Partition is evenly distributed and The data volume of each sub-Partition is not greater than the second data volume threshold. In some further embodiments, the second data volume threshold is smaller than the first data volume threshold, and specific values of the first data volume threshold and the second data volume threshold may be defined according to actual conditions, which are not limited in the embodiments of the present application. In an implementation example, the second data volume threshold may be, for example, the ideal data volume described above.

In some embodiments, when multiple sub-Partitions split into a target Partition are allocated to multiple work nodes of the Join stage, one Partition may be allocated to one work node of the Join stage.

In step S513, the Partition belonging to the same partition number as the sub-Partition in the input data of other channels is broadcasted to the working node allocated by the sub-Partition.

In order to realize the correct Join operation in the Join stage, in this embodiment of the present application, the target Partition of a certain channel of input data is split into multiple sub-Partitions, and after the sub-Partitions are allocated to the work nodes of the Join stage, the target Partition is split for the target Partition. Each sub-Partition of the sub-Partition, the embodiment of the present application can determine that in the input data of other channels (input data of a different channel from the input data of the certain channel in the multi-channel input data), and the sub-Partition belong to the Partition of the same partition number, so that the determined Partitions are broadcast to work nodes assigned by sub-Partitions, so that Partitions belonging to the same partition number in the sub-Partition and other input data can be correctly joined in the Join stage.

In some embodiments, the multiplex input data provided by the directly upstream stage may include at least the first channel input data and the second channel input data, and the first channel input data and the second channel input data may be any of the multiplex input data. Two-way input data, the number of multiple-way input data can be more than two, not limited to only two-way input data. In this case, there may be the following situations when the target Partition exists in the multiplexed input data.

In the first case, there is a target Partition in one of the input data in the multi-channel input data; One target Partition in the first input data is divided into multiple sub-Partitions, and each sub-Partition is assigned to the work node of the Join stage; while other input data (inputs other than the first input data in the multiple input data) The Partition that belongs to the same partition number as the target Partition in the data) will be broadcast to the worker nodes allocated by each sub-Partition.

As an implementation example, for the above-mentioned first case, FIG. 5D shows yet another example diagram of the Join process. As shown in FIG. 5D , FIG. 5D shows the input of two channels of Join (M1 and M2), and one of them has a scenario where the data is skewed. Assuming that there is no need for additional operators to maintain features such as Partition and Sort on the worker node that executes Join, the master node can determine that Partition1 output by M1 is the target with data skew after counting the data volume of each Partition output by M1 and M2. Partition, thereby splitting Partition1 output by M1 into multiple sub-Partitions. In FIG. 5D , a small box in the box where Partition1 of M1 is located may represent a split sub-Partition. Partition1 of M1 is divided into multiple sub-Partitions, which can be allocated to multiple work nodes in the Join phase for Join processing. At the same time, in order to perform the correct Join operation on the split sub-Partition in the Join phase, the master node needs to broadcast the Partition with the same partition number as the target Partition in M2 to the worker nodes assigned by each sub-Partition; for example, The master node needs to broadcast the Partition1 generated by M2 to the working nodes allocated by each sub-Partition of M1. By broadcasting Partitions with the same partition numbers of M1 and M2 to multiple worker nodes in the Join phase, when the target Partition is split into multiple sub-Partitions, it is possible to ensure that the data with the same partition number can be joined correctly and the result can be produced.

In the second case, there are multiple target Partitions in one input data in the multi-channel input data; The processing of each target Partition in the input data is the same as the first case. In some embodiments, for each target Partition in the first input data, the master node may split the target Partition into multiple sub-Partitions, and assign the sub-Partitions of the target Partition to the work nodes of the Join stage, while other Partitions belonging to the same partition number as the target Partition in the input data will be broadcast to the work nodes allocated by each sub-Partition. However, in the case where there are multiple target Partitions in the first input data, each target Partition needs to be processed in the above manner.

That is, if there is one target Partition or multiple target Partitions in the first input data, for any target Partition, the embodiment of the present application may split the target Partition into multiple sub-Partitions, and divide the target Partition into multiple sub-Partitions. Each sub-Partition is allocated to the work node of the Join stage, and at the same time, the Partition belonging to the same partition number as the target Partition in the input data of other channels is broadcast to the work node allocated by each sub-Partition of the target Partition.

Further combined as shown in Figure 5D, if data skew occurs in multiple partitions in the single input of Join, each partition with data skew can be split, and the same partition number of another channel can be broadcast to the worker nodes in the Join phase. In the Join phase, the split sub-Parition is correctly joined with the data of the same partition number in another way. For example, assuming that in Figure 5D, Partition3 output by M1 also has data skew, then Partition3 output by M1 can also be split into multiple sub-Partitions and assigned to multiple work nodes in the Join phase. At the same time, Partition3 in M2 will be Broadcast to the worker nodes assigned by the sub-Partitions corresponding to Partition3 of M1 respectively.

In the third case, at least two input data in the multi-channel input data have a target Partition; The node can split a target Partition in the first input data into multiple sub-Partitions, and assign each split sub-Partition to the work node of the Join stage; at the same time, the master node can split one of the second input data The target Partition is split into multiple sub-Partitions, and each split sub-Partition is assigned to the work node of the Join stage. For the sub-partitions with the same partition number in the first channel of input data and the second channel of output, it is necessary to broadcast the sub-partitions to the work nodes allocated by other sub-partitions of the same partition number. For example, broadcast the sub-Partition of the first input data to the working node assigned by the sub-Partition with the same partition number of the second input data, and broadcast the sub-Partition of the second input data to the first input data. Worker nodes assigned to child Partitions with the same partition number.

Figure 5E shows yet another example diagram of the Join process. As shown in Figure 5E, both Partition1 of M1 and M2 have data skew. The Partition1 output by the left channel M1 needs to be split into multiple sub-Partitions, and the Partition1 output by the right channel M2 needs to be split into multiple sub-Partitions. Then the Partition1 of M1 needs to be split into multiple sub-Partitions. Multiple sub-partitions that are split will be assigned to multiple work nodes in the Join stage, and multiple sub-Partitions split by Partition1 of M2 will also be assigned to multiple work nodes in the Join stage, and one sub-Partition is assigned to one work node. In FIG. 5E , a small box in the box where Partition1 of M1 and M2 is located may represent a sub-Partition that is split. In order to ensure that the sub-Partitions with the same partition number of M1 and M2 can be correctly joined, the master node can broadcast the multiple sub-Partitions split by Partition1 of M1 to the work nodes allocated by the sub-Partitions of the same partition number of M2. Similarly, The master node can broadcast the multiple sub-Partitions split by Partition1 of M2 to the worker nodes allocated by the sub-Partitions of the same partition number of M1.

In the fourth case, at least two input data in the multi-channel input data have multiple target Partitions; in the fourth case, take the first input data and the second input data as an example where there are multiple target Partitions , the processing process for the same target Partition in the first input data and the second input data is the same as the third case. However, in the case where multiple target Partitions exist in both the first input data and the second input data, each same target partition of the first input data and the second input data needs to be processed in the above-mentioned manner.

In the above example, the worker nodes in the Join phase mainly implement pure Join connections, and do not include operators with data sorted (sorted)/Partitioned (sharded) attributes. In practice, as the most complex and flexible Join operation in SQL syntax, its work node may contain various operators. In this case, when the data of the input Join is skewed, in addition to using the above example to split the Partition, it is also necessary to add a union (union) operation after completing the data Join, so as to realize the split The subsequent sub-Partitions are re-folded after the Join. When the special properties of the data need to be preserved (for example, the data is not directly placed on the disk, but there are downstream operations/stages), the union operation needs to be used to ensure the correctness of subsequent executions.

FIG. 5F is an example diagram of a union operation further shown on the basis of FIG. 5D . As shown in FIG. 5D and FIG. 5F , when the data skew occurs in the Parition1 of the single input M1 of Join, in addition to using the process shown in FIG. 5D to implement the Join operation on the sub-Parition corresponding to the Parition1 of M1 and the Parition1 of M2, the The Join operation results of each sub-Parition need to be combined to generate a new Partition1.

FIG. 5G is an example diagram of a union operation further shown on the basis of FIG. 5E . When both the multiple-input M1 of the Join and the Partition1 of the M2 have data skew, in addition to using the process shown in FIG. 5E to realize the sub-Parition of the Partition1 of M1, and to perform the Join operation with the sub-Parition of the same partition number of the Partition1 of the M2, it is necessary to perform the Join operation. Combine the Join operation results of each sub-Parition to generate a new Parition1.

That is to say, the master node also needs to be configured to combine the Join operation results belonging to the same data partition. The Join operation results of the same data partition include: the sub-data partition of one input data and the data of the same partition number in the other input data The join operation result of the partition, or the join operation result of the sub-data partition of one input data and the sub-data partition of the same partition number in the other input data.

The embodiments of the present application can perform adaptive splitting and allocation when there is a skewed partition in the multi-channel or single-channel input of the Join operation, so as to realize the uniform distribution of the input data of the Join operation, and ensure that the input data of the Join operation is evenly distributed. The correctness of the Join operation in the new shuffle mode. Compared with manual adjustment of job data processing logic and operations that require a large number of end-user interventions, such as data preprocessing, the embodiment of the present application can make the tilt processing of input data of the Join operation adaptive and universal. Especially in the case where the master node collects the statistical information of the output data of each stage, the embodiment of the present application enables the adjustment of the data inclination to be performed automatically without requiring manual adjustment by the user. The self-adaptability of the solution provided by the embodiments of the present application enables it to make dynamic decisions according to the data characteristics of the actual Join input during the operation of the job, without the need for end-user perception and participation, and enables the data to be evenly distributed among multiple distributed work nodes. Distribution can significantly accelerate jobs with skewed data and significantly improve the performance of distributed systems.

In addition to the dynamic adjustment for Partition, according to the statistical information of the upstream stage, the embodiment of the present application can adaptively select an execution path for subsequent execution. The specific implementation scheme is introduced below.

This embodiment of the present application can dynamically adjust the logic diagram of the execution plan during the job execution process (that is, dynamically adjust the logic of the execution plan during the job execution process). It should be noted that the choice of logic diagram may be related to data distribution and characteristics, and these data characteristics can only be accurately obtained during job execution. Different data characteristics may require different logic execution plans to be implemented effectively and accurately. Therefore, for the static execution plan, if the execution plan is determined and cannot be dynamically adjusted during the execution of the job, it is undoubtedly impossible to achieve a reasonable and accurate configuration of the logic of the execution plan.

Taking the Join operation in distributed SQL as an example, to realize the Join operation of a distributed job, there are many different Join algorithms that are completely equivalent in logic, such as Sort Merge Join (sort merge join) and Broadcast Join (broadcast join). Taking the Join of the two source tables, Table1 and Table2, as an example, the implementation process of Sort Merge Join and BroadcastJoin is described below.

Figure 6A shows an example diagram of Sort Merge Join. As shown in Figure 6A, the two input source tables, Table1 and Table2, can be read and preliminarily processed by M1 and M2. For example, after M1 performs data filter (filtering) on Table1, the output data is partitioned according to the shuffle/Join key; The output data of M1 and M2 can be merged with the same key on the downstream worker nodes. In the above process, the implementation of Merge Join requires full shuffle and sort (sorting) operations on the output data of M1 and M2, by ensuring that the data in the same partition can be allocated to the same downstream worker node. However, in a distributed system, Sort Merge Join relies on specific implementations such as outflow to process any amount of data, but the large number of shuffle and sort operations involved in the process require a lot of computing and network resources; When the data distribution is uneven, the shuffled data may cause serious long tails and affect the execution efficiency.

FIG. 6B shows an example diagram of Broadcast Join. Distributed systems need to have the ability to process large tables (such as fact tables) and large tables. For Join operations of large tables and small tables (such as dimension tables), if the size of the data source of the small table can be loaded into a single worker node memory, then the computation/network consumption of distributed shuffle and sort can be reduced. With reference to Figure 6B, Table1 in Figure 6B is a large table, and Table2 is a small table, then the data of the small table can be broadcast to all the working nodes of the large table, and a full hash table (hash table) can be established according to the data of the small table. , so that after the large table data is read, Join can be performed through the hash table lookup (hash table lookup table). In this way, the data of the large table (Table1) only needs to be read once, and there is no need to perform any data shuffle and sort. At the same time, in addition to avoiding the computation/network consumption of shuffle and sort, Broadcast Join can also avoid data skew and long tails that may be caused by shuffle. This is because in the Join pattern (pattern) of Broadcast Join, the logic of Join is actually This happens in the read logic of the large table (for example, the Map stage). Based on this, Broadcast Join is also known as Broadcast Map Join. Of course, when Broadcast Join brings many benefits such as resources/performance, it also has its specific scope of application: the small table data used to build the hash table must be fully loaded by a single worker node. If the optimizer chooses Broadcast Join executes the job, but the data volume of the small table exceeds the memory limit during execution, and the entire job will fail.

Based on the above description, Broadcast Join has significant advantages in performance, and Sort Merge Join itself has better versatility. Therefore, if the optimizer of a distributed system needs to choose a suitable Join algorithm between the two, it needs to be more Accurate and reasonable judgment: On the premise that the job can be successfully completed, configure the efficient execution plan logic as much as possible. However, in the actual online scenario, it is very difficult for the optimizer to make the above judgment before the job is executed. The main reasons are as follows:

Lack of accurate statistical information. The sources of data stored in the distributed system are various. Due to various reasons such as source table data import channels and import time, the statistical information of source table data may be missing or inaccurate (for example, a The table has just been imported and the statistics have not been generated, or the update of the table content has just hit the threshold for discarding statistics, etc.). In short, the lack and inaccuracy of statistical information makes it impossible for the optimizer to accurately estimate the size of the upstream input of Join. The upstream input of Join may be the source table, or the output of the source table after a certain logical transformation.

The data processing logic and data characteristics are complex and changeable. Even if complete and accurate statistical information can be obtained from the source table data, the Join operation may objectively exist in any work node in a data processing process (inside the DAG). With the complex transformation of upstream selection (selection)/filter/aggregation (aggregation), the data of the table may be interspersed with various user-defined code logic (UDF), which cause the optimizer to estimate the input data volume of Join in advance. difficulty.

Because of these limitations, if the optimizer statically selects the Join algorithm before job execution, it needs to face a dilemma. The choice of , can only be conservative as much as possible, for example, setting the threshold of the small table as low as possible, thus losing a lot of optimization opportunities. Even if the small table threshold has been configured very low, due to data estimation errors, data expansion and other reasons, the job execution will still fail due to misjudgment of Broadcast Join. Since these extreme cases will lead to the consequences of job execution failure, it is fed back to the optimizer's strategy, and the optimizer will further choose a more conservative strategy, resulting in a negative cycle. In addition, the triggering of Broadcast Join is largely based on the manually added Map Join hint (MapJoin hint), that is, the generation of the Broadcast Join plan is left to the user to decide; this kind of externalization of the optimizer function brings about the maintenance of user logic. In fact, the user can only accurately perceive the data volume of the source table, but cannot accurately know the output size of the non-source table data after the data change, so the user can not avoid the data by specifying the Map Join hint. and job execution failures caused by changes in upstream processing logic.

Based on the above description, the optimizer of the master node is required to make an accurate selection of the Join algorithm before the job is executed, and its accuracy is affected by various objective reasons. Generally speaking, the characteristics of the data (including data size and distribution, etc.) can only be obtained after the upstream worker node completes the job execution process. Therefore, to make an accurate judgment of the Join algorithm, it is necessary to execute the distributed job process. The judgment is made during the execution of the job, rather than before the job is executed. However, the judgment and selection of the Join algorithm during job execution poses a challenge to the DAG dynamic capability of the execution engine: when different Join algorithms such as Sort Merge Join and Broadcast Join are selected, the execution plan generated is not only in physical attributes (concurrency) degree, shuffle mode, etc.) will be different, and there will be major changes in the topology and logical structure of DAG. Therefore, if you want to make dynamic adjustments during job execution, you need to provide the ability to provide dynamic logic diagrams (also That is, the logic of the execution plan can be dynamically adjusted). It should be noted that the adjustment of the logic diagram is often accompanied by the adjustment of the physical attributes of the execution plan, so this actually puts forward requirements for both the DAG dynamic logic diagram and the dynamic physical diagram capabilities.

Based on the situation described above, FIG. 6C shows yet another flowchart of the distributed job adjustment method provided by the embodiment of the present application. The method process can be implemented by the master node. Referring to FIG. 6C , the method flow may include the following steps.

In step S610, the job submitted by the user is acquired.

In step S611, an execution plan of the job is generated, the execution plan includes multiple stages and a control node, the multiple stages include an upstream stage, and there are multiple candidate execution paths downstream of the upstream stage, An execution path includes one or more downstream stages of the upstream stage; the control node is configured to select a target execution path that the upstream stage actually executes downstream from the multiple execution paths.

In this embodiment of the present application, when the execution plan of the job is generated, it may carry multiple candidate execution paths downstream of the upstream stage, and one execution path may include one or more downstreams of the upstream stage. For example, an execution path may include a directly downstream stage of the upstream stage, or, a direct downstream stage and an indirect downstream stage. In some embodiments, one or more upstream stages in an execution plan may carry candidate execution paths downstream thereof.

In this embodiment of the present application, in addition to including multiple stages, the execution plan may also set a control node downstream of the upstream stage. The control node may be communicated with multiple execution paths of the upstream stage, and may be executed from the multiple execution paths. An execution path is selected from the path as the target execution path of the upstream stage actually executed downstream. It should be noted that the control node does not belong to the execution stage in the execution plan, and does not actually schedule computing resources such as worker nodes, but identifies that the execution plan needs to be added downstream of the upstream stage. Select the target execution path for actual execution.

In some embodiments, the execution plan referred to in step S611 may be described by a DAG, for example, one or more upstream stages of the DAG may have multiple candidate execution paths downstream. As an example, FIG. 6D exemplarily shows a schematic diagram of an execution plan carrying multiple execution paths. As shown in Figure 6D, there are two candidate execution paths downstream of M1: Path0 and Path1. In addition, a control node C8_1 is arranged downstream of M1, and the control node C8_1 communicates with Path0 and Path1, and can be used to select a target execution path actually executed downstream of M1 from Path0 and Path1.

In step S612, during the job execution process, the statistical information of the output data of the upstream stage is acquired.

In step S613, the control node selects a target execution path from the plurality of execution paths based on the statistical information.

During job execution, the master node can collect statistical information on the output data of each stage of the execution plan, including the statistical information on the output data of the upstream stage that carries multiple execution paths downstream. For example, as shown in FIG. 6D , after the execution of M1 is completed during the job execution process, the statistical information of the output data of M1 can be collected by the master node. Based on the statistical information of the output data of the upstream stage collected by the master node, the master node can select the actual execution target execution path from the multiple execution paths of the upstream stage based on the statistical information through the control node. For example, as shown in FIG. 6D , the control node C8_1 may select the target execution path actually executed downstream of M1 from Path0 and Path1 based on the statistical information of the output data of M1 .

In this embodiment of the present application, when an execution plan is generated, the upstream stage of the execution plan can have multiple candidate execution paths downstream, and during the job execution process, based on the statistical information of the output data of the upstream stage, multiple candidate execution paths can be selected from the upstream stage. Select the final execution target execution path from the execution path of . In this way, during the execution of the job, the final execution path can be selected according to the actual situation of the output data of the upstream stage, so that the selection of the execution path is more reasonable and accurate. Executing jobs based on reasonable and accurate execution paths can significantly improve the performance of distributed systems.

As an optional implementation, the candidate execution paths carried downstream by the upstream stage of the execution plan may include two execution paths, namely the first execution path and the second execution path. FIG. 7A exemplarily shows a flow chart of generating an execution plan carrying multiple execution paths. The method process can be implemented by the master node. Referring to FIG. 7A , the method flow may include the following steps.

In step S710, according to the estimated data size of the source data of the job, a physical plan with multiple candidate information is generated. The information includes first candidate information and second candidate information.

When the job is submitted, the optimizer can estimate the data size of the source data (for example, the source table) of the job. If the estimated data size of the source data is smaller than the preset first threshold value, the optimizer can trigger the conditional executionplan (Conditional execution plan), based on the tasks that the source data may use during job execution, the optimizer can generate multiple candidates (candidate information) for the source data, and a physical plan with the multiple candidates. In some embodiments, a task may include one or more operators, for example, the execution of a specific task is implemented through the processing of the operator.

In one example, assuming that the source data is the source table, the source table can use Broadcast Join and Sort Merge Join during the Join operation during job execution, so the optimizer can use Broadcast Join and Sort Merge Join that the source table may use during the Join operation. Sort Merge Join, generate two candidates for the source table, one candidate represents Broadcast Join, and the other candidate represents Sort Merge Join. In some embodiments, a physical plan can be thought of as a data structure of a tree of nodes.

In step S711, the physical plan is converted into an operator tree, the operator tree includes one or more original tasks of the physical plan, and one original task uses one or more operators.

In step S712, when the preset operator is traversed for the first time, according to the first candidate information corresponding to the first execution path, the task of the first execution path is added to the operator tree, and the new task is added to the The operator is connected with the relevant operator in the original task by data pipe (data pipe) to convert the first execution path in the operator tree.

In order to generate an execution plan with multiple execution paths, after converting the physical plan into an operator tree, the optimizer can convert multiple execution paths in the operator tree according to multiple candidate information. In some embodiments, the optimizer may traverse the operator tree multiple times, and preset a preset operator for performing path conversion. When the optimizer traverses the preset operator for the first time, it may traverse multiple The first execution path corresponding to the first candidate information in the candidate information is converted. In some embodiments, the plurality of candidate information includes at least first candidate information and second candidate information, wherein the first candidate information corresponds to the first execution path, the task corresponding to the first execution path can be recorded, and the second candidate information It may correspond to the second execution path, and the task corresponding to the second execution path may be recorded.

In some embodiments, when the optimizer traverses the preset operator for the first time, it may add a task of the first execution path to the operator tree according to the first candidate information, and the newly added task may include one or more tasks Operator, in order to enable the task of the first execution path to communicate with the original task of the operator tree, the optimizer can connect the operator of the new task with the related operator in the original task, and the related operator in the original task can be connected by pipeline. An operator can be considered as an operator in the original task that is related to the operator of the newly added task (for example, the operator in the original task that is upstream and downstream of the operator of the newly added task). Through the above processing, the optimizer can convert the first execution path in the algorithm tree.

In step S713, when the preset operator is traversed for the second time, according to the second candidate information corresponding to the second execution path, the task of the second execution path is added to the operator tree, and the new task is added to the The operator is connected with the relevant operator in the existing task by data pipe, so as to convert the second execution path in the operator tree.

In some embodiments, when the optimizer traverses the preset operator for the second time, it can convert the second execution path in the operator tree according to the second candidate information. The optimizer can add a task of the second execution path in the operator tree according to the second candidate information, and the newly added task can include one or more operators, so that the task of the second execution path can be combined with the operator tree. The optimizer can connect the operator of the new task with the related operator in the existing task through datapipe connection, and the related operator in the existing task can be considered as the connection between the existing task and the new task. Operator-related operators (for example, operators in the original task that are upstream and downstream of the operator of the newly added task). Through the above processing, the optimizer can convert the second execution path in the algorithm tree.

As an optional implementation, step S712 can be considered as an optional implementation of converting the first execution path based on the first candidate information in the algorithm tree, and step S713 can be considered as an optional implementation of converting the second execution path based on the second candidate information in the algorithm tree Choose to implement.

In step S714, control nodes are set upstream of the first execution path and the second execution path of the operator tree to obtain an execution plan.

After the transformation of the first execution path and the second execution path is completed in the operator tree, in order to enable the job execution process, after the execution of the upstream stages of the first execution path and the second execution path is completed, the first execution path can be executed from the first execution path. The optimizer can also set control nodes in the upstream of the first execution path and the second execution path of the operator subtree, so as to realize the first execution path and the second execution path in the job execution process through the control nodes. 2. Dynamic selection of execution paths. Through the above process, the embodiment of the present application can generate an execution plan that carries multiple execution paths.

In order to facilitate the description of the above process of generating an execution plan carrying multiple execution paths, the embodiment of the present application introduces the concept of a conditional execution plan: allowing the optimizer to generate based on multiple candidates (candidate information) when a job is submitted Execution plans with multiple execution paths. In the end, which candidate execution path is selected to be used is dynamically selected by the master node during the job execution process based on the statistical information of the output data generated by the upstream stage.

As an optional implementation, the optimizer can generate an execution plan with multiple execution paths through two links, Cost-based Optimization and Execution plan generation.

In the Cost-based Optimization section, taking the Join scenario as an example, the optimizer can estimate the data size of the Join's small table in memory through the source table statistics and other information in the Join's buildrule (construction rule). RowCount (number of rows)*AverageRowSize (average row size) estimates the data size of the small table in memory. Further, the optimizer can determine whether the data size of the small table in the memory is smaller than a preset first threshold value (the first threshold value can be preset and defined as threshold1). If the data size of the small table in memory is smaller than the preset first threshold value, the optimizer will trigger the conditional Map Join of the database (conditional map join), such as generating multiple candidates for Map Join with small table data , a candidate can represent a Join operator (such as Broadcast Join or Sort Merge Join) used by small tables. Based on the multiple candidates provided by conditional Map Join to realize subsequent dynamic decision-making, the optimizer can configure the first threshold value threshold1 of the Cost-based Optimization link to be relatively large (the default value is 512M), that is, as long as the looser value is used Probability, it is judged that a job may use Broadcast Join to generate a conditional Map Join plan, and the final selection of the execution path is handed over to the execution engine for dynamic selection during the job execution process.

The above conditional Map Join plan can be regarded as a physical plan (physical plan), rather than the final execution plan, the physical plan can be a RelNode tree (control node tree) structure, where RelNode represents a relational expression (relational expression). In the Cost-based Optimization link, the physical plan contains control nodes such as conditional Map Join, and the control node can contain multiple Paths (execution paths) based on multiple candidates to express the paths that may be selected for the Join algorithm, such as Execution paths corresponding to Broadcast Join and Sort Merge Join. No matter which Path is finally selected, the calculation and data of the small table will be shared among the multiple Paths. The cost defined by conditional Map Join in the cost model of the optimizer can be between the costs of Broadcast Join and Sort Merge Join, and the cost ratio of the two is determined according to the probability of the final selected Path.

In the Execution plan generation step, the optimizer converts the physical plan generated in the previous step into the final execution plan (execution plan), for example, converts the physical plan generated by the optimizer into a physical operator tree that can be understood by the runtime (runtime component). (Physical operator tree) and construct DAGtopology (topology). Therefore, the final execution plan includes the DAG composed of tasks and edges, as well as the operator tree (ie, the physical operator tree) inside each worker node. In general, the above process is a post-order traversal process of a physical plan, and dynamically divides tasks and adds edges during the traversal process. Different from ordinary query, Conditional Map Join needs to implement dynamic decision-making and path selection at runtime, so it is necessary to add control tasks and dependent edges (dependent edge) in the process of execution plan generation, so as to accurately describe the upstream and downstream work nodes through the dependent edge. Scheduling dependencies between (no data flow) and a pure control node. For example, connect a normal worker node containing a normally executing operator to a control node via a dependent edge. Through the above settings, the control node can select the subsequent execution path by judging the size of the small table after the actual operation of the DAG during the operation of the DAG.

7B, 7C, 7D and 7E exemplarily show the process of converting a physical plan into an execution plan. The master node may execute FIG. 7C after completing FIG. 7B , execute FIG. 7D after completing FIG. 7C , and execute FIG. 7E after completing FIG. 7D . Among them, the thick arrows in the figure refer to the data pipeline (data pipeline connection), the thin dashed arrow represents the control pipeline (control pipeline connection), and the thin solid arrow represents the runtime operator data flow (the data flow of the operator during operation). In an implementation example, when the master node traverses the physical plan, if it traverses to the conditional Map Join and is ready to transform multiple Paths included in the conditional Map Join, the master node can input the conditional Map Join into the relNode tree (control node tree) is converted into an operator tree (operator tree), and the operators in the operator tree can form the two tasks M1 and M2 shown in Figure 7B. For example, M1 uses TableScan1 (table scan 1) and Filter1 (filter 1). operator, M2 uses the three operators TableScan2 (table scan 2), Filter2 (filter 2) and StreamlineWrite1 (streamline write 1).

Assume that the multiple candidates contained in the conditional map join are two candidates, and one candidate corresponds to Path0 and the other candidate corresponds to Path1. Path0 can correspond to the Join implementation of Broadcast HashJoin, and Path1 can correspond to the Join implementation of Sort Merge Join. With reference to FIG. 7C , in the process of traversing the operator tree, if the master node traverses to filter1 of M1 for the first time, Path0 can be converted. Specifically, based on the task recorded by the candidate of Path0, the master node can add a task of Path0 to the operator tree, and connect the operator in the newly added task to M1 and M2 with a data pipe. As shown in Figure 7C, R3 is a new task and has two new operators, StreamlineRead1 (streamline read 1) and StreamlineWrite2 (streamline write 2). R1 adds StreamlineRead2 (streamline read). 2) and ConditionalMapJoin1. Among them, ConditionalMapJoin1 newly added in M1 is the operator id (operator identifier) of Broadcast Hash Join, and its operator type is HashJoin.

The above process is executed when traversing filter1 in M1 for the first time, so as to transform Path0 on the basis of the operator tree, so as to express the execution path of Path0 in the execution plan; in this process, the master node can The original operator of M1 is recorded for subsequent copying. From the four newly added operators, it can be inferred that R3 and M1 are currently being processed. At this time, they are recorded as a candidate Path0.

Then start converting Path1. Path1 and Path0 share StreamlineWrite and Filter, such as StreamlineWrite1 and Filter1. Based on the operator Filter1 shared by Path1 and Path0, when the main node traverses to Filter1 for the second time, Path1 can be transformed. Specifically, based on the task recorded by the candidate of Path1, the master node can add a task of Path1 to the operator tree of Path0 that has been transformed, and perform data analysis between the operator in the newly added task and the relevant operator in the current task. pipe connection. As shown in Figure 7D, M5 and J6 are newly added tasks. When adding a M5 task, you can copy the previously recorded operators of M1 (such as TableScan1 and Filter1 in Figure 7D) to M5, and at the same time in M5 Added StreamlineWrite3 operator in . When adding a J6 task, you can add three operators, StreamlineRead3 (streamline read 3), StreamlineRead4 (streamline read 4), and ConditionalMapJoin1 in J6, and realize the computation between J6 and M2 and M5. Sublevel data pipe.

As shown in Figure 7E, after the conversion of Path0 and Path1 is completed, a control node C8 needs to be created in the operator tree. There is only one operator in C8, that is, the Conditional operator. The size of the process to select the execution path for Join from Path0 and Path1.

In some further embodiments, after generating an execution plan with multiple execution paths, the master node can obtain the data volume of the output data of the upstream stage during the job execution process, and determine whether the data volume is less than the preset first Two threshold values, if yes, select the first execution path as the target execution path; if not, select the second execution path as the target execution path. Assuming that the data processed by the upstream stage is a small table, during the job execution process, if the task corresponding to the small table is executed, the control node can judge whether the data output result of the small table is smaller than the preset value according to the actual execution size of the small table. The second threshold value (defined as threshold2), if the execution output result of the small table is less than the second threshold value, the execution path of Broadcast Join is selected (that is, the first execution path is the execution path for executing Broadcast Join); If the execution output result is greater than the second threshold value, the execution path of Sort Merge Join is selected (that is, the second execution path is the execution path for executing Sort Merge Join). It should be noted that threshold2 considers the memory size occupied by the small table data loaded into the hash table in the actual operation, and its default value can be the same as that of threshold1 (for example, 512M). Of course, the embodiment of this application can also set threshold2 Different from the value of threshold1. In addition, it should be noted that the conditional Join plan can be triggered only when the estimated size of the small table meets the threshold1. During the job operation, when the master node collects the real size of the small table, it uses the threshold2 to decide the final join of the small table. Execution path; so threshold1 is used to make decisions in the optimizer stage, and threshold2 is used to make decisions in the DAG running stage. Through the adjustment of the two values of threshold1 and threshold2, as well as the cooperation between the optimizer and DAG, the entire conditional is completed. generation and selection of the final execution plan.

In terms of dynamic selection of execution paths, as shown in Figure 6D, when the job is initially submitted, because the final execution path has not been confirmed, the submitted DAG contains two possible execution paths (for example, the dotted line in Figure 6D shows Two execution paths Path0 and Path1). At the same time, a new control node (C8_1) is added to the DAG. This control node has only logical control meaning and will not pull up any working nodes. The control node chooses to execute Path0 or Path1 according to the output of M1. For example, during job execution, after M1 reads the candidate output data of the small table and processes it, the actual output data volume will be collected by the master node, and the control node will execute the execution path based on the actual output data volume of the small table. decision choice. If Path0 is selected to be executed, the complete execution plan may be as shown in FIG. 7F , and if Path1 is selected to be executed, the complete execution plan may be as shown in FIG. 7G .

In this embodiment of the present application, the execution plan can be adjusted before the job is executed, so that the upstream stage of the execution plan can have multiple candidate execution paths downstream, and during the job execution process, based on the statistical information of the output data of the upstream stage, the The final execution target execution path is selected from a plurality of candidate execution paths. In this way, during the execution of the job, the final execution path can be selected according to the actual situation of the output data of the upstream stage, so that the selection of the execution path is more reasonable and accurate, so that the job can be executed based on a reasonable and accurate execution path. It can significantly improve the performance of distributed systems.

With the wide application of deep learning (Deep-Learning), distributed systems need to meet more and more processing requirements for deep learning jobs, and a variety of execution engines suitable for deep learning jobs have emerged. However, for deep learning jobs, distributed systems still have various shortcomings in scheduling and execution. For example, for the native logic of a deep learning system (such as Tensorflow), the scheduling and execution of the native logic completely depends on the external system and is not configured in the execution engine of the distributed system. Taking the Parameter Server (parameter server, PS) architecture as an example, the working nodes of a distributed system can be divided into two categories: PS nodes and Worker (worker) nodes, where the PS nodes store deep learning parameters (such as parameters of a deep learning model) , and the Worker node is used to calculate the gradient of the deep learning parameters. In each iteration of deep learning, the Worker node obtains the deep learning parameters from the PS node, and then returns the calculated gradient to the PS node. The PS node aggregates the gradients returned from the Worker node and updates the deep learning parameters, and the PS node will update the The deep learning parameters are broadcast to the Worker node, and they are continuously executed in this way, so as to realize the iterative adjustment of the deep learning parameters. In the PS architecture of the distributed system, the PS nodes and the Worker nodes have the following characteristics in the running process:

The responsibilities performed by the PS node and the Worker node are significantly different, and correspond to different stages of the execution plan, among which. The stage corresponding to the PS node can be called the PS stage (parameter server execution stage), and the stage corresponding to the Worker node can be called the Worker stage (worker execution stage);

As the serving entity of Parameter, the PS node can run independently;

When the Worker node uses and updates the Parameter (parameter), it needs the PS node to run effectively before it runs, and it needs to continuously exchange data with the PS node during the running process.

The above characteristics are difficult to describe in many distributed execution frameworks: although there is a scheduling dependency between PS nodes and Worker nodes, because PS nodes and Worker nodes can run at the same time, this dependency cannot be mapped to In the logic of scheduling downstream nodes after the upstream node has finished running. Based on this, in many external systems, the PS node and the Worker node can only be scheduled and run separately in the execution plan corresponding to two isolated and unconnected stages, which may cause the Worker node to be scheduled before the PS node is scheduled. It is idling, resulting in wasted resources of the Worker node. In addition, many basic functions and dynamic functions cannot be realized due to the lack of description of the relationship between PS nodes and Worker nodes in different stages.

In the field of deep learning, resources (especially GPU resources) used by deep learning jobs are generally directly open to end users to specify. For example, the end user specifies the worker node concurrency of the deep learning job, the resource size used by each worker node (such as the number of GPUs used by a worker node, etc.). However, it is difficult for users to choose appropriate resources. In order to ensure sufficient resources for deep learning operations, users often apply for resources excessively, resulting in a waste of resources. For example, in order to ensure that the GPU usage of deep learning jobs is guaranteed, users will apply for multiple GPU cards for each worker node when the actual GPU resources used cannot be accurately predicted. However, during the actual execution of deep learning jobs, it is possible that Only 25% of it is utilized, which results in wasted and idle GPU resources. A cumulative effect caused by this situation is that a large number of GPU resources on the distributed system are reserved due to the user's application, and even the GPU resources requested by the user exceed the total GPU resources of the distributed system, and the actual GPU resources of the distributed system appear. Utilization is low, while other jobs need to queue up to use GPU resources. On the other hand, many deep learning jobs are particularly sensitive to the use of resources (especially GPU resources). If you blindly reduce the upper limit of resources that users are allowed to apply for, the execution performance of deep learning jobs may be degraded.

Based on the above situation, how to ensure the resource application accuracy of deep learning jobs and improve the resource utilization of distributed systems, so as to save the computing resources of distributed systems and ensure that the execution performance of deep learning jobs is not affected, has become an urgent solution. The problem.

In order to solve the above problems, based on the fact that the vertices and connecting edges in the DAG can correspond to different logical and physical attributes, the embodiment of the present application introduces the physical attributes of sequential edges (sequential edges) and concurrent edges (parallel edges) on the physical attributes of the connecting edges. properties, and decouples sequential and parallel edges from data transfer. The parallel edge describes that the worker nodes of the upstream and downstream stages connected by the parallel edge can be running at the same time, but there is still a sequence in the scheduling, and the timing of the scheduling can be customized. For example, the worker nodes of the upstream and downstream stages connected by parallel edges can be scheduled synchronously, or the worker nodes of the downstream stage can be scheduled after the worker nodes of the upstream stage run, or the worker nodes of the downstream stage can run to the worker nodes of the upstream stage. After a certain period of time, it will be triggered by events and scheduled. The sequence edge describes the work nodes of the downstream stage connected by the sequence edge, and it needs to wait until all or part of the work nodes of the upstream stage are executed before they can be scheduled to run.

Based on parallel edges, the relationship between PS nodes and Worker nodes of deep learning jobs can be described more accurately. FIG. 8A exemplarily shows an example diagram of a PS stage (parameter server execution stage) and a Worker stage (worker execution node) connected by parallel edges. As shown in Figure 8A, the PS stage is the stage corresponding to the PS node in the execution plan, and the Worker stage is the stage corresponding to the Worker node in the execution plan. In the execution plan, the PS stage is connected to the Worker stage through a parallel edge, and the parallel edge is connected by the PS stage Enter Worker stage. Thus, the Worker node and the Worker node can be running at the same time, and the scheduling timing can be customized. In some embodiments, the following table 1 exemplarily shows the types of scheduling opportunities (called scheduling types) of downstream nodes connected by parallel edges, for reference, wherein the upstream nodes can be considered as the upstream stages connected by parallel edges Corresponding worker nodes (eg, PS nodes corresponding to PS stage shown in FIG. 8A ), downstream nodes can be considered as worker nodes corresponding to downstream stages connected by parallel edges (eg, worker nodes corresponding to Worker stage shown in FIG. 8A ).

Table 1

Based on the sequential edges of connecting edges, the physical properties of parallel edges, and the scheduling types of downstream nodes, the embodiments of the present application can completely describe various complex DAG logics, thereby supporting various loads, such as batch jobs, streaming type jobs, near-real-time/quasi-real-time jobs, deep learning jobs, etc.

For deep learning jobs, take PS jobs as an example, because the Worker node can use and update Parameter, and the Worker node can run after the PS node runs, so in the execution plan of the deep learning job, the connection between the PS stage and the Worker stage The edge is a parallel edge, and the parallel edge is input to the Worker stage from the PS stage; that is, in the execution plan, the Worker stage acts as a direct downstream stage of the PS stage and is connected by the parallel edge. Since the Worker only makes sense to run after the PS starts running, the scheduling type of the downstream Worker node is SOURCE_TASK_STARTED, that is, after the instance of the upstream PS node is started, the downstream Worker node starts to be scheduled. In terms of data generation, since the data on the PS node is only valid when it is in the running state, and the upstream data source does not perceive the downstream state, the data source type is EPHEMERAL_STATELESS (ephemeral stateless); and in terms of data transmission type, Since the data transmission between the PS node and the Worker node may not be perceived by the execution framework, the data transmission type is NONE (null). So the connecting edge connecting PS stage and Worker stage can be described as {CONCURRENT, EPHEMERAL_STATELESS, NONE, SOURCE_TASK_PROGRESS}.

Based on the above description, during the execution of a deep learning job in this embodiment of the present application, the PS node and the Worker node can run at the same time, and the Worker node can only schedule resources for data processing after the PS node is running. This description of the execution plan for a deep learning job makes it possible to dynamically adjust the job's running process. Specifically, the execution engine of deep learning requires end users to provide a large number of configuration parameters for the execution plan, such as stage concurrency, required resource size and type, distribution strategy, etc. These configuration parameters are extremely difficult to provide by end users . Based on this, the embodiment of the present application introduces a Resource Optimization (resource optimization) node into the execution engine of the distributed system. The resource optimization node, as a control node for job execution, can coordinate and dynamically adjust resource-related requests.

In an example, taking a PS job as an example, this embodiment of the present application adds a new resource optimization node other than the PS node and the Worker node. The resource optimization node is responsible for determining how to dynamically adjust the resources of the worker node according to certain rules. On this basis, in addition to setting the parallel edge input from the PS stage to the Worker stage in the execution plan, it is also necessary to add the Resource Optimization stage (resource optimization execution stage) corresponding to the Resource Optimization node in the execution plan, and the Resource Optimization stage Enter the parallel edges of the Worker stage. The connecting edge connecting the Resource Optimization stage and the Worker stage can be described as {CONCURRENT, EPHEMERAL_STATELESS, NONE, SOURCE_TASK_PROGRESS}. Different from the description of the connecting edge connecting the PS stage and the Worker stage, the connecting edge connecting the Resource Optimization stage and the Worker stage adopts the scheduling type whose downstream node is SOURCE_TASK_PROGRESS. That is, the Worker node needs to be scheduled after the execution progress of the upstream Resource Optimization node reaches a certain threshold.

During job execution, the PS node and the Resource Optimization node will be started first. When the instance of the PS node is started, the downstream Worker node will be notified for scheduling, and when the instance progress of the Resource Optimization node reaches the threshold, the downstream Worker node will be notified for scheduling. After receiving the above two notifications, the Worker node can start the corresponding instance after updating the scheduled resource based on the notification of the Resource Optimization node (the resource changes dynamically based on the control of the Resource Optimization node). In some embodiments, the Resource Optimization node can dynamically adjust the following resources of the Worker node: node concurrency, and usage requirements of resources such as GPU, CPU, and MEMORY of the node.

As an optional implementation, FIG. 8B shows yet another flowchart of the distributed job adjustment method provided by the embodiment of the present application. This method can be implemented by the master node. Referring to FIG. 8B , the method flow may include the following steps.

In step S810, the deep learning assignment submitted by the user is acquired.

In step S811, an execution plan of the deep learning job is generated, where the execution plan includes multiple stages, and the multiple stages include: a Worker stage and a Resource Optimization stage.

In some embodiments, the Resource Optimization stage feeds the Worker stage through parallel edges.

In some further embodiments, the plurality of stages may further include: a PS stage. Among them, the PS stage and the Resource Optimization stage are respectively input to the Worker stage through parallel edges.

In this embodiment of the present application, a deep learning job (for example, a PS job) can be described as a structure of PS, Worker, and Resource Optimization. That is to say, when generating an execution plan of a deep learning job (the execution plan can be described by DAG), the execution plan can include PS stage, Worker stage, and Resource Optimization stage. And in the DAG diagram, the PS stage is input to the Worker stage from the parallel side, and the Resource Optimization stage is also input to the Worker stage from the parallel side. In some further embodiments, for the description of the parallel edges of the PS stage and the Worker stage, and the description of the parallel edges of the Resource Optimization stage and the Worker stage, reference may be made to the corresponding descriptions above, which will not be expanded here.

In step S812, during the execution of the deep learning job, the Resource Optimization node corresponding to the Resource Optimization stage is scheduled, and the resource information currently adapted to the Worker stage is determined through the Resource Optimization node.

In this embodiment of the present application, the Resource Optimization node may be scheduled before the Worker node. For example, when the instance progress of the Resource Optimization node reaches a threshold, the Worker node is notified to perform scheduling. In some embodiments, after the Resource Optimization node is scheduled, the Resource Optimization node may determine resource information currently adapted to the Worker stage. In some embodiments, the Resource Optimization node may determine historically used resource information that matches the current execution state of the deep learning job, and uses the historically used resource information that matches the current execution state as the resource information currently adapted to the Worker stage. resource information.

As an optional implementation, the historically used resource information matching the current execution state of the deep learning job may include: historically used resource information of a historical execution state that is the same as or similar to the current execution state.

In some embodiments, the Resource Optimization node can determine resource information suitable for the Worker stage from a historical database based on the current execution state of the deep learning job, and the historical database can record the resource information actually used by the historical deep learning job in each historical execution state , that is to say, the historical database can record the resource information actually used by the deep learning jobs whose historical execution ends in each execution state. Therefore, the ResourceOptimization node can determine resource information suitable for the current execution state of the deep learning job based on the records in the historical database, and configure resources for the worker node.

In some embodiments, the Resource Optimization node may, based on the current execution state of the deep learning job, search a historical database for a historical execution state similar to or the same as the current execution state, and use the resource information actually used by the searched historical execution state, Determined as the resource information currently adapted to the Worker stage. As an optional implementation, the Resource Optimization node can first search the historical database for the same historical execution state as the current execution state. If found, the resource information corresponding to the same historical execution state recorded in the historical database will be used as the current adaptation. Resource information in the Worker stage; if not found, search for a historical execution state similar to the current execution state (for example, find the historical execution state with the smallest difference from the current execution state), and record the similar historical execution state in the historical database. The resource information corresponding to the state is used as the resource information currently adapted to the Worker stage. In some embodiments, the current execution status of the deep learning job is, for example, the current learning mode of the deep learning job, the characteristics of the current input data (eg, the data volume of the current training data), the current remaining number of parameter iterations, and the like.

In step S813, the resource information is configured for the Worker stage through the Resource Optimization node.

After determining the resource information currently suitable for the Worker stage through the Resource Optimization node (for example, the historically used resource information matching the current execution state), the master node can configure the resource for the Worker stage in the execution plan through the ResourceOptimization node information, so that the resource information configured by the Worker stage is adapted to the current execution state of the deep learning job. Furthermore, when the Worker node is scheduled, the Worker node can schedule resources to be used based on the resource information, so that the Worker node executes tasks with reasonable resources and improves the resource utilization rate of the Worker node.

In some embodiments, the Worker node needs to be scheduled after the instance of the PS node is started and when the progress of the instance of the ResourceOptimization node reaches a threshold.

In some embodiments, if the user specifies the resource information of the Worker stage in advance, the ResourceOptimization node can adjust the resource information specified by the user based on the determined resource information currently suitable for the Worker stage, so as to prevent the user from specifying too much of the Worker stage. resources that lead to idle resources. For example, the user specifies that the Worker node uses 2 CPU cores (cores) and 1 GPU core (core) resource units. If the Resource Optimization node determines based on the current execution state of the deep learning job, the Worker node does not actually need to use so many resources. , the Resource Optimization node can adjust the number of CPU cores and GPUcores specified by the user in advance. For example, it can be adjusted to use 1 CPU core and half a GPU core for the Worker node, so as to avoid the excessive number of CPU cores specified by the user for the Worker node. and GPU core, resulting in idle waste of CPU and GPU resources.

As some implementation examples, Figure 8C shows an example of the Resource Optimization node adjusting the resources of the Worker node. As shown in Figure 8C, the planned resources of the Worker node are 200 CPU cores and 100 GPU cores (which can be specified by the user). During the execution of the deep learning job, the Resource Optimization node can be based on the current execution status of the job, from the historical Find the resource information actually used by the similar or the same historical execution state in the database. For example, if the current execution state of the job is similar or the same historical execution state actually uses 100 CPU cores and 50 GPU cores, the Resource Optimization node can convert the Worker node The configured number of CPU cores is adjusted to 100, and the number of GPU cores is adjusted to 50, so that worker nodes can ensure the specific execution of deep learning jobs by configuring lower resources.

In some further embodiments, the Resource Optimization node can preset multiple resource demand schemes of the Worker node, and the Resource Optimization node can select from multiple resource demand schemes according to the current execution status of the deep learning job, which is related to the current execution status. Match the resource requirement scheme, and configure resource information for the Worker stage with the selected resource requirement scheme. The resource requirement scheme that matches the current execution state can be considered as the scheme that can satisfy the job execution of the current execution state and has the least amount of resource requirements.

The embodiments of the present application can configure resources with higher precision for the worker nodes based on the current execution state of the deep learning jobs, which can save the computing resources of the distributed system on the premise that the execution performance of the deep learning jobs executed by the worker nodes is not affected. Improve the resource utilization of distributed systems. The embodiments of the present application can greatly improve the utilization rate of resources (especially GPU resources) of deep learning jobs after the large-scale production cluster goes online, without affecting the actual training time of users for deep learning. The throughput rate of incoming jobs is greatly improved, and the situation of user job queuing can be greatly alleviated.

The embodiment of the present application can describe and expand the execution plan of Map-Reduce, offline job running mode, etc., so that the execution plan can accurately describe the deep learning operation including PS. This way of describing the execution plan can avoid the idle waiting of resources of the worker nodes, and can provide a more accurate job execution control flow. In addition, the embodiments of the present application can dynamically select and adjust resources such as GPUs that are actually used by deep learning operations during the operation of the job, so as to ensure that the use of resources can be adapted according to the actual operation requirements of the job and the characteristics of the algorithm, ensuring efficient operation. resource usage, thus avoiding the contradiction between excessive resource application and low actual resource utilization in a large-scale multi-user distributed system.

The embodiment of the present application further provides a master node, where the master node can be configured to execute the distributed job adjustment method provided by the embodiment of the present application.

The embodiment of the present application further provides a distributed system, the structure of the distributed system may be combined with the description in the corresponding part above, and the distributed system may include the above-mentioned master node.

The embodiment of the present application further provides a physical machine, and the physical machine can be set with the master node provided by the embodiment of the present application. As an optional implementation, FIG. 9 shows a structural block diagram of a physical machine. As shown in FIG. 9 , the physical machine may include: at least one processor 1 , at least one communication interface 2 , at least one memory 3 and at least one communication bus 4 . In the embodiment of the present application, the number of the processor 1 , the communication interface 2 , the memory 3 , and the communication bus 4 is at least one, and the processor 1 , the communication interface 2 , and the memory 3 communicate with each other through the communication bus 4 . Optionally, the communication interface 2 may be an interface of a communication module used for network communication. Optionally, the processor 1 may be a CPU (central processing unit), a GPU (Graphics Processing Unit, graphics processing unit), an NPU (embedded neural network processor), an FPGA (Field Programmable GateArray, field programmable logic gate array), A TPU (Tensor Processing Unit), an AI chip, an Application Specific Integrated Circuit (ASIC), or one or more integrated circuits configured to implement the embodiments of the present application, or the like. The memory 3 may include high-speed RAM memory, and may also include non-volatile memory, such as at least one disk memory. The memory 3 stores one or more computer-executable instructions, and the processor 1 invokes the one or more computer-executable instructions to execute the distributed job adjustment method provided by the embodiment of the present application.

Embodiments of the present application further provide a storage medium, where the storage medium can store one or more computer-executable instructions. When the one or more computer-executable instructions are executed, the distributed job adjustment as provided by the embodiments of the present application is implemented. method.

The embodiment of the present application further provides a computer program, and the computer program is used to execute the distributed job adjustment method provided by the embodiment of the present application.

The multiple embodiments provided by the embodiments of the present application have been described above, and the optional modes introduced by the embodiments can be combined and cross-referenced without conflict, thereby extending a variety of possible embodiments. All of these can be considered as embodiments disclosed and disclosed in the embodiments of the present application.

Although the embodiments of the present application are disclosed as above, the present application is not limited thereto. Any person skilled in the art can make various changes and modifications without departing from the spirit and scope of the present application. Therefore, the protection scope of the present application should be based on the scope defined by the claims.

Claims (10)

1. A distributed job adjustment method, comprising:

obtaining deep learning operation;

generating an execution plan for the deep learning job, the execution plan including a plurality of execution phases including: a working machine execution phase and a resource optimization execution phase; the working device execution phase is used for calculating the gradient of the deep learning parameter;

in the execution process of the deep learning operation, scheduling a resource optimization node corresponding to the resource optimization execution stage, and determining the resource information which is used historically and is matched with the current execution state of the deep learning operation through the resource optimization node;

and configuring the resource information for the execution stage of the working machine through a resource optimization node.

2. The method of claim 1, wherein the execution plan is described by a DAG; the plurality of execution phases further comprises: a parameter server execution phase; the parameter server execution stage inputs the working machine execution stage through a parallel edge, the resource optimization execution stage inputs the working machine execution stage through a parallel edge, and working nodes of an upstream execution stage and a downstream execution stage connected by the parallel edge can be in a running state at the same time.

3. The method of claim 2, wherein the downstream worker node schedules when an instance of the upstream parameter server node is started and the execution progress of the upstream resource optimization node reaches a certain threshold; the worker nodes are corresponding to the worker execution stage, the parameter server nodes are corresponding to the parameter server execution stage, and the resource optimization nodes are corresponding to the resource optimization execution stage.

4. The method of claim 1, wherein the determining, by the resource optimization node, historically used resource information that matches a current execution state of a deep learning job comprises:

based on the current execution state of the deep learning operation, searching a historical execution state similar to or the same as the current execution state from a historical database, and determining the resource information actually used by the searched historical execution state as the resource information currently adapted to the execution stage of the working machine; wherein the history database records the resource information actually used by the history deep learning operation in each history execution state.

5. The method of claim 3, further comprising:

and when the worker node schedules, the worker node schedules the used resources based on the resource information.

6. The method of claim 1, further comprising:

and if the user specifies the resource information of the execution stage of the working device in advance, the resource optimization node adjusts the resource information specified by the user based on the determined resource information.

7. A master node, wherein the master node is configured to perform the distributed job adjustment method of any one of claims 1-6.

8. A distributed system, wherein the distributed system comprises a master node and a plurality of worker nodes, the master node being the master node of claim 7.

9. A physical machine, wherein the physical machine comprises at least one memory and at least one processor, the memory storing one or more computer-executable instructions that the processor invokes to perform the distributed job adjustment method of any of claims 1-6.

10. A storage medium, wherein the storage medium stores one or more computer-executable instructions that, when executed, implement a distributed job adjustment method as recited in any of claims 1-6.

Images