Patent Publication Number: US-9846598-B2

Title: Modifying a flow of operations to be executed in a plurality of execution environments

Description:
BACKGROUND 
     Operations on data can be performed by different types of execution environments. For example, one execution environment can be a database management system (DBMS) environment, in which data is stored in relational tables and subject to database-based operations. As another example, an execution environment can include a MapReduce environment, which performs operations using map tasks and reduce tasks. There can also be other types of execution environments. 
    
    
     
       BRIEF DESCRIPTION OF THE DRAWING 
       Some embodiments are described with respect to the following figures: 
         FIG. 1  is a block diagram of an example system according to some implementations; 
         FIG. 2  is a flow diagram of a scheduling process according to some implementations; 
         FIG. 3  is a schematic diagram of a process of scheduling a hybrid flow, according to some implementations; 
         FIG. 4  is a schematic diagram of a process of decomposing a flow into flow fragments, according to some implementations; 
         FIG. 5  is a flow diagram of a process of applying an aggregate policy, according to some implementations; and 
         FIG. 6  is a schematic diagram of a process of scheduling a decomposed hybrid flow, according to further implementations. 
     
    
    
     DETAILED DESCRIPTION 
     Enterprises (e.g. business concerns, educational organizations, government agencies, etc.) can depend on reports and analyses (generally referred to as “computations”) that integrate data from a diverse collection of data repositories and that operate on the data using a variety of execution environments. In some examples, a single analytic computation can be modeled as a directed graph in which starting nodes are data sources, ending nodes are data targets, intermediate nodes are data operations, and arcs represent data flow. Such a computation can be referred to as an analytic data flow (or simply a “flow”). In other examples, a flow can have a different representation (other than a directed graph). An analytic data flow that utilizes more than one data repository or execution environment is referred to as a hybrid analytic data flow (or simply a “hybrid flow”). A collection of analytic data flows that is managed as a unit for some objective (e.g. the flows should complete before a deadline, or an average response time of individual flows should not exceed a threshold, etc.) is referred to as a workload. 
     A hybrid flow can include collections of operations that are performed in different execution environments. Each collection of operations that is performed in a respective execution environment can be referred to as a flow fragment of operations. 
     Examples of different types of execution environments include at least some of the following: database management system (DBMS) environment, MapReduce environment, an extract, transform, and load (ETL) environment, or other execution environments. Each execution environment can include an execution engine and a respective storage repository of data. An execution engine can include one or multiple execution stages for applying respective operators on data, where the operators can transform or perform some other action with respect to data. A storage repository refers to one or multiple collections of data. An execution environment can be available in a public cloud or public network, in which case the execution environment can be referred to as a public cloud execution environment. Alternatively, an execution environment that is available in a private network can be referred to as a private execution environment. 
     A DBMS environment stores data in relational tables and applies database operators (e.g. join operators, update operators, merge operators, and so forth) on data in the relational tables. A MapReduce environment includes map tasks and reduce tasks that can apply a map function and a reduce function, respectively. A map task processes input data to produce intermediate results, based on the respective map function that defines the processing to be performed by the map task. A reduce task takes as input partitions of the intermediate results from the map task to produce an output, based on the corresponding reduce function that defines the processing to be performed by the reduce tasks. 
     Another example execution environment includes an ETL environment, which extracts data from a source (or multiple sources), transforms the data, and loads the transformed data into a destination. 
     Although specific types of different execution environments are listed above, it is noted that in other examples, other types of execution environments can be used to perform operations on data. 
     A workload (which includes a set of flows) can be associated with a target performance objective, which identifies one or multiple goals that are to be met by the execution of the workload. A performance objective can also be referred to as a service level objective. An example performance objective relates to an execution time (e.g. a time duration for executing the workload or a target deadline by which the workload is to be completed). Another example performance objective is a resource usage objective, which can specify that usage of resources, such as computing resources, storage resources, or communication resources, should not exceed a target level. In other examples, other performance objectives can be employed. 
     A multi-flow has multiple flow fragments (a flow fragment includes a portion of a flow), where each of the flow fragments run entirely in one execution environment, the flow fragments have a partial execution order, and the flow fragments are linked through a data transfer from source flow fragment to target flow fragment. A multi-flow can have flow fragments that all run in the same execution environment. Alternatively, a multi-flow can have flow fragments that run in multiple execution environments this latter multi-flow is referred to as a hybrid flow. Although a hybrid flow is a type of multi-flow, note that a multi-flow may or may not be a hybrid flow. 
     An execution plan for a hybrid flow specifies where (target execution environments) the flow fragments of the hybrid flow are to execute, and can specify other details associated with execution of the flow fragments (such as order). A single hybrid flow can have many alternative execution plans due to overlap in functionality among the execution environments, multiple implementation details for operations, objectives for the execution plans (e.g. objectives relating to fault-tolerance, latency, etc.), and so forth. Based on an execution plan for a hybrid flow, a management system can deploy the flow fragments of the hybrid flow in the target execution environments, and can orchestrate the execution of the hybrid flow. There can be several issues associated with deploying an execution plan in the target execution environments. First, the state of a computing infrastructure of at least one execution environment may have changed between the time the execution plan was produced and the time the execution plan is executed. For example, the execution environment may have become overloaded (such that there is contention for resources) or the computing infrastructure may have experienced a fault. Second, the hybrid flow is associated with a performance objective that has to be met. In some cases, penalties may be specified for not meeting performance objectives. Thus, the management system should ensure that target performance objectives are achieved. 
     In some examples, a workload manager may exist within an individual execution environment, and this workload manager can adjust a priority of a task, the number of concurrently running tasks, and so forth, to increase the likelihood that a workload within the individual execution environment meets a respective target objective. However, a workload manager within a single execution environment can only make decisions optimal for that environment. Such workload manager has no knowledge of that state of other execution environments. So, for a workload that has flows to be executed across a number of different types of execution environments, workload management becomes more challenging. 
     In accordance with some implementations, a hybrid flow management system is provided to apply a policy-based hybrid flow management that uses at least one policy to modify an execution plan to increase the likelihood that a hybrid flow that is to execute in multiple different types of execution environments can meet a performance objective of the hybrid flow (or its associated workload). A policy can specify performance of at least one action to apply to the hybrid flow in response to a corresponding at least one condition occurring. The condition can relate to a characteristic of the hybrid flow. As discussed further below, application of the at least one action causes a modification of the execution plan for the hybrid flow. 
       FIG. 1  is a block diagram of an example computer system  100  according to some implementations. The computer system  100  includes one or multiple processors  102  that are coupled to a volatile or non-volatile storage medium (or storage media)  104 . A processor can include a microprocessor, microcontroller, processor module or subsystem, programmable integrated circuit, programmable gate array, or another control or computing device. The computer system  100  also includes a network interface  118  to allow the computer system  100  to communicate over a data network. 
     The storage medium (or storage media)  104  can store various data and machine-readable instructions. The machine-readable instructions can be executed by the processor(s)  102 . In some implementations, the machine-readable instructions are part of a hybrid flow management system  106  that includes various example modules, including an interface module  108  an optimizer module  110 , and an executor module  112 . 
     The executor module  112  includes a scheduler  114  for scheduling flow fragments of a hybrid flow for execution in respective execution environments. The executor module  112  also includes a dispatcher  115  for dispatching the scheduled flow fragments to the respective execution environments. Note that the dispatched flow fragments can be in the form of executable code generated by the executor module  112  for the respective flow fragments. In addition, the executor module  112  is able to monitor flow execution. 
     The storage medium (or storage media)  104  can also store one or multiple policies  116 , which can be used by the hybrid flow management system  106  to modify an execution plan to meet a performance goal of a hybrid flow. 
     The interface module  108  can receive a flow definition for creating or modifying a flow. As an example, the interface module  108  can present a graphical user interface (GUI) to allow for users to interactively create or modify a flow. Alternatively, a flow can be written in a declarative language and imported through the interface module  108 . The flow that is created or modified using the interface module  108  can be a hybrid flow or non-hybrid flow. 
     The optimizer module  110  generates multiple candidate execution plans for each flow (hybrid flow or a non-hybrid flow). The optimizer module  110  is able to consider alternative candidate execution plans for a given flow, and can estimate the respective costs of the candidate execution plans. Examples of costs can include processing resource usage cost, storage resource usage cost, communication cost, input/output (I/O) cost, and so forth. An execution plan (which can be an optimal execution plan) from among the candidate execution plans can be selected for execution. Where an optimal execution plan can refer to an execution plan that is associated with a lowest cost or that satisfies some other criterion. The selected execution plan is provided by the optimizer module  110  to the executor module  112 . 
       FIG. 2  is a flow diagram of a process that can be performed by the hybrid flow management system  106  according to some implementations. The process of  FIG. 2  receives (at  202 ) a hybrid flow of operations to be executed in multiple execution environments according to a distribution, as specified by an execution plan. The execution plan can be provided by the optimizer module  110  to the executor module  112 . The distribution of a hybrid flow can refer to where (which execution environments) the flow fragments of the hybrid flow are to execute, and/or to an order or sequence of the flow fragments. 
     In response to determining that the distribution is unable to achieve a target performance objective or unable to satisfy some other criterion, the distribution can be modified (at  204 ) according to at least one policy that specifies at least one action to apply to the hybrid flow in response to a corresponding at least one condition relating to a characteristic of the hybrid flow. In some implementations, the at least one action is applied by the optimizer module  110  to modify the execution plan. Modification of the execution plan causes the distribution of the hybrid flow to change. As discussed further below, policies that can be used for modifying a distribution of a hybrid flow can include a policy relating to decomposing a flow into flow fragments, a policy relating to shipping a flow from one execution environment to another execution environment, an aggregate policy that relates to both decomposing and shipping flows, or other policies. 
       FIG. 3  is a schematic diagram illustrating scheduling of flows, including hybrid flows. Flows are received by the scheduler  114 . In some examples, each flow can include start time information and a class identifier. The start time information can specify a start time, or a range (window) of start times, at which a flow is to execute. A class identifier can associate a flow with a workload. 
     In examples according to  FIG. 3 , the scheduler  114  uses two data structures: a ready queue  302  and a set of run vectors B 1 , B 2 , . . . , B N . Each flow f that is ready to execute (based on the start time information of the flow) enters the ready queue  302 . To simplify the discussion, each entry in the ready queue is a flow for a single execution environment. Hybrid flows are considered later. The ready queue  302  orders flows by their arrival time or requested execution time, t. In the example of  FIG. 3 , the ready queue  302  includes three respective flows f 1   E1 , f 2   E2 , and f 3   E2  that are associated with three respective start times t 1 , t 2 , t 3 . Flow f 1   E1  is designated (by the execution plan provided by the optimizer module  104 ) to execute on execution engine E 1 , while flows f 2   E2  and f 3   E2  are designated to execute on execution engine E 2 . An execution engine is part of a corresponding execution environment. An execution engine can include one or multiple execution stages for applying respective operators on data, where the operators can transform or perform some other action with respect to data. 
     Each run vector B i  (i=1 to N) identifies the flow(s) currently executing on the respective execution engine E 1 , E 2 , . . . E N . Each run vector B i  includes one slot per executing flow. An execution engine E i  can be associated with a respective multi-programming level (MPL), which specifies the maximum number of flows that can concurrently execute on the respective execution engine. The MPL specifics the size of a run vector, since there is one slot per executing flow. 
     When the run vector B i  for an execution engine E i  has a free slot, an orchestrator  304  in the scheduler  114  searches in order through the ready queue  302  for a flow that uses the execution engine E i . Alternatively, the dispatcher  304  can be external of the scheduler  114 . If such a flow is found in the ready queue  302 , then the identified flow is dequeued from the ready queue  302 , and after plan validation and code generation, the flow is dispatched by the dispatcher  115  to the respective execution engine E i . 
     In some cases, plan validation may fail. Plan validation includes determining that the execution plan is feasible, such as by determining that requested resources are available in the execution engines. Also, plan validation includes determining that an execution plan is consistent with a target performance objective. For example, an execution plan may have been designed to execute a flow on execution engine E 1 , but that execution engine may not be available when the execution plan is to be executed. Alternatively, if the average response time of flows in a workload is exceeding a threshold, a flow may be redirected to an execution environment that is less heavily loaded. If the execution plan is deemed not useable, then the plan validation would fail. In response to failure of the plan validation, the scheduler  114  can request (at  306 ) the optimizer module  104  to provide an alternative execution plan. The alternative plan can be created by the optimizer module  104  by performing a policy-based modification of an execution plan. 
     The policy-based modification can include a decomposition action and/or a flow shipping action, in some implementations. In other examples, other policy-based modifications can be performed. 
     A decomposition action takes a single flow and produces an equivalent multi-flow in which the flow fragments are executed sequentially. For example, given a flow implemented as a nested Structured Query Language (SQL) query (that has multiple levels of sub-queries), decomposition can create flow fragments for respective nested sub-queries, where the result of a child sub-query is saved in a table that can be read by a parent sub-query. Each flow fragment can run independently, such as in a separate session. 
     Executing multiple flow fragments may cause less resource contention single, large flow that uses a large fraction of an execution engine. The flow fragments perform more I/O operations than the single flow due to storing of intermediate results. However, other concurrently running flows may benefit because the resources of the execution environment are shared more equitably and so the overall workload performance may improve. 
     An example decomposition policy relating to use of a decomposition action can be as follows. First, the resource usage cost of a flow is determined. If the cost of the flow exceeds a threshold budget θ, then the flow is decomposed. 
     Decomposition returns a set S f  of flow fragments. At least some of the flow fragments can be put back into the ready queue  302 . As depicted in  FIG. 4 , it is assumed that the ready queue  302  initially contains two flows f 1  and f 2  to be executed on an execution engine. Based on evaluation of the decomposition policy, the optimizer module  110  decides to decompose f 1  into the flow fragments S f1 ={f 11 , f 12 , f 13 , . . . }. Although flow fragment f 11  is placed back into the ready queue  302 , note that the other flow fragments of the set S f1  are not put immediately put into the ready queue  302 . Instead, these other flow fragments (f 12 , f 13 ) can be kept in a buffer  402  associated with the ready queue  302 . 
     The first flow fragment, here f 11 , can be dequeued and executed on the execution engine. In response to dequeuing f 11 , the next flow fragment f 12  can be moved from the buffer  402  to the ready queue  302 . If f 11  terminates before the arrival time t 2  of flow f 2  in the ready queue  302 , the next flow fragment, f 12 , can also be executed on the execution engine. Otherwise, if f 11  does not terminate before the arrival time t 2  of flow f 2 , the flow f 2  is executed on the execution engine, and any other pending flow that has arrived before f 11  continues until completion. 
     The foregoing process continues until all flows and flow fragments in the ready queue  302  have been processed. 
     A flow shipping action takes a flow that is initially designated for running on a first execution engine and reassigns the flow to run on a different engine. Shipping a flow from one execution engine to another execution engine may improve overall performance by off-loading work from a heavily used execution engine onto a lightly used execution engine. A flow shipping policy can specify any one or some combination of the following. A flow is shipped from a first execution engine to a second execution engine if a cost of the flow on the first execution engine exceeds a cost threshold. A flow is not shipped from a first execution engine to a second execution engine if the cost of providing the flow to the second execution engine (where this cost can be an aggregate of the cost of shipping the flow and the cost of executing the flow on the second execution engine) is higher than the cost on the first execution engine. 
     Note that in addition to the cost of shipping operations associated with a flow, there is also a cost associated with where data accessed by the flow is stored. The cost of data access should be considered when determining whether a flow is to be shipped to a different execution engine and this consideration may include copying the data from one location to another). 
     The flow shipping action can first identify all execution engines, which are listed in a set Φ, that are able of executing a given flow. For example, an execution engine is able to execute the given flow if all operators of the given flow are supported in the engine. Once this set Φ is identified, then the next-best available execution engine can be identified, such as by using a function top(E x ); E x ∈Φ. This function may use the top-k (k≧1) execution plans for the given flow produced by the optimizer module  110  to find the best execution plan that uses an available execution engine. Alternatively, the function may ask the optimizer module  110  to re-optimize the execution plan for the new execution engine. 
     The previous discussion regarding decomposition and flow shipping actions assumes single-engine flows. Hybrid flows can be handled with minor changes. 
     In the context of a hybrid flow that has multiple flow fragments, decomposition can be applied on a flow fragment, to decompose the flow fragment into further fragments. Note that, under certain conditions, the decomposition policy can further cause at least two of the flow fragments of the hybrid flow to be composed (combined together). After combining at least two flow fragments together to form a combined flow, the combined flow can subsequently be decomposed again. 
     For a hybrid flow, one or multiple flow fragments of the hybrid flow can be shipped using the flow shipping action discussed above. Note that the shipping action can also cause the shipping of a flow combined from at least two flow fragments of the hybrid flow. 
     Let f h   E     x    denote a flow f h  to be run on engine E x . A hybrid flow, say f h :f h1   E1 →f h2   E2 , can be considered as a combination of multiple flow fragments, f h1   E1  and f h2   E2 , with dependencies (where the flow fragment f h2   E2  is dependent on the output of the flow fragment f h1   E1 ). Similarly, a multi-flow, say f m :f m1   E →f m2   E , which executes on one engine E can be considered as a combination of multiple flow fragments, f m1   E  and f m2   E , with dependencies. Thus, each flow fragment gets its own position in the ready queue  302 , but the dependencies among the flow fragments are considered before a flow fragment is executed. 
     Two types of dependencies can exist between a pair of flow fragments. A sequential dependency means a first flow fragment has to terminate before a second flow fragment can begin. For example, flow fragments resulting from decomposition generally have a sequential dependency to ensure that intermediate result tables are written before they are read. A parallel dependency means that the two flow fragments have to start together. This dependency is used when a producer flow fragment pipes data to a consumer flow fragment. The scheduler  114  considers these dependencies when choosing to run a flow fragment. For example, if two flow fragments have a parallel dependency, the scheduler  114  cannot run them until their respective target execution engines both have free slots in the corresponding run vectors. 
     In the foregoing discussion, it is assumed that the policy-based decomposition action and flow shipping action are performed separately. It is noted that, in some implementations, both a decomposition action and a flow shipping action can be applied. When scheduling a hybrid flow, it is possible to decompose the hybrid flow into a number of flow fragments, and some or all of the flow fragments can be shipped from one execution engine to a different execution engine. Alternatively, it is also possible to ship a hybrid flow to another execution engine first, and then decomposition is applied after the shipping. 
     A policy that allows for application of both the decomposition action and a flow shipping action can be referred to as an aggregate policy. An example pseudocode for the aggregate policy is provided below. 
     
       
         
           
               
               
               
             
               
                   
               
             
            
               
                  1 
                 input 
                 : The queue Q 
               
               
                  2 
                 output 
                 : An execution action 
               
            
           
           
               
               
            
               
                  3 
                 f ← Q.pop( ); 
               
               
                  4 
                 e ← eng(f); 
               
               
                  5 
                 if B E  has a free slot &amp;&amp; eval(f, E) &lt; θ then 
               
            
           
           
               
               
               
            
               
                  6 
                   
                 execute f on E; 
               
            
           
           
               
               
            
               
                  7 
                 else 
               
            
           
           
               
               
               
            
               
                  8 
                   
                 f x   E     y    = best (dcmp(f), ship(f E , f E′ ) ), where E′ is top-1 c{acute over (h)}oice 
               
               
                   
                   
                 from Φ for f; 
               
               
                  9 
                   
                 execute f x  on E y ; 
               
            
           
           
               
               
            
               
                 10 
                 end 
               
               
                   
               
            
           
         
       
     
     A discussion of the pseudocode is made in conjunction with  FIG. 5 . The input to the pseudocode is the ready queue Q ( 302  in  FIG. 3 ), and the output of the pseudocode is an execution action of a flow. It is assumed that, at the head of the ready queue Q (bottom of the ready queue  302  in  FIG. 3 ), there is a flow f that is planned for execution on execution engine E. As also shown in  FIG. 5 , this flow f is dequeued (at  502 ) (line 3 of the pseudocode) from the ready queue  302 . Next, it is determined (at  504 ) if the execution engine E has empty slots in the associated run vector (B E ) and the flow f can be executed on E while meeting a performance objective (this is computed by a function eval(f, E) at line 5 of the pseudocode). If so, then the flow f can simply be executed (at  506 ) on the execution engine E (line 6 of the pseudocode). In some examples, the function eval(f, E) can evaluate whether a resource usage cost is less than a threshold budget θ, in examples where the performance objective relates to resource usage. If the determination (at  504 ) that the performance objective cannot be met or the execution engine E is not available at this time (line 5 of the pseudocode evaluates to false), then a search is performed (at  508 ) to look for better execution plans that satisfy a performance objective (lines 7-9 of the pseudocode). 
     The better execution plan can be produced based on performing both the decomposition action and the flow shipping action. At line 7 of the pseudocode, the function best(dcmp(f),ship(f E ,f E′ ) compares two possible strategies, namely the decomposition and shipping strategies, and outputs a flow to execute. The function dcmp(f) decomposes a flow f to a number of flow fragments. The function ship(f E , f E′ ) examines the possible variants (f E ,f E′ ) of the same flow executed on different execution engines. Assuming that the set of all valid execution engines that can be used to execute f is noted as Φ (this is computed by the function eng(f) at line 4 of the pseudocode), then for each flow variant to be executed on engine E x , the function eval(f,E x ), ∀E x ∈Φ, is computed. The top-1 (i.e. best) solution, which is an execution plan produced based on performing both a decomposition action and flow shipping action, is returned by this computation. 
     The flow is then executed (at  510 ) according to the execution plan produced by the search at  508 . 
     If a flow includes more than one flow fragment, then the flow fragments have dependencies defined by f 1 →f 2 , which specifies that execution of f 2  starts only after f 1  has finished. If f 1  and f 2  are scheduled to run on different execution engines, then the orchestrator  304  can use a dependency buffer  602  to track the dependencies, as shown in  FIG. 6 . In  FIG. 6 , one entry of the dependency buffer  602  keeps track of the dependency between f 1  and f 2 . 
     Assume that the ready queue  302  in  FIG. 6  contains a hybrid flow f including two flows f 1  and f 2 . Then, the orchestrator  304  can dequeue the first flow f 1  from the head of the ready queue  302 . If evaluation of the aggregate policy indicates that this flow f 1  should be decomposed into three fragments, f 11  to f 13 , then the flow f 2  cannot be executed until f 13  finishes. Each of these three flow fragments f 11  to f 13  can be shipped to either engine E x  or E y . 
     Note that if decomposition and shipping are continually performed one after the other, a flow may be trapped in an infinite loop. In practice, this can be avoided by several using safety conditions, such as by setting a maximum number of actions that can be applied on a given flow (e.g. no more than two different actions may be applied to the same flow); setting a minimum size of a flow (e.g. a. flow containing x nodes cannot be further decomposed); disallowing the repeating of the same action to the same flow (e.g. a flow cannot be decomposed or shipped twice), and so forth. 
     Data and machine-readable instructions can be stored in a machine-readable or computer-readable storage medium or storage media (such as  104  in  FIG. 1 ), which can include different forms of memory including semiconductor memory devices such as dynamic or static random access memories (DRAMs or SRAMs), erasable and programmable read-only memories (EPROMs), electrically erasable and programmable read-only memories (EEPROMs) and flash memories; magnetic disks such as fixed, floppy and removable disks; other magnetic media including tape; optical media such as compact disks (CDs) or digital video disks (DVDs); or other types of storage devices. Note that the instructions discussed above can be provided on one computer-readable or machine-readable storage medium, or alternatively, can be provided on multiple computer-readable or machine-readable storage media distributed in a large system having possibly plural nodes. Such computer-readable or machine-readable storage medium or media is (are) considered to be part of an article (or article of manufacture). An article or article of manufacture can refer to any manufactured single component or multiple components. The storage medium or media can be located either in the machine running the machine-readable instructions, or located at a remote site from which machine-readable instructions can be downloaded over a network for execution. 
     In the foregoing description, numerous details are set forth to provide an understanding of the subject disclosed herein. However, implementations may be practiced without some or all of these details. Other implementations may include modifications and variations from the details discussed above. It is intended that the appended claims cover such modifications and variations.