Dynamic cache management for in-memory data analytic platforms

• US 10,467,152 B2
• Filed: 05/18/2016
• Issued: 11/05/2019
• Est. Priority Date: 05/18/2016
• Status: Active Grant

• Current Assignee: International Business Machines Corporation
• Original Assignee: International Business Machines Corporation
• Inventors: Li, Min, Wang, Yandong, Zhang, Li
• Primary Examiner(s): Lo, Ann J
• Assistant Examiner(s): Wei, Jane
• Application Number:
  US15/157,637
• Publication Number:
  US 20170337138A1
• Time in Patent Office: 1,266 Days
• CPC Class Codes: G06F 12/0246 in block erasable memory, e.g. flash memory
• CPC Class Codes: G06F 12/0862 with prefetch
• CPC Class Codes: G06F 12/121 using replacement algorithms
• CPC Class Codes: G06F 16/24552 Database cache management
• CPC Class Codes: G06F 2212/1021 Hit rate improvement
• CPC Class Codes: G06F 2212/1044 Space efficiency improvement
• CPC Class Codes: G06F 2212/6026 Prefetching based on access pattern detection, e.g. stride based prefetch
• CPC Class Codes: G06F 2212/7205 Cleaning, compaction, garbage collection, erase control

Abstract

At a cache manager of a directed acyclic graph-based data analytic platform, from each of a plurality of monitor components on a plurality of worker nodes, statistics are obtained for a plurality of tasks, including which of the tasks have been processed and which are in a task queue. Each of the tasks has at least one associated distributed dataset. Each worker has a distributed dataset cache. A current stage directed acyclic graph is obtained from a directed acyclic graph scheduler component. For a given one of the tasks which has been processed, and for which it is determined that no other ones of the tasks depend on the at least one distributed dataset associated with the given one of the tasks, the distributed dataset is evicted from a corresponding one of the distributed dataset caches.

Claims

1. A method comprising:

obtaining, at a cache manager of a directed acyclic graph-based data analytic platform, from each of a plurality of monitor components on a plurality of worker nodes of said directed acyclic graph-based data analytic platform, statistics for a plurality of tasks executing on said worker nodes, said statistics comprising which of said tasks have been processed and which are in a task queue, each of said tasks having at least one distributed dataset associated therewith, each of said worker nodes having a distributed dataset cache;

obtaining, at said cache manager, from a directed acyclic graph scheduler component of said directed acyclic graph-based data analytic platform, a current stage directed acyclic graph;

for a given one of said tasks which has been processed, and for which, based on said current stage directed acyclic graph, it is determined that no other ones of said tasks depend on said at least one distributed dataset associated with said given one of said tasks, evicting said distributed dataset associated with said given one of said tasks from a corresponding one of said distributed dataset caches;

monitoring, with said monitor components, memory usage statistics for said worker nodes of said directed acyclic graph-based data analytic platform; and

increasing a size of a given resilient distributed dataset cache of a plurality of resilient distributed dataset caches if said memory usage statistics indicate that corresponding ones of said tasks are using too little memory, said distributed dataset caches comprising said resilient distributed dataset caches, said memory usage statistics comprising garbage collection time.

2. The method of claim 1, wherein, in said step of obtaining said statistics and said step of evicting, said at least one distributed dataset comprises a resilient distributed dataset.

3. The method of claim 2, further comprising:

determining whether a corresponding one of said resilient distributed dataset caches is full; and

responsive to determining that said corresponding one of said resilient distributed dataset caches is not full, pre-fetching at least one additional resilient distributed dataset from disk into said corresponding one of said resilient distributed dataset caches, based on said current stage directed acyclic graph.

4. The method of claim 3, wherein said evicting comprises:

said cache manager instructing a block manager master of said directed acyclic graph-based data analytic platform to evict said resilient distributed dataset associated with said given one of said tasks from said corresponding one of said resilient distributed dataset caches; and

said block manager master of said directed acyclic graph-based data analytic platform instructing said corresponding one of said resilient distributed dataset caches to evict said resilient distributed dataset associated with said given one of said tasks.

5. The method of claim 4, further comprising storing said evicted resilient distributed dataset associated with said given one of said tasks on said disk.

6. The method of claim 5, wherein said pre-fetching of said at least one additional resilient distributed dataset from disk into said corresponding one of said resilient distributed dataset caches, based on said current stage directed acyclic graph, is carried out when input-output bandwidth utilization for a corresponding one of said worker nodes is smaller than a predetermined threshold.

7. The method of claim 6, further comprising pre-fetching at least a second additional resilient distributed dataset from said disk into said corresponding one of said resilient distributed dataset caches, based on said current stage directed acyclic graph, where said pre-fetching of said at least one additional resilient distributed dataset from said disk into said corresponding one of said resilient distributed dataset caches does not fill said corresponding one of said resilient distributed dataset caches.

8. The method of claim 2, further comprising

decreasing a size of another given one of said resilient distributed dataset caches if said memory usage statistics indicate contention between corresponding ones of said tasks and a corresponding one of said resilient distributed datasets.

9. The method of claim 8, wherein:

said memory usage statistics indicate that corresponding ones of said tasks are using too little memory when said garbage collection time does not exceed a threshold; and

said memory usage statistics indicate contention between corresponding ones of said tasks and a corresponding one of said resilient distributed datasets when said garbage collection time does exceed said threshold.

10. A directed acyclic graph-based data analytic platform comprising:

