CN110287245A - Method and system for scheduling and executing distributed ETL (extract transform load) tasks - Google Patents

Abstract

The embodiment of the invention provides a method and a system for scheduling and executing distributed ETL tasks, which extract the association between an entity and an affiliated table, the association between the entity and a dimension table and the one-to-many association between the entity and the entity involved in the ETL task from an acquired target table contained in the ETL task to be scheduled and executed; determining the scheduling priority of the ETL task based on the preset weight for each association and the number of each association in the ETL task; and distributing each ETL task to each execution node according to the sequence from high to low of the scheduling priority. In the technical scheme of the embodiment of the invention, the ETL tasks are distributed to the execution nodes according to different weights based on factors such as the complexity of the service corresponding to the ETL tasks, the importance degree of the service data to be integrated and the like, so that the timeliness of core data loading and the load balance among the nodes are met, and the efficiency of data integration and the utilization rate of resources are improved.

Description

Method and system for scheduling and executing distributed ETL tasks

technical field

The invention relates to a data warehouse, in particular to a method and system for ETL task scheduling and execution.

Background technique

At present, data extraction, transformation and loading technology (Extract-Transform-Load, ETL) is one of the key steps in building a data warehouse in a big data environment. It is to integrate scattered and heterogeneous data into a unified standard library process. The steps of data extraction, transformation, and loading can be combined into a schedulable ETL script job (also called an ETL task). In a big data environment, dozens or even tens of thousands of ETL tasks often need to be executed. How to efficiently schedule these tasks is an important part of building a data warehouse. At present, the distributed cluster scheduling scheme is mainly used for ETL task scheduling, and scheduling algorithms such as polling algorithm, first-come-first-serving algorithm, and Min-Min algorithm are used to distribute ETL tasks to each execution node in the cluster. However, due to the different execution time of each ETL task, the different amount of data contained in the task, and the current load of each execution node, it is easy to cause problems such as unbalanced cluster resource load and low resource utilization, resulting in low data integration efficiency.

Contents of the invention

According to the research of the inventors, it is found that during data integration, the business and related business data involved in different ETL tasks have different importance. directly affect the efficiency of data integration. However, the existing ETL task scheduling method does not consider the complexity of the business corresponding to the ETL task and the importance of the business data to be integrated. Therefore, the purpose of the embodiments of the present invention is to overcome the above-mentioned defects in the prior art, and provide a new method and system for scheduling and executing distributed ETL tasks.

The above-mentioned purpose is achieved through the following technical solutions:

According to the first aspect of the embodiments of the present invention, there is provided a method for scheduling execution of distributed ETL tasks, the method includes: for each acquired ETL task to be scheduled for execution, based on the target of data loading in the ETL task table, to extract the association between the entity and the subsidiary table involved in the ETL task, the association between the entity and the dimension table, and the one-to-many association between the entity and the entity; based on the weight preset for each association and each The number associated with the ETL task determines the scheduling priority of the ETL task; and each ETL task is allocated to each execution node according to the order of scheduling priority from high to low.

In some embodiments of the present invention, the method may also include querying performance indicators of each execution node before allocating ETL tasks; and determining the current load of each execution node according to the obtained performance indicators of each execution node, according to the execution The current load of the node is from low to high to select the corresponding execution node to distribute the ETL task.

In some embodiments of the present invention, the scheduling priority of the ETL task can be calculated by the following formula:

Among them, Wl1 represents the weight of the association between the entity and the subsidiary table; Wl2 represents the weight of the association between the entity and the dimension table; Wl3 represents the weight of the association between the entity and the entity; where ni represents the i-th value that appears in the ETL task number of associations.

In some embodiments of the present invention, said assigning each ETL task to each execution node may include:

a) Count the data volume of each ETL task to be scheduled and executed;

b) Count the total amount of data of all ETL tasks on each execution node;

c) Select the ETL task corresponding to the maximum amount of data from the ETL tasks to be scheduled and executed;

d) Select the execution node with the smallest amount of data and currently not assigned ETL tasks;

e) assign the selected ETL task to the selected execution node, and mark the execution node as assigned;

f) Repeat steps c)-e) until the ETL tasks to be scheduled for execution are allocated or until all execution nodes are marked as allocated;

g) Detect whether there are ETL tasks to be scheduled for execution, and if so, re-mark all execution nodes as unallocated, and repeat steps c)-g) until the ETL tasks to be scheduled for execution are allocated.

