In-Memory Dataflow Execution with Dynamic Placement of Cache Operations and Action Execution Ordering

• US 20200133859A1
• Filed: 10/30/2018
• Published: 04/30/2020
• Est. Priority Date: 10/30/2018
• Status: Active Grant

• Current Assignee: Emc IP Holding Company LLC (Dell Technologies Inc.)
• Original Assignee: Emc IP Holding Company LLC (Dell Technologies Inc.)
• Inventors: Gottin, Vinicius Michel, Porto, Fabio Andre Machado, Souto, Yania Molina
• Granted Patent: US 10,698,821 B2
• Time in Patent Office: Days
• CPC Class Codes: G06F 12/0806 Multiuser, multiprocessor or multiprocessing cache systems
• CPC Class Codes: G06F 17/10 Complex mathematical operations function generation by table look-up G06F1/03; evaluation of elementary functions by calculation G06F7/544
• CPC Class Codes: G06F 2212/1041 Resource optimization
• CPC Class Codes: G06F 2212/152 Virtualized environment, e.g. logically partitioned system
• CPC Class Codes: G06F 2212/502 using adaptive policy
• CPC Class Codes: G06F 2212/6046 Using a specific cache allocation policy other than replacement policy
• CPC Class Codes: G06F 2212/608 Details relating to cache mapping
• CPC Class Codes: H04L 67/568 Storing data temporarily at an intermediate stage, e.g. caching

Abstract

A dataflow execution environment is provided with dynamic placement of cache operations and action execution ordering. An exemplary method comprises: obtaining a current cache placement plan for a dataflow comprised of multiple operations and a corresponding current cache gain estimate; selecting an action to execute from a plurality of remaining dataflow actions based on a predefined policy; executing one or more operations in a lineage of the selected action and estimating an error as a difference in an observed execution time and an estimated execution time given by a cost model; obtaining an alternative cache placement plan for the dataflow following the execution in conjunction with a predefined new plan determination criteria being satisfied and a corresponding alternative cache gain estimate; implementing the alternative cache placement plan in conjunction with a predefined new plan implementation criteria being satisfied; and selecting a next action to execute from a plurality of remaining actions in the dataflow based on a predefined policy.

Claims

1. A method, comprising the steps of:

obtaining a cost model for the execution of operations of a dataflow in a parallel processing framework with a given infrastructure and input dataset;

obtaining a current cache placement plan for the dataflow, wherein the current cache placement plan comprises a combination of output datasets of a subset of the operations in the dataflow to cache, using one or more cache operations in the dataflow, based on an estimated reduction in a total execution cost for the dataflow in conjunction with the current cache placement plan being implemented given an input dataset;

obtaining a current cache gain estimate for the current cache placement plan;

selecting an action of the dataflow to execute from a plurality of remaining actions in the dataflow based on a predefined next action policy that selects the next action of the dataflow to execute from the plurality of remaining actions in the dataflow;

executing one or more operations in a lineage of the selected action of the dataflow;

determining, using at least one processing device, an alternative cache placement plan for the dataflow following the execution in conjunction with a predefined new plan determination criteria being satisfied, wherein the alternative cache placement plan comprises an alternative combination of output datasets of a second subset of the operations in the dataflow to cache, using one or more alternative cache operations in the dataflow, relative to the current cache placement plan;

obtaining an alternative cache gain estimate for the alternative cache placement plan;

implementing, using the at least one processing device, the alternative cache placement plan in conjunction with the predefined new plan implementation criteria being satisfied; and

selecting a next action of the dataflow to execute from a plurality of remaining actions in the dataflow based on the predefined next action policy.

2. The method of claim 1, wherein the new plan determination criteria comprises a difference between prior estimates of an execution time of the one or more executed operations obtained from the cost model and an actual execution time of the one or more executed operations exceeding a predefined error threshold.

3. The method of claim 2, further comprising the step of updating the cost model for the one or more executed operations based on the difference.

4. The method of claim 3, further comprising the step of updating the predefined error threshold in conjunction with the current cache placement plan being substantially the same as the alternative cache placement plan.

5. The method of claim 3, wherein the executing the one or more operations in the lineage further comprises the step of updating a representation of an available computational infrastructure considering variations to an execution environment to account for the difference.

6. The method of claim 3, wherein the executing the one or more operations in the lineage further comprises the step of updating a representation of the input dataset following the executing to account for the difference.

7. The method of claim 1, wherein the predefined new plan implementation criteria comprises a difference between the alternative cache gain estimate and the current cache gain estimate exceeding a predefined threshold.

8. The method of claim 1, further comprising the steps of removing one or more of the cached datasets from memory based on a number of references to the operations that generated the cached datasets following an updating of the number of references and determining a new alternative cache placement plan for the dataflow following the removing.

9. The method of claim 1, wherein the plurality of operations comprises a plurality of lazy-evaluated data transformations and a plurality of data actions that enact the lazy-evaluated data transformations.

10. The method of claim 1, wherein the executing the one or more operations in the lineage of the selected action further comprises the steps of, following an execution of each given one of the one or more operations in the lineage:

determining a new alternative cache placement plan for the dataflow following the execution of the given operation;

implementing the new alternative cache placement plan in conjunction with a difference between the current cache gain and the new alternate cache gain for the new alternative cache placement plan exceeding a predefined change cost threshold; and