a plurality of worker nodes;

a plurality of monitor components on said plurality of worker nodes;

a plurality of distributed dataset caches on said plurality of worker nodes;

a directed acyclic graph scheduler component; and

a cache manager, coupled to said plurality of monitor components and said directed acyclic graph scheduler component, which;

obtains, from each of said plurality of monitor components, statistics for a plurality of tasks executing on said worker nodes, said statistics comprising which of said tasks have been processed and which are in a task queue, each of said tasks having at least one distributed dataset associated therewith;

obtains, from said directed acyclic graph scheduler component, a current stage directed acyclic graph;

for a given one of said tasks which has been processed, and for which, based on said current stage directed acyclic graph, it is determined that no other ones of said tasks depend on said at least one distributed dataset associated with said given one of said tasks, sends instructions to evict said distributed dataset associated with said given one of said tasks from a corresponding one of said distributed dataset caches;

wherein said monitor components monitor memory usage statistics for said worker nodes of said directed acyclic graph-based data analytic platform, and wherein said cache manager facilitates increasing a size of a given resilient distributed dataset cache of a plurality of resilient distributed dataset caches if said memory usage statistics indicate that corresponding ones of said tasks are using too little memory, said distributed dataset caches comprising said resilient distributed dataset caches, said memory usage statistics comprising garbage collection time.

11. The directed acyclic graph-based data analytic platform of claim 10, wherein said at least one distributed dataset comprises a resilient distributed dataset.

12. The directed acyclic graph-based data analytic platform of claim 11, wherein said cache manager:

determines whether a corresponding one of said resilient distributed dataset caches is full; and

responsive to determining that said corresponding one of said resilient distributed dataset caches is not full, pre-fetches at least one additional resilient distributed dataset from disk into said corresponding one of said resilient distributed dataset caches, based on said current stage directed acyclic graph.

13. The directed acyclic graph-based data analytic platform of claim 12, further comprising a block manager master coupled to said cache manager and said plurality of resilient distributed dataset caches, wherein:

said cache manager instructs said block manager master to evict said resilient distributed dataset associated with said given one of said tasks from said corresponding one of said resilient distributed dataset caches; and

said block manager master instructs said corresponding one of said resilient distributed dataset caches to evict said resilient distributed dataset associated with said given one of said tasks.

14. The directed acyclic graph-based data analytic platform of claim 13, wherein said cache manager stores said evicted resilient distributed dataset associated with said given one of said tasks on said disk.

15. The directed acyclic graph-based data analytic platform of claim 14, wherein said pre-fetching of said at least one additional resilient distributed dataset from disk into said corresponding one of said resilient distributed dataset caches, based on said current stage directed acyclic graph, is carried out when input-output bandwidth utilization for a corresponding one of said worker nodes is smaller than a predetermined threshold.

16. The directed acyclic graph-based data analytic platform of claim 15, wherein said cache manager pre-fetches at least a second additional resilient distributed dataset from said disk into said corresponding one of said resilient distributed dataset caches, based on said current stage directed acyclic graph, where said pre-fetching of said at least one additional resilient distributed dataset from said disk into said corresponding one of said resilient distributed dataset caches does not fill said corresponding one of said resilient distributed dataset caches.

17. The directed acyclic graph-based data analytic platform of claim 11, wherein said cache manager facilitates

decreasing a size of another given one of said resilient distributed dataset caches if said memory usage statistics indicate contention between corresponding ones of said tasks and a corresponding one of said resilient distributed datasets.

18. The directed acyclic graph-based data analytic platform of claim 17, wherein:

said memory usage statistics indicate that corresponding ones of said tasks are using too little memory when said garbage collection time does not exceed a threshold; and

said memory usage statistics indicate contention between corresponding ones of said tasks and a corresponding one of said resilient distributed datasets when said garbage collection time does exceed said threshold.

19. A non-transitory computer readable medium comprising computer executable instructions which when executed by a computer cause the computer to perform a method comprising:

obtaining, at a cache manager of a directed acyclic graph-based data analytic platform, from each of a plurality of monitor components on a plurality of worker nodes of said directed acyclic graph-based data analytic platform, statistics for a plurality of tasks executing on said worker nodes, said statistics comprising which of said tasks have been processed and which are in a task queue, each of said tasks having at least one distributed dataset associated therewith, each of said worker nodes having a distributed dataset cache;

obtaining, at said cache manager, from a directed acyclic graph scheduler component of said directed acyclic graph-based data analytic platform, a current stage directed acyclic graph;

for a given one of said tasks which has been processed, and for which, based on said current stage directed acyclic graph, it is determined that no other ones of said tasks depend on said at least one distributed dataset associated with said given one of said tasks, evicting said distributed dataset associated with said given one of said tasks from a corresponding one of said distributed dataset caches;

wherein said monitor components monitor memory usage statistics for said worker nodes of said directed acyclic graph-based data analytic platform, and wherein said cache manager facilitates increasing a size of a given resilient distributed dataset cache of a plurality of resilient distributed dataset caches if said memory usage statistics indicate that corresponding ones of said tasks are using too little memory, said distributed dataset caches comprising said resilient distributed dataset caches, said memory usage statistics comprising garbage collection time.

20. The non-transitory computer readable medium of claim 19, wherein, in said method step of obtaining said statistics and said method step of evicting, said at least one distributed dataset comprises a resilient distributed dataset.