In some embodiments of the present invention, the method may further include: in response to the execution node receiving a new ETL task, storing the ETL task to be executed in the task cache queue, and recording the arrival time of the ETL; based on the The amount of data in the ETL task is used to estimate the execution time of the ETL task; in response to the execution of the current task of the execution node, for each ETL task to be executed, the ETL task is determined according to the waiting time of the ETL task and the estimated execution time. The execution priority of the ETL task; and the ETL task with the highest execution priority is selected from the ETL tasks to be executed for execution.

In some embodiments of the present invention, estimating the execution time of the ETL task based on the amount of data in the ETL task may include: determining the amount of data in the ETL task; Among the ETL tasks, a batch of ETL tasks with similar data volume to the ETL tasks to be executed is selected; the execution time of these ETL tasks is averaged, and the obtained average value is used as the estimated execution of the ETL task time.

In some embodiments of the present invention, the execution priority of the ETL task can be determined using the following formula:

Where EP _i represents the execution priority of the i-th ETL task ei; Tei represents the execution time of the ETL task ei; Twi represents the waiting time of the ETL task ei, which is equal to the current time minus the time when the ETL task arrives at the execution node.

According to the second aspect of the embodiments of the present invention, there is also provided a system for scheduling and executing distributed ETL tasks, including a scheduler and a plurality of executors, and the scheduler is used to allocate one or more ETL tasks to be scheduled and executed To multiple executors, the executors are used to execute the received ETL tasks. The scheduler includes a relationship analysis module, a priority determination module and a scheduling module. The relationship analysis module is used for each ETL task to be scheduled and executed, based on the target table of data loading in the ETL task, to extract the association between the entity and the subsidiary table involved in the ETL task, and the relationship between the entity and the dimension table Associations, one-to-many associations between entities. The priority determination module is used to determine the scheduling priority of the ETL task based on the preset weight for each type of association and the number of each type of association in the ETL task. The scheduling module is used to assign each ETL task to each executor in descending order of scheduling priority.

In some embodiments of the present invention, the scheduler may also include a load monitoring module, which is used to query the performance indicators of each executor, and determine the current load of each executor according to the obtained performance indicators of each executor; and The scheduling module can also be configured to select corresponding executors from low to high according to the current load of the executors to allocate ETL tasks.

In some embodiments of the present invention, the executor may be configured to: in response to receiving a new ETL task, store the ETL task to be executed in the task cache queue, and record the arrival time of the ETL; The amount of data in the ETL task is used to estimate the execution time of the ETL task; in response to the completion of the current task, for each ETL task to be executed, the ETL task is determined according to the waiting time of the ETL task and the estimated execution time the execution priority; and select the ETL task with the highest execution priority from the ETL tasks to be executed for execution.

The technical solutions of the embodiments of the present invention may include the following beneficial effects:

According to the complexity of the business corresponding to the ETL task, the importance of the business data to be integrated, node performance and other factors, the ETL task is allocated among the nodes, and it can also be based on the execution time of the ETL task and the pending processing time on each execution node. Adjusting the execution order of ETL tasks such as data volume not only meets the timeliness of core data loading and load balancing among execution nodes, but also improves the efficiency of data integration and resource utilization as a whole.

It is to be understood that both the foregoing general description and the following detailed description are exemplary and explanatory only and are not restrictive of the invention.

Description of drawings

The accompanying drawings, which are incorporated in and constitute a part of this specification, illustrate embodiments consistent with the invention and together with the description serve to explain the principles of the invention. Apparently, the drawings in the following description are only some embodiments of the present invention, and those skilled in the art can obtain other drawings according to these drawings without creative efforts. In the attached picture:

FIG. 1 shows a schematic flowchart of a method for scheduling and executing distributed ETL tasks according to an embodiment of the present invention.

Fig. 2 shows a schematic diagram of a process of determining weights of ETL tasks according to an embodiment of the present invention.

FIG. 3 shows a schematic diagram of an ETL task execution process on an execution node according to an embodiment of the present invention.

FIG. 4 shows a schematic structural diagram of a system for scheduling and executing distributed ETL tasks according to an embodiment of the present invention.

Detailed ways

In order to make the object, technical solution and advantages of the present invention clearer, the present invention will be further described in detail below through specific embodiments in conjunction with the accompanying drawings. It should be understood that the described embodiments are some, not all, embodiments of the present invention. Based on the embodiments of the present invention, all other embodiments obtained by persons of ordinary skill in the art without making creative efforts belong to the protection scope of the present invention.

Furthermore, the described features, structures, or characteristics may be combined in any suitable manner in one or more embodiments. In the following description, numerous specific details are provided in order to give a thorough understanding of embodiments of the invention. However, those skilled in the art will appreciate that the technical solutions of the present invention may be practiced without one or more of the specific details, or other methods, components, means, steps, etc. may be employed. In other instances, well-known methods, apparatus, implementations, or operations have not been shown or described in detail to avoid obscuring aspects of the invention.

