In-memory dataflow execution with dynamic placement of cache operations
A dataflow execution environment is provided with dynamic placement of cache operations. An exemplary method comprises: obtaining a first cache placement plan for a dataflow comprised of multiple operations; executing operations of the dataflow and updating a number of references to the executed operations to reflect remaining executions of the executed operations; determining a current cache gain by updating an estimated reduction in the total execution cost for the dataflow of the first cache placement plan; determining an alternative cache placement plan for the dataflow following the execution; and implementing the alternative cache placement plan based on a predefined threshold criteria. A cost model is optionally updated for the executed operations using an actual execution time of the executed operations. A cached dataset can be removed from memory based on the number of references to the operations that generated the cached datasets.
1. A method, comprising the steps of:
obtaining a first cache placement plan for a dataflow comprised of a plurality of operations, wherein said first cache placement plan comprises a combination of output datasets of a subset of said operations in said dataflow to cache using one or more caching operations inserted into the dataflow to implement the first cache placement plan based on an estimated reduction in a total execution cost for the dataflow when said first cache placement plan is implemented given an input dataset;
executing one or more operations of said dataflow and updating a count of references to said one or more executed operations to reflect remaining executions of said one or more executed operations in the dataflow;
determining, using at least one processing device, a current cache gain by updating the estimated reduction in the total execution cost for the dataflow of said first cache placement plan based on the updated count of references to said one or more executed operations;
determining, using the at least one processing device, an alternative cache placement plan for the dataflow following said execution based on the determined current cache gain and the updated count of references to said one or more executed operations, wherein said alternative cache placement plan comprises a combination of output datasets of a different subset of said operations in said dataflow to cache using one or more additional caching operations inserted into the dataflow to implement the alternative cache placement plan than said first cache placement plan; and
implementing, using the at least one processing device, said alternative cache placement plan when a difference between said current cache gain for said first cache placement plan and an estimated reduction in the total execution cost for the dataflow when said alternative cache placement plan is implemented satisfies one or more predefined threshold criteria.
2. The method of claim 1 , further comprising the step of updating a cost model for said one or more executed operations based on 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.
3. The method of claim 2 , wherein said determining of the current cache gain is further based on the cost model.
4. The method of claim 1 , further comprising the step of removing one or more of said cached datasets from memory based on the number of references to the operations that generated said cached datasets following said updating of the number of references.
5. The method of claim 1 , wherein said executing said one or more operations of said dataflow further comprises the step of updating a representation of an available computational infrastructure considering variations to the environment.
6. The method of claim 1 , wherein said executing said one or more operations of said dataflow further comprises the step of updating a representation of the input dataset following said executing.
7. The method of claim 1 , wherein said plurality of operations comprises a plurality of lazy-evaluated data transformations and a plurality of data actions that enact the lazy-evaluated data transformations.
8. A computer program product, comprising a non-transitory processor-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 cause the at least one processing device to perform the following steps:
obtaining a first cache placement plan for a dataflow comprised of a plurality of operations, wherein said first cache placement plan comprises a combination of output datasets of a subset of said operations in said dataflow to cache using one or more caching operations inserted into the dataflow to implement the first cache placement plan based on an estimated reduction in a total execution cost for the dataflow when said first cache placement plan is implemented given an input dataset;
executing one or more operations of said dataflow and updating a count of references to said one or more executed operations to reflect remaining executions of said one or more executed operations in the dataflow;
determining, using at least one processing device, a current cache gain by updating the estimated reduction in the total execution cost for the dataflow of said first cache placement plan based on the updated count of references to said one or more executed operations;
determining, using the at least one processing device, an alternative cache placement plan for the dataflow following said execution based on the determined current cache gain and the updated count of references to said one or more executed operations, wherein said alternative cache placement plan comprises a combination of output datasets of a different subset of said operations in said dataflow to cache using one or more additional caching operations inserted into the dataflow to implement the alternative cache placement plan than said first cache placement plan; and
implementing, using the at least one processing device, said alternative cache placement plan when a difference between said current cache gain for said first cache placement plan and an estimated reduction in the total execution cost for the dataflow when said alternative cache placement plan is implemented satisfies one or more predefined threshold criteria.
9. The computer program product of claim 8 , further comprising the step of updating a cost model for said one or more executed operations based on 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.
10. The computer program product of claim 8 , further comprising the step of removing one or more of said cached datasets from memory based on the number of references to the operations that generated said cached datasets following said updating of the number of references.
11. The computer program product of claim 8 , wherein said executing said one or more operations of said dataflow further comprises the step of updating a representation of an available computational infrastructure considering variations to the environment.
12. The computer program product of claim 8 , wherein said executing said one or more operations of said dataflow further comprises the step of updating a representation of the input dataset following said executing.
13. The computer program product of claim 8 , wherein said plurality of operations comprises a plurality of lazy-evaluated data transformations and a plurality of data actions that enact the lazy-evaluated data transformations.
14. An apparatus, comprising:
a memory; and
at least one processing device, coupled to the memory, operative to implement the following steps:
obtaining a first cache placement plan for a dataflow comprised of a plurality of operations, wherein said first cache placement plan comprises a combination of output datasets of a subset of said operations in said dataflow to cache using one or more caching operations inserted into the dataflow to implement the first cache placement plan based on an estimated reduction in a total execution cost for the dataflow when said first cache placement plan is implemented given an input dataset;
executing one or more operations of said dataflow and updating a count of references to said one or more executed operations to reflect remaining executions of said one or more executed operations in the dataflow;
determining, using at least one processing device, a current cache gain by updating the estimated reduction in the total execution cost for the dataflow of said first cache placement plan based on the updated count of references to said one or more executed operations;
determining, using the at least one processing device, an alternative cache placement plan for the dataflow following said execution based on the determined current cache gain and the updated count of references to said one or more executed operations, wherein said alternative cache placement plan comprises a combination of output datasets of a different subset of said operations in said dataflow to cache using one or more additional caching operations inserted into the dataflow to implement the alternative cache placement plan than said first cache placement plan; and
implementing, using the at least one processing device, said alternative cache placement plan when a difference between said current cache gain for said first cache placement plan and an estimated reduction in the total execution cost for the dataflow when said alternative cache placement plan is implemented satisfies one or more predefined threshold criteria.
15. The apparatus of claim 14 , further comprising the step of updating a cost model for said one or more executed operations based on 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.
16. The apparatus of claim 15 , wherein said determining of the current cache gain is further based on the cost model.
17. The apparatus of claim 14 , further comprising the step of removing one or more of said cached datasets from memory based on the number of references to the operations that generated said cached datasets following said updating of the number of references.
18. The apparatus of claim 14 , wherein said executing said one or more operations of said dataflow further comprises the step of updating a representation of an available computational infrastructure considering variations to the environment.
19. The apparatus of claim 14 , wherein said executing said one or more operations of said dataflow further comprises the step of updating a representation of the input dataset following said executing.
20. The apparatus of claim 14 , wherein said plurality of operations comprises a plurality of lazy-evaluated data transformations and a plurality of data actions that enact the lazy-evaluated data transformations.