updating the current cache gain with the new alternative cache placement plan in conjunction with the new alternative cache placement plan being implemented.

11. The method of claim 1, wherein the predefined next action policy selects the next action of the dataflow to execute from the plurality of remaining actions in the dataflow one or more of:

(a) having a substantially least number of referenced cached operations and (b) substantially minimizing the total execution cost of the dataflow considering future changes in the current cache placement plan.

12. A computer program product, comprising a tangible machine-readable storage medium having encoded therein executable code of one or more software programs, wherein the one or more software programs when executed by at least one processing device perform the following steps:

obtaining a cost model for the execution of operations of a dataflow in a parallel processing framework with a given infrastructure and input dataset;

obtaining a current cache placement plan for the dataflow, wherein the current cache placement plan comprises a combination of output datasets of a subset of the operations in the dataflow to cache, using one or more cache operations in the dataflow, based on an estimated reduction in a total execution cost for the dataflow in conjunction with the current cache placement plan being implemented given an input dataset;

obtaining a current cache gain estimate for the current cache placement plan;

selecting an action of the dataflow to execute from a plurality of remaining actions in the dataflow based on a predefined next action next action policy that selects the next action of the dataflow to execute from the plurality of remaining actions in the dataflow;

executing one or more operations in a lineage of the selected action of the dataflow;

determining an alternative cache placement plan for the dataflow following the execution in conjunction with a predefined new plan determination criteria being satisfied, wherein the alternative cache placement plan comprises an alternative combination of output datasets of a second subset of the operations in the dataflow to cache, using one or more alternative cache operations in the dataflow, relative to the current cache placement plan;

obtaining an alternative cache gain estimate for the alternative cache placement plan;

implementing, using the at least one processing device, the alternative cache placement plan in conjunction with the predefined new plan implementation criteria being satisfied; and

selecting a next action of the dataflow to execute from a plurality of remaining actions in the dataflow based on the predefined next action policy.

13. The computer program product of claim 12, wherein the new plan determination criteria comprises a difference between prior estimates of an execution time of the one or more executed operations obtained from the cost model and an actual execution time of the one or more executed operations exceeding a predefined error threshold and the predefined new plan implementation criteria comprises a difference between the alternative cache gain estimate and the current cache gain estimate exceeding a predefined threshold.

14. The computer program product of claim 12, further comprising the steps of removing one or more of the cached datasets from memory based on a number of references to the operations that generated the cached datasets following the an updating of the number of references and determining a new alternative cache placement plan for the dataflow following the removing.

15. The computer program product of claim 12, wherein the predefined next action policy selects the next action of the dataflow to execute from the plurality of remaining actions in the dataflow one or more of:

(a) having a substantially least number of referenced cached operations and (b) substantially minimizing the total execution cost of the dataflow considering future changes in the current cache placement plan.

16. A system, comprising:

a memory; and

at least one processing device, coupled to the memory, operative to implement the following steps;

obtaining a cost model for the execution of operations of a dataflow in a parallel processing framework with a given infrastructure and input dataset;

obtaining a cost model for the execution of operations of a dataflow in a parallel processing framework with a given infrastructure and input dataset;

obtaining a current cache placement plan for the dataflow, wherein the current cache placement plan comprises a combination of output datasets of a subset of the operations in the dataflow to cache, using one or more cache operations in the dataflow, based on an estimated reduction in a total execution cost for the dataflow in conjunction with the current cache placement plan being implemented given an input dataset;

obtaining a current cache gain estimate for the current cache placement plan;

selecting an action of the dataflow to execute from a plurality of remaining actions in the dataflow based on a predefined next action next action policy that selects the next action of the dataflow to execute from the plurality of remaining actions in the dataflow;

executing one or more operations in a lineage of the selected action of the dataflow;

determining an alternative cache placement plan for the dataflow following the execution in conjunction with a predefined new plan determination criteria being satisfied, wherein the alternative cache placement plan comprises an alternative combination of output datasets of a second subset of the operations in the dataflow to cache, using one or more alternative cache operations in the dataflow, relative to the current cache placement plan;

obtaining an alternative cache gain estimate for the alternative cache placement plan;

implementing, using the at least one processing device, the alternative cache placement plan in conjunction with the predefined new plan implementation criteria being satisfied; and

selecting a next action of the dataflow to execute from a plurality of remaining actions in the dataflow based on the predefined next action policy.

17. The system of claim 16, wherein the new plan determination criteria comprises a difference between prior estimates of an execution time of the one or more executed operations obtained from the cost model and an actual execution time of the one or more executed operations exceeding a predefined error threshold and the predefined new plan implementation criteria comprises a difference between the alternative cache gain estimate and the current cache gain estimate exceeding a predefined threshold.

18. The system of claim 16, further comprising the steps of removing one or more of the cached datasets from memory based on a number of references to the operations that generated the cached datasets following an updating of the number of references and determining a new alternative cache placement plan for the dataflow following the removing.

19. The system of claim 16, further comprising the steps of removing one or more of the cached datasets from memory based on a number of references to the operations that generated the cached datasets following an updating of the number of references and determining a new alternative cache placement plan for the dataflow following the removing.

20. The system of claim 16, wherein the predefined next action policy selects the next action of the dataflow to execute from the plurality of remaining actions in the dataflow one or more of:

(a) having a substantially least number of referenced cached operations and (b) substantially minimizing the total execution cost of the dataflow considering future changes in the current cache placement plan.