The block diagrams shown in the drawings are merely functional entities and do not necessarily correspond to physically separate entities. That is, these functional entities may be implemented in software, or in one or more hardware modules or integrated circuits, or in different networks and/or processor devices and/or microcontroller devices entity.

The flow charts shown in the drawings are only exemplary illustrations, and do not necessarily include all contents and operations/steps, nor must they be performed in the order described. For example, some operations/steps can be decomposed, and some operations/steps can be combined or partly combined, so the actual order of execution may be changed according to the actual situation.

FIG. 1 shows a schematic flowchart of a method for scheduling and executing distributed ETL tasks according to an embodiment of the present invention. As shown in Figure 1, the method mainly includes: step S101) For each ETL task to be scheduled and executed, based on the target table of data loading in the ETL task, extract the relationship between the entity involved in the ETL task and the subsidiary table associations, associations between entities and dimension tables, and associations between entities; step S102) determine the scheduling of the ETL task based on the weight preset for each association and the number of each association in the ETL task Priority; Step S103) Allocating each ETL task to the execution node executing the ETL task in descending order of scheduling priority.

More specifically, in step S101), firstly, a plurality of ETL tasks to be scheduled and executed may be acquired from the ETL task resource library. After the ETL task is constructed, the ETL task-related information is usually stored in the ETL task repository in the form of metadata. These metadata include metadata description information such as the name, file name, directory, status, description, and extended description of the ETL task. The status of the ETL task can be used to indicate whether the ETL task has been scheduled for execution, and its specific value can be set or changed according to the actual scheduling execution situation. For example, the status of the ETL task that has been scheduled for execution can usually be set to 1, but not yet The status of ETL tasks scheduled for execution can usually be set to 0. In one embodiment, the ETL tasks to be scheduled for execution can be obtained from the ETL task resource library according to the status and creation time of each ETL task. Whether the ETL task is waiting to be scheduled can be known through the state of the ETL task, and the waiting time of the ETL task can be obtained according to the creation time of the ETL task. In this way, each time scheduling is performed, a batch of ETL tasks that have not been scheduled for execution can be selected from the ETL task resource library according to the waiting time. The acquisition of ETL tasks can be based on a request-response mechanism or periodically. For example, the ETL task resource library may be read periodically to extract a batch of ETL tasks to be scheduled for execution. The cycle can be set or changed according to actual conditions, for example, it can be set to 2 hours, 1 hour, 0.5 hours, 10 minutes and so on.

In the data warehouse, the description of entities and the relationship between entities are usually reflected in various forms. When using ETL tasks to extract, transform and load data, it mainly extracts the required data from various distributed data sources, converts them and loads them into the set target table. Each ETL task usually contains one or more target tables, such as a target table that describes entities and their attributes, a target table that describes one-to-many relationships between entities, and a many-to-many relationship between entities The target table (may also be referred to as an attached table) for the description. In addition, in the data warehouse, all possible values of the attributes of the entity are usually stored in the dimension table, so when the relevant data of the specified entity is loaded in the ETL task, one or more dimension tables associated with the entity are usually Set as the target table to load. ETL tasks related to core business often involve more types of entities, and the relationships between entities are also more complex and diverse. In an embodiment of the present invention, the importance of the business corresponding to the ETL task is measured by the entities involved in the ETL task and various association relationships between entities and entities, and thus the scheduling priority of the ETL task is set (also can be called the weight).

In step S101), after obtaining the ETL task to be scheduled and executed, based on the target table of data loading in each ETL task, the association between the entity and the subsidiary table involved in the ETL task, and the relationship between the entity and the dimension table can be extracted. Associations between entities, one-to-many associations between entities, and count the number of each type of association in the ETL task. For example, multiple entities involved in the ETL can be counted by traversing the target table in the ETL task, and the relationships (including one-to-many relationships and many-to-many relationships) between entities can be determined at the same time. Wherein, for two entities with a many-to-many relationship, the many-to-many correspondence between the two is usually stored in the subsidiary table in the form of data records, and the two entities are associated with the subsidiary table. When counting the number of associations between entities and subsidiary tables, each entity needs to be counted once. For two entities with a one-to-many relationship, it can be directly determined that the two entities are related to each other. When counting the number of associations between entities, the two entities only need to be counted once. Each entity may also have multiple attributes, and the dimension table is used to store all possible values of each attribute. Therefore, based on the attributes of the entity appearing in the target table, it can be determined which dimension table or tables the entity is associated with. In statistics When counting the number of associations between entities and dimension tables, each dimension table needs to be counted once.

After determining the number of associations between entities and subsidiary tables involved in each ETL task, associations between entities and dimension tables, and associations between entities and entities, in step S102), for each to-be For scheduling the ETL tasks to be executed, the scheduling priority or weight of the ETL tasks is determined based on the preset weights for each type of association and the number of each type of association in the ETL task. For example, it can be calculated by the following formula (1):

Among them, Wl1 represents the weight of the association between the entity and the subsidiary table; Wl2 represents the weight of the association between the entity and the dimension table; Wl3 represents the weight of the association between the entity and the entity; where ni represents the i-th value that appears in the ETL task The number of associations can also be understood as the number of Wli in the formula, and i is a natural number. Wl1, Wl2, and Wl3 are weights preset according to specific business demand scenarios, and the value range of these weights is usually between 2-10, and the values of these weights can be changed correspondingly according to business changes.

The following takes the science and technology management data integration business as an example to illustrate the above-mentioned ETL task weights. FIG. 2 shows a schematic diagram of a process of determining weights of ETL tasks according to an embodiment of the present invention. As shown in Figure 2, the ETL task includes four entities: project, subject, unit, and personnel. When constructing the ETL task, in the target table to be loaded, there is a one-to-many relationship between the project and the subject ( Indicated by "1.n" in the figure), that is, there can be multiple topics for one project, but each topic can only correspond to one project, and cannot belong to two projects at the same time. There are many-to-many relationships between projects and personnel, between projects and units, between subjects and units, and between subjects and personnel. For example, the same person can participate in multiple projects and subjects at the same time, and the same unit can Corresponds to multiple projects and multiple subjects. In this ETL task, it is set to load a dimension table for each entity. It can be seen from Figure 2 that in this ETL task, there are 8 associations between entities and subsidiary tables, 4 associations between entities and dimension tables, and 1 association between entities, assuming Wl1, Wl2 and Wl3 are assigned values of 6, 5, and 10 respectively, and the weight of the ETL task can be determined accordingly:

Continuing to refer to FIG. 1 , step S103) sorts the ETL tasks to be scheduled and executed according to the scheduling priority of each ETL task determined in step S102). For example, the weights of a new batch of ETL tasks are {2, 6, 8, 4, 10, 3, 9}, and the sequence of ETL tasks after sorting is {10, 9, 8, 6, 4, 3, 2}. After sorting, the ETL task sequence is obtained in descending order according to the weight, and then the ETL tasks are assigned to each execution node in the distributed environment for execution in this order.

In this embodiment, the association between entities and subsidiary tables, the association between entities and dimension tables, and the association between entities in the ETL task are extracted through the target table included in the ETL task. For the ETL task The complexity of the corresponding business and the importance of the business data to be integrated are effectively quantified and evaluated, and the optimal expected scheduling task sequence sorted by weight is formed, which can meet the timely loading requirements of core business data and improve improve the efficiency of data integration.

In yet another embodiment, step S103) may further include acquiring performance indicators of each execution node, and assigning ETL tasks to be scheduled for execution to each execution node based on the performance indicators of each execution node. This is because when ETL task scheduling is assigned to each execution node in a distributed environment, the number of tasks currently run by different execution nodes and the amount of data contained in the task are different, that is, the performance and current load of each execution node at the same time are different. , if the number of tasks assigned to each execution node can be reasonably controlled according to the performance of the execution nodes, it can not only ensure the load balance of each execution node and even the entire distributed environment, but also improve the efficiency of task execution as a whole. Therefore, in step S103) before allocating ETL tasks, the performance index of each execution node can be queried earlier, and the current load of each execution node is classified according to the obtained performance index of each execution node, according to the current load of the execution node from Sort from low to high, select the corresponding execution node to allocate ETL tasks. The current load of each execution node can be determined according to the obtained performance index of the execution node. For example, assuming that CPU usage and memory usage are used as performance indicators as an example, the following formula (2) can be used to determine the execution node. Current load:

Among them, C is the CPU usage rate of the execution node; R is the memory usage rate of the execution node, L indicates the current load of the execution node, the larger L, the smaller the current load of the execution node; the smaller L, the smaller the execution node The greater the current load. Therefore, the priority allocation sequence of each execution node can be obtained according to the value of L in descending order. In yet another embodiment, the current load of the execution node can also be determined by the weighted average of each performance index, for example, L=w1*C+w2*R, where w1 and w2 are the weights set for performance indexes C and R , whose value is between 0-1. A larger L indicates that the current load of the execution node is greater; a smaller L indicates that the current load of the execution node is smaller. Therefore, the priority allocation sequence of each execution node can be obtained by arranging according to the value of L from small to large. It should be understood that the determination of the current load of the node using the CPU usage rate and the memory usage rate as performance indicators is only an example for illustration rather than any limitation, and those skilled in the art may make adjustments or modifications according to actual requirements.

In yet another embodiment, each execution node can also be classified according to the determined current load of each execution node, for example, each execution node can be divided into a high-load node and a medium-load node by using L determined by the above formula (2). , low-load nodes:

That is, the execution nodes in the distributed cluster environment are divided into three groups, each group consists of zero or more nodes, and the loads of the node members in the same group are similar. The execution node in the low-load node has a low load and has the strongest ability to accept task execution. Priority should be given to assigning ETL task scheduling to low-load executor nodes. If the group of low-load nodes is empty, assign ETL tasks to the group of medium-load nodes, and so on. If the above-mentioned low and medium load nodes are all empty, it means that the current load of all execution nodes in the entire distributed environment is very high. If all execution nodes are divided into groups of high-load nodes for a long time, an alarm mechanism needs to be set to prompt the distributed environment to be under high load for a long time, so as to prompt the system administrator to improve the performance of the distributed environment or increase the corresponding executor nodes to improve the loadability of the entire distributed environment.

In the solution of the above-mentioned embodiment, execution nodes are sequentially selected from low load to high load for ETL task distribution, so that ETL tasks with high scheduling priority are preferentially assigned to execution nodes with low current load for execution, which is not only beneficial to each execution node It can also improve the execution efficiency of ETL tasks.

In yet another embodiment, in step S103), the ETL tasks participating in scheduling may be allocated to execution nodes based on the data volume of the ETL tasks. The total amount of data involved in different ETL tasks is different, and the execution time of the ETL tasks is also different accordingly. If multiple ETL tasks with a large amount of task data are allocated to one or several execution nodes, then the ETL tasks on these execution nodes The waiting time will become longer, and the resources of each execution node cannot be effectively used in a balanced manner. Therefore, in this embodiment, the data volume of the ETL task is introduced as a reference factor for allocation, and a greedy balance algorithm is used to allocate the ETL task. Assuming that the initial processing capabilities of each execution node in the distributed cluster are the same, and each node can work independently, that is, without the assistance of other nodes, E={e1,e2,e3...,en} represents the newly acquired batch A set of mutually independent ETL tasks to be scheduled, in which there are n ETL tasks in total, ei represents the i-th task; D={d1,d2,d3...,dn} represents the set of data contained in n ETL tasks , where di is the amount of data contained in the i-th task ei; N={n1,n2,n3...nj} represents the set of execution nodes in the distributed cluster, a total of j nodes, where ni is the i-th execution dni _pre indicates the amount of data contained in the existing ETL tasks on the i-th execution node ni, and dni _aft indicates the amount of data contained in all ETL tasks on the i-th execution node ni after the tasks are allocated. The total amount of data contained in the ETL task on the execution node is The optimal expected distribution task data volume Opt _i of the i-th execution node can be expressed as:

The variance of the amount of data calculated by the following formula represents the data load index μ i of the execution node, then the data load index μ _i of the _i -th executor node ni can be expressed as:

μ _i ＝(dni _aft -dni _pre -Opt _i ) ² (5)

The overall data load index μ of execution nodes in a distributed cluster can be expressed as:

In the distribution process of ETL tasks, try to ensure the data load balance of cluster resources, that is, μ is relatively small. The maximum value of μ in the task distribution process can be limited by defining a threshold δ. If μ exceeds δ, it is considered that the node has a heavy data load and cannot accept new tasks. In this way, the value of μ is calculated in real time during the task distribution process, and nodes with μ _i <δ are selected each time to distribute tasks, thereby ensuring the load balance of cluster resources. In an example, the allocation of ETL tasks based on the greedy balance algorithm mainly includes the following steps:

(1) Initialize the ETL task set E={e1,e2,e3...,en}, the data volume set D={d1,d2,d3...,dn} contained in the ETL task, the executor node set N= {n1,n2,n3...nj};

(2) Sort the ETL tasks according to the amount of data from large to small, and store them in the queue Q, Q={q ₁ ,q ₂ ,q ₃ ,q ₄ ,...q _n }, where q ₁ is ( e ₁ ,d ₁ ),q ₂ is (e ₂ ,d ₂ ),...q _n is (e _n ,d _n ),d ₁ ≥d ₂ ≥d _n ;

(3) Calculate the data load index μ ₁ , μ ₂ , μ ₃ ,...μ _j of all execution nodes in the execution node set in real time; adjust the order of nodes from small to large according to the data load index to achieve the following effect: if there is μ ₁ <μ ₂ <μ ₃ <...<μ _j , then adjust the order of nodes to be n ₁ ,n ₂ ,n ₃ ,...,n _j ;

(4) Assign the number of nodes with μ _i < δ to the variable K, indicating the number of nodes that can be allocated and executed at this time; if K=0, it indicates that the load of the distributed environment is too high at this time, and the continuation of the task needs to be temporarily stopped Distribute or add new execution nodes;

(5) For n tasks in Q, if n>K, take out K tasks and distribute them to K nodes in turn, n=nK; otherwise, if 0<n≤K, take out all tasks and distribute them to the first n in turn Execution nodes, for example, e ₁ is distributed to n ₁ , and e ₂ is distributed to n ₂ . If n≤0, it means that all tasks in this batch have been executed, and the algorithm ends, otherwise execute (3).

In yet another embodiment, step S103) may include a) obtaining the performance index of each execution node, and determining the current load of each execution node according to the performance index of each execution node; b) distributing The execution nodes in the environment are divided into three groups: high-load node group, medium-load node group, and low-load node group; c) firstly, task allocation is performed in the low-load node group, and the data volume and Use the greedy balance algorithm introduced above to assign tasks to the execution nodes in the low-load node group; if the group of low-load nodes is empty and there are ETL tasks If distribution is required, use the greedy balance algorithm introduced above to continue to distribute the remaining ETL tasks to the execution nodes in the medium load node group, and so on. If the above-mentioned low and medium load nodes are all empty, it means that the current load of all execution nodes in the entire distributed environment is very high. An alarm mechanism can also be set to prompt the distributed environment to be under high load for a long time, so as to prompt the system administrator to The performance of the distributed environment is improved or the number of corresponding executor nodes is increased to improve the loadability of the entire distributed environment. In yet another embodiment, when task distribution fails, if the cause of the failure is caused by the destination execution node of the distribution, it can be set that within a period of time (penalty time) no more execution of the task execution request distribution operation to the execution node will be performed. In this way, the failure rate of task distribution can be reduced to a certain extent.

After the ETL tasks are assigned to each execution node, each executor node has an execution queue responsible for storing tasks, and each task occupies a thread resource on the queue. Due to the different amount of data contained in the ETL task, the corresponding execution time is different. In yet another embodiment, the execution efficiency of the ETL task is improved by balancing the execution time and waiting time of the ETL task, thereby indirectly improving the efficiency of data integration in the entire distributed environment. In this embodiment, the execution priority of the ETL task is set based on the execution time and waiting time of the ETL task, so that the execution node executes the ETL task in the order of the execution priority of the ETL task from high to low, and the set ETL task The execution priority of a task is constantly adjusted according to its execution time and waiting time. The following describes the execution process of the ETL task on a certain execution node in conjunction with FIG. 3 .

As shown in Figure 3, the process mainly includes step S301) In response to the execution node receiving a new ETL task, store the ETL task in the task cache queue, and record the arrival time of the ETL. Step S302) Estimating the execution time of the ETL task based on the amount of data in the ETL task. First obtain the amount of data involved in the ETL task, and then select a batch of ETL tasks with similar data volume as the ETL tasks to be executed from the ETL tasks that have been executed on the execution node in the most recent period of time. The execution time of the ETL task that has been obtained is used to estimate the execution time of the ETL task that has not yet been executed, for example, the execution time of this batch of ETL tasks is averaged as an estimate of the execution time of the ETL task to be executed. Step S303) In response to the execution of the current task of the execution node, for each ETL task to be executed, the execution priority of the ETL task is determined according to the waiting time of the ETL task and the estimated execution time. Assuming that there are currently n ETL tasks waiting to be executed on the execution node, set Tei to represent the execution time of the i-th ETL task ei (estimated according to the amount of data contained), and Twi to represent the waiting time of the i-th ETL task ei, then The objective function TotalTime when n ETL tasks are executed on the execution node can be expressed as:

The purpose of adjusting the execution order of ETL tasks through priority is to ensure that the time spent on the entire execution process is as low as possible (that is, the TotalTime is the smallest) during the execution of the ETL task by the execution node, that is, to make Tei and Twi as close as possible to the relative balance. In an embodiment, for each ETL task to be executed in the task cache queue, the execution priority of the ETL task is calculated based on the task execution time estimated in step S302) and the waiting time of the ETL task. For example, the following formula is used to determine the execution priority EP _i of the i-th ETL task ei:

Where Tti represents the time when the task ei arrives at the execution node, preferably between the task execution based on the estimated amount of data in the ETL task in step S302); and the waiting time Twi of each ETL task can be calculated in the following manner: T _wi =T _ni -T _ti , that is, the waiting time of each task is equal to the current time minus the time when the ETL task arrives at the execution node. It can be seen from formula (8) that EP _i must be greater than 1. When Twi is constant, the smaller Tei is, the higher the priority of EP _i is, similar to the short job priority algorithm; when Tei is constant, the larger Twi is, the higher the priority is. The higher the EP _i is, it is similar to the first-come-first-served algorithm; when both Twi and Tei are in an undeterminable state, this priority setting combines the current task execution status and task waiting time on the execution node to achieve the ETL task execution time and The overall relative balance of wait times. Continuing to refer to FIG. 3 , in step S304 ), the ETL task with the highest execution priority is selected from the ETL tasks to be executed for execution.

FIG. 4 is a schematic structural diagram of a system for scheduling and executing distributed ETL tasks according to an embodiment of the present invention. As shown in Figure 4, the system scheduler 401 and a plurality of executors 402a-n (collectively referred to as 402), the scheduler 401 obtains one or more ETL tasks to be scheduled and executed from the ETL task resource library, and distributes them to Execute on multiple executors in a distributed environment. Executor 402 is used to execute received ETL tasks. Although the block diagram depicts components in a functionally separate manner, such depiction is for illustration purposes only. The components shown in the figures may be arbitrarily combined or separated into separate software, firmware and/or hardware components. Moreover, no matter how such components are combined or divided, they may execute on the same computing device or on multiple computing devices, which may be connected by one or more networks.

The scheduler 401 includes a relationship analysis module, a priority determination module, and a scheduling module. The relationship analysis module is used for each ETL task to be scheduled and executed, based on the target table of data loading in the ETL task, to extract the association between the entity and the subsidiary table involved in the ETL task, and the relationship between the entity and the dimension table Associations, one-to-many associations between entities and entities; a priority determination module, configured to determine the scheduling priority of the ETL task based on the weight preset for each association and the number of each association in the ETL task a scheduling module, configured to assign each ETL task to each executor 402 in descending order of scheduling priority.

In yet another embodiment, the scheduler 401 may also include a load monitoring module, which is used to query the performance indicators of each executor, and determine the current load of each executor according to the obtained performance indicators of each executor. The scheduling module can also be configured to select corresponding executors from low to high according to the current load of executors to allocate ETL tasks. In yet another embodiment, the executor 402 may be configured to store the ETL task to be executed in the task cache queue in response to receiving a new ETL task, and record the arrival time of the ETL; The amount of data to estimate the execution time of the ETL task; in response to the completion of the current task, for each ETL task to be executed, the ETL is determined according to the waiting time and estimated execution time of the ETL task as described above The execution priority of the task; select the ETL task with the highest execution priority from the ETL tasks to be executed to execute.

In another embodiment of the present invention, there is also provided a computer-readable storage medium, on which computer programs or executable instructions are stored, and when the computer program or executable instructions are executed, the above-mentioned embodiment can be realized. The implementation principles of the technical solutions described above are similar and will not be repeated here. In the embodiments of the present invention, a computer-readable storage medium may be any tangible medium capable of storing data and readable by a computing device. Examples of computer readable storage media include hard drives, network attached storage (NAS), read only memory, random access memory, CD-ROM, CD-R, CD-RW, magnetic tape, and other optical or non-optical data storage devices. The computer readable storage medium may also include computer readable media distributed over network coupled computer systems so that the computer programs or instructions are stored and executed in a distributed manner.

References in this specification to "various embodiments," "some embodiments," "one embodiment," or "an embodiment" refer to particular features, structures, or properties described in connection with the embodiments, including In at least one embodiment. Thus, appearances of the phrases "in various embodiments," "in some embodiments," "in one embodiment," or "in an embodiment" in various places throughout this specification are not necessarily referring to the same implementation. example. Furthermore, the particular features, structures, or properties may be combined in any suitable manner in one or more embodiments. Therefore, a particular feature, structure, or property shown or described in connection with one embodiment may be combined in whole or in part with features, structures, or properties of one or more other embodiments without limitation, as long as the combination is not incompatible. Logical or not working.

In this specification, "comprising" and "having" and terms with similar meanings are intended to cover non-exclusive inclusion, for example, a process, method, system, product or device comprising a series of steps or units is not limited to the listed steps or units, but optionally further include steps or units not listed, or optionally further include other steps or units inherent to these processes, methods, products or devices. "A" or "an" also does not exclude a plurality. In addition, each element in the drawings of the present application is only for illustration and not drawn to scale.

Although the present invention has been described by the above-mentioned embodiments, the present invention is not limited to the embodiments described here, and includes various changes and changes made without departing from the scope of the present invention.

Claims (10)

1. a kind of method executed for distributed ETL task schedule, comprising:

For each ETL task of the execution to be dispatched of acquisition, based on the object table that data in the ETL task load, extracting should Being associated between entity and attached table, being associated with, one between entity and entity between entity and dimension table involved in ETL task To more associations；

It is based upon every kind of preset weight of association and every kind of number being associated in the ETL task determines the scheduling of the ETL task Priority；

Each ETL task is distributed to each execution node according to the order of dispatching priority from high to low.

2. according to the method described in claim 1, further including inquiring each execution node before the distribution for carrying out ETL task Performance indicator；And each present load for executing node is determined according to the performance indicator of each execution node of acquisition, according to The present load for executing node carries out the distribution of ETL task from node is accordingly executed down to high selection.

3. according to the method described in claim 1, wherein the dispatching priority of the ETL task is calculated by following formula:

Wherein, the associated weight between Wl1 presentation-entity and attached table；It is associated between Wl2 presentation-entity and dimension table Weight；Associated weight between Wl3 presentation-entity and entity；Wherein ni indicates i-th kind occurred in ETL task associated Number.

4. according to the method described in claim 2, described distribute each ETL task to each execution node includes:

A) data volume of each ETL task of execution to be dispatched is counted；

B) each total amount of data for executing all ETL tasks on node is counted；

C) the corresponding ETL task of maximum amount of data is selected from the ETL task of execution to be dispatched；

D) selection total amount of data is minimum and is currently assigned the execution node of ETL task not yet；

E) selected ETL task is distributed to selected execution node, and is to have distributed by the execution vertex ticks；

F) repeat step c)-e) until execution to be dispatched ETL task be assigned finish or until all execution nodes all by Labeled as having distributed；

G) the ETL task for needing to be dispatched execution is detected whether, if so, then re-flagging all execution nodes not divide Match, repeat step c)-g) it is finished until the ETL task of execution to be dispatched is assigned.

5. according to the method described in claim 1, further include:

New ETL task is received in response to executing node, which is stored in task buffer queue, and remembers Record the arrival time of the ETL；

The execution time of the ETL task is estimated based on the data volume in the ETL task；

Current task in response to executing node is finished, for pending each ETL task, according to the ETL task Waiting time and the execution time estimated determine the execution priority of the ETL task；

The highest ETL task of execution priority is selected from pending ETL task to execute.

6. according to the method described in claim 5, wherein estimating holding for the ETL task based on the data volume in the ETL task The row time includes:

Determine the data volume in the ETL task；

From the ETL task that execution is completed in a period of time nearest on the execution node, a batch and pending ETL are filtered out Task has the ETL task of set of metadata of similar data amount；

The execution time of this batch of ETL task averages, using obtained average value as the execution for the ETL task estimated Time.

7. according to the method described in claim 6, wherein the execution priority of ETL task is determined using following formula:

Wherein EP_iIndicate the execution priority of i-th of ETL task ei；Tei indicates the execution time of ETL task ei；Twi is indicated The waiting time of ETL task ei is equal to current time and subtracts the time that the ETL task reaches execution node.

8. a kind of system executed for distributed ETL task schedule, including scheduler and multiple actuators, scheduler is used for will One or more ETL tasks of execution to be dispatched are distributed to multiple actuators, and actuator is for executing the ETL task received； Wherein scheduler includes:

Relationship analysis module is added for each ETL task of the execution to be dispatched for acquisition based on data in the ETL task The object table of load extracts involved in the ETL task being associated between entity and attached table, the pass between entity and dimension table It is one-to-many between connection, entity and entity to be associated with；

Priority Determination module, for being based upon every kind of preset weight of association and every kind of number being associated in the ETL task Determine the dispatching priority of the ETL task；

Scheduler module, for distributing each ETL task to each actuator according to the order of dispatching priority from high to low.

9. system according to claim 8, wherein the scheduler further includes that load monitoring mould is fast, for inquiring each hold The performance indicator of row device, and determine according to the performance indicator of each actuator of acquisition the present load of each actuator；And The scheduler module is additionally configured to according to the present load of actuator from carrying out ETL task down to high selection respective actuators Distribution.

10. system according to claim 8, wherein the actuator is configured as:

In response to receiving new ETL task, which is stored in task buffer queue, and record the ETL's Arrival time；

The execution time of the ETL task is estimated based on the data volume in the ETL task；

Be finished in response to current task, for pending each ETL task, according to the waiting time of the ETL task and The execution time estimated determines the execution priority of the ETL task；

The highest ETL task of execution priority is selected from pending ETL task to execute.