Anticipatory warm-up of cluster resources for jobs processed on multiple cluster nodes

Systems and methods are disclosed for reducing latency in processing data sets in a distributed fashion. A job-queue operable for queuing data-processing jobs run on multiple nodes in a cluster may be communicatively coupled to a job analyzer. The job analyzer may be operable to read the data-processing jobs and extract information characterizing those jobs in ways that facilitate identification of resources in the cluster serviceable to run the data-processing jobs and/or data to be processed during the running of those jobs. The job analyzer may also be coupled to a resource warmer operable to warm-up a portion of the cluster to be used to run a particular data-processing job prior to the running of the job. In some embodiments, mappers and/or reducers may be extracted from the jobs and converted into compute node identifiers and/or data units identifying blocks for processing, informing the warm-up operations of the resource warmer.

FIELD OF THE INVENTION

This invention relates to the handling of data-processing jobs and more particularly to the handling of data-processing jobs run on partitioned subsets of data in parallel on multiple nodes of a cluster of nodes.

BACKGROUND OF THE INVENTION

The ability to process large amounts of data within shorter periods of time is growing in importance. For one thing, more and more data is being produced as more mobile technologies with larger information-sensing capacities are spreading, more people interact over the internet via social media, and more devices become equipped with smart technologies, among other reasons. Some of such sources include web searches, email, logs, internet marketing, geospatial data, financial data, space research, healthcare data, scientific research, and more. Furthermore, the world's ability to store data is increasing, according to one study, for example, the world's per-capita data storage capacity has been doubling every 40 months since the 1980s.

Not only are larger and larger data sets becoming more common, but the processing of such data sets is becoming increasingly important in more areas. Large data sets are frequently involved in several areas of research from meteorology, to genetics, to many other fields of research requiring complex modeling. The ability to process large amounts of data has also become important in more every-day applications from finance, marketing, e-commerce, social media, and internet searches. However, the growing size of data sets that must be processed to support functionalities in these and other areas is often so large that traditional processing approaches are either impractical, or simply impossible.

To make possible the processing of large data sets, often the presence of multiple chunks that can be processed independently is leveraged to break up the job for parallel processing. Parallel processing can occur on several nodes, or machines, simultaneously, greatly speeding up the underlying job. However, sending large amounts of data across a network for processing can introduce its own complexities, take time, and occupy large amounts of bandwidth within a network. Many other complex problems arise in distributed processing generally, such as the details of the parallel and distributed processing itself and the handling of errors during processing.

In the late 1990s and early 2000s, in the process of addressing problems associated with indexing the massive amounts of information that its search engine relies on, Google noticed several features common to many big-data processing problems. As a result, it developed a distributed file system, the Google File System (GFS), that provides a framework for breaking-up and storing large data sets across physically independent commodity machines interlinked by a network and lends itself to the processing of those large data sets. Additionally, Google developed a framework, known as the MapReduce framework, for processing distributed data sets implemented in two main phases. These main phases comprise a map phase that takes input files with key value pairs and produces intermediate files with new key value pairs and a reduce phase that combines values from common keys in the intermediate files.

In 2003 and 2004 Google published its GFS and MapReduce framework respectively in two papers. These papers, together with a lot of collaboration from large corporations and other contributors have led to open source versions of the foregoing system and framework, respectively referred to as the Hadoop Distributed File System (HDFS) and Hadoop MapReduce engine, or collectively as simply Hadoop. Whether in terms of Google's version, Hadoop, or some other version, these distributed file systems and MapReduce frameworks have proved a boon to big data processing, in such areas as search, analytical functions, transformations, aggregations, data mining, among others, and have become ubiquitous in the field. However, additional demands, such as those of larger data sets and needs for quicker processing times, require additional innovations that can sit atop Hadoop-like approaches and potentially other approaches to distributed processing. The following description and claims set forth such innovations.

DETAILED DESCRIPTION

Although distributing processing duties for large data-processing jobs across several nodes can make the running of such jobs possible and/or practical, getting the data to those nodes can introduce complexities and slow down the jobs. Hadoop and Hadoop-like approaches (hereinafter “Hadoop approaches”) provide a framework that reduces these complexities by providing frameworks that transparently and intelligently handle the storage of and processing of large data sets in a way that, in a sense, takes the processors to the data, as opposed to taking the data to the processors. However, the very strength of these approaches in reducing data handling obscures additional approaches where improvements can be made in terms of the details of the way data is handled.

Hadoop approaches are designed to place processors in close proximity to the data blocks which they process, which likely reside on or near those nodes. However, such data blocks still need to be loaded into caches at those nodes. Additionally, although the distances are usually reduced by Hadoop approaches, some data blocks may need to be transferred to the nodes where they are processed. In some instances, which extend the capabilities of Hadoop approaches, data blocks may even reside outside the cluster where they are processed, for example in the cloud. Typical data block sizes involved in Hadoop approaches are 128 Mega Bytes (MBs) and above, indicative of the delays that warm-up activities like loading caches and transferring data blocks can produce.

The present application discloses innovations to reduce these latencies. For example, a system to reduce latencies may involve a job analyzer, which may be implemented as a module as discussed below, that is communicatively coupled to a job queue. The job queue may be operable for queuing data-processing jobs to be run in a distributed fashion on multiple nodes in a cluster of nodes. The job analyzer may be operable to read one or more data-processing jobs in the job queue and/or extract characterization information from one or more data-processing jobs. The characterization information extracted by the job analyzer may characterize one or more resources in a cluster of nodes for processing such jobs. The one or more resources may be serviceable and/or designated to run a given data-processing job, or pool of data-processing jobs. Additionally, or in the alternative, such characterization information may characterize, identify, and/or facilitate identification of data to be processed by the given data-processing job, or pool of data-processing jobs.

Such a system may also include a resource warmer, which may also be implemented as module, communicatively coupled to the job analyzer and to the cluster. The resource warmer may be operable to warm up a portion of the cluster to be used to run the given data-processing job or pool of data-processing jobs. The portion may be identified by the characterization information extracted from the given data-processing job, or pool of data-processing jobs. Once the portion of the cluster has been warmed-up, the given data-processing job, or pool of data-processing jobs, may be run without the latency of the previously accomplished warm-up period.

In some examples, the system may also include a conversion model communicatively coupled to the job analyzer. The conversion module may be operable to convert characterization information, which may comprise a set of mappers and/or a set of reducers, into a set of compute node identifiers for nodes at which the given data-processing job, or pool of data-processing jobs, is to be run and/or a set of data units identifying data blocks/replicas to be processed during the data-processing job. In such examples, the resource warmer may receive the set of compute node identifiers and/or the set of data units. Furthermore, in certain examples, the resource warmer may warm up the portion of the cluster by provisioning one or more of the data blocks/replicas identified by the set of data units to one or more nodes in the cluster. In some embodiments, provisioning data blocks/replicas may further involve loading them in one or more caches at the nodes indicated by the compute node identifiers.

To provide a more thorough account of embodiments of the present innovations, it is helpful to provide some additional contextual information about approaches to processing large data sets, such as Hadoop approaches. Therefore,FIGS. 1 and 2are provided that explain the two key concepts involved in Hadoop approaches. These two concepts are automated, distributed filing systems, like GFS and HDFS, and the MapReduce framework/engine.

Referring toFIG. 1, an Automated, Distributed Filing System (ADFS)10is depicted, consistent with examples such as GFS or HDFS, as applied in Hadoop approaches. The ADFS10may be implemented in software and/or firmware and/or the like residing at various hardware components12a-12g,14with the use of modules, as discussed below. The hardware components12a-12g,14may comprise commodity hardware and/or specially purposed hardware. The various hardware components12a-12g,14may provide the infrastructure for various data nodes16a-16gand a name node18, to be discussed in greater detail below, which comprise a cluster20a.

The ADFS10may be configured to receive a large data file, or data set,22and split the large data set22into multiple blocks24a-24n(also referred to as data blocks) for storage on multiple data nodes16, thereby increasing the potentially available storage capacity of the ADFS10. In some examples, the data set22may include multiple files that share a logical grouping. The blocks16may be fairly large in size, for example, from tens of megabytes to gigabytes. To provide redundancy, in case a data node16on which a given block24is stored fails and/or to provide greater access to the blocks24, the blocks24may be replicated to produce a number of replicas26a-c,26d-f,26n-pof each block24a,24b,24n. As used in this application, the term block24is synonymous with any replica26carrying the same data. Although the example depicted inFIG. 1depicts three replicas26for each block24, as can be appreciated, any number of ratios of replicas26to different blocks24may be used. These replicas26may then be stored at the various data nodes16, which may store one or more blocks24and/or replicas26.

The ADFS10may be configured for fault tolerance protocols to detect faults and apply one or more recovery routines. To assist in fault tolerance, the data nodes16and the name node18may be configured with a web server. Also, the ADFS10may be configured to store blocks/replicas24/26as close to processing logic on the hardware components12a-12g,14as possible so that a data-processing job can be run on the blocks24pertaining to the data set22with minimal block transfers. Also, multiple different data sets22, may be stored on the cluster18in this way.

The name node18may fill a role as a master server in a master/slave architecture with data nodes16a-gfilling slave roles. Since the name node18may manage the namespace for the ADFS10, the name node18may provide awareness, or location information, of the various locations at which the various blocks/replicas24/26are stored. For example, the name node18may maintain a directory tree of all the blocks/replicas24/26in ADFS10and may track where the various blocks/replicas24/26are stored across the cluster20a. Furthermore, the name node18may determine the mapping of blocks/replicas24/26to data nodes16. Under the direction of the name node18, the data nodes16may perform block creation, deletion, and replica functions.

Although only seven data nodes16a-gare depicted inFIG. 1for purposes of illustration, as can be appreciated, any number of nodes16, including numbers in the thousands, are possible. As described with respect toFIG. 1, an ADFS10may provide automation and infrastructure for placing a large data set22on several data nodes16as blocks/replicas24/26, spread out on physically different machines/nodes12,14,16,40,42and even across data centers, in effect, taking the processing logic close to the data. In the process, the ADFS10may set the stage for various approaches to distributed and/or parallel processing. For example, as discussed below, the locational awareness of the ADFS10, may be leveraged to provide more efficient approaches to distributed and/or parallel processing. The following figure is used to provide context relevant to the innovations explained herein of one such approach.

Referring toFIG. 2, elements of a typical MapReduce engine28are depicted. A MapReduce engine28may implement a map phase30and a reduce phase32, described in further detail below. A MapReduce engine28may comprise additional phases, such as a combination phase and/or a shuffle phase34, also described below, between the map phase30and the reduce phase32.

A master/slave architecture, as discussed with respect to the ADFS10in terms of the relationship between the name node18and the data nodes16, may be extended to the MapReduce engine28in terms of a job tracker36, which also may be implemented as a resource manager and/or application master, in a master role and one or more task trackers38a-e, which also may be implemented as node managers, in a slave role. Together, the job tracker36and the name node18may comprise a master node40, and individual parings of task trackers38a-eand data nodes16a-emay comprise individual slave nodes42a-e. In some examples, the master node40may also be configured with its own data node16and task tracker38.

Consistent with the concept of distributed, parallel processing, a data-processing job may involve multiple component tasks. The job tracker36may schedule and monitor the component tasks, coordinating the re-execution of a task where there is a failure. The job tracker36may be operable to harness the locational awareness provided by the name node18to determine the nodes42/40on which various data blocks/replicas24/26pertaining to a data-processing job reside and which nodes42/40and/or machines/hardware and/or processing logic12are nearby.

The job tracker36may further leverage such locational awareness to optimize the scheduling of component tasks on available slave nodes42to keep the component tasks as close to the underlying data blocks/replicas24/26as possible. In the event that the requisite processing logic on a node42on which a relevant block/replica24/26resides is unavailable, the job tracker36may select a node42on which another replica26resides, or select a node42in the same rack, or otherwise geographically proximate, to which to transfer the relevant block/replica24/26, reducing the load on a network backbone. Owing to its monitoring and rescheduling capabilities, the job tracker36may reschedule a component task that fails.

The component tasks scheduled by the job tracker36may involve multiple map tasks and reduce tasks to be carried out on various slave nodes42in the cluster20a. Individual map and reduce tasks may be overseen at the various slave nodes42by individual instances of task trackers38residing at those nodes42. Such task trackers38may spawn separate Java Virtual Machines (JVM) to run their respective tasks and/or may provide status updates to the job tracker36, for example and without limitation, via a heartbeat approach. AlthoughFIG. 2only depicts five such nodes42a-3efor purposes of illustration. However, any number of nodes may be involved, easily including numbers in the thousands.

During a map phase30, a first set of slave nodes42a-cmay be utilized and/or dedicated for map tasks. For example, a data-processing job may involve processing the various blocks/replicas24/26that make up a data set22. Although the tasks may be run in parallel, the processing of each block/replica24/26at the various slave nodes42a-cmakes up an individual map task. To execute a map task, a task tracker36may apply a mapper44to a block/replica24/26pertaining to a job being run.

For example, the job tracker36may assign a first task tracker38ato apply a first mapper44ato a first data block24apertaining to a data-processing job being run by the job tracker36. In some examples, the job tracker36may provide the first mapper44ato the first task tracker38a. In other examples, the first mapper44a, or a portion thereof, may already reside at the slave node42aat which the task tracker38aalso resides. The first data block24amay reside at a first data node16athat also stores several other blocks/replicas24/26. The first task tracker38amay select the appropriate data block24afrom among the other blocks/replicas24/26in a storage volume46aused to maintain the first data node16aat the first slave node42a. A storage volume46may comprise a disk on hardware12supporting a slave node42, a solid-state drive, or any other technology for storing data.

Data blocks24a,24f,24h,24pcorresponding to a given data-processing job being run by the job tracker36inFIG. 2are depicted with boxes having white, diagonal stripes on a black background, similar to the one labeled16a. As can be appreciated, these blocks24a,24f,24h,24pmay be stored in various locations relative to other blocks/replicas24/26pertaining to other data-processing jobs and/or data sets22, indicated by alternative shading patterns. Also, as depicted with respect to the first data node16a, a single data node16may store multiple blocks/replicas24/26pertaining to a given data-processing job.

A mapper44may be executable to treat an input data block/replica24/26as a set of key-value pairs. A mapper44may proceed to process the block/replica24/26by producing a set of intermediate key-value pairs written in an intermediate record48. An illustrative, but non-limiting example, may be provided in the context of search engines and something like the page-rank algorithm with respect to which, at least in part, MapReduce engines28were first developed. In such a context, a web crawler may collect a large data set22that stores web pages searched and links between them. In such an example, an ADFS10, such as GFS, may split this data set22into many blocks/replicas24/26, each of which may become an input for a mapper44.

A mapper44, in such an example, may view the input block/replica24/26as a set of key-value pairs, where a key may correspond to a source page and a value may correspond to a target page to which a link on the source page points. The mapper44may run as a function applied to each key-value pair, in the process counting the number of pages that link to a given target page. In such an example, the target page, which corresponds to the value parameter of the input, may become the key parameter in a new key-value pair produced by the mapper44and recorded in an intermediate record48. In such example, the value parameter of the new key-value pair, may be the number of links to the target page from various source pages. An intermediate record48may, therefore, be a set on new key-value pairs generated by applying a map function to a set of key-value pairs in the input block/replica24/26. In the example, an intermediate record48would correspond to a set of new key-value pairs comprising a set of target pages paired with counts of links to those target pages.

However, after a map phase30, the results for a data-processing job may be scattered across several different intermediate records48a-d. The reduce phase32may be applied to bring these scattered results together. A shuffle phase34may be implemented to facilitate the reduce phase32. During the shuffle phase34, the intermediate records48may be shuffled so that intermediate records with common keys, e.g.48a,48band48c,48d, may be directed across a network50, which may connect various nodes40,42in a cluster20a, to nodes42d,42ewith the same reducers52, e.g.,52aand52brespectively.

The job tracker36may schedule a reducer phase32to task trackers38d-eon slave nodes42d-ewithin a cluster20a. Data nodes16d-eat those slave nodes42d-emay receive intermediate records48a-dover a network50for storage in corresponding storage volumes46d-e. Individual task trackers38d-emay apply a reducer52a/52bbto the intermediate records48a-b/48c-dstored by the data node16d/16eat the corresponding slave node42d/42e. In some examples, the job tracker36may provide a reducer52ato a first task tracker38d. In other examples, the reducer52a, or a portion thereof, may already reside at the slave node42dat which the task tracker38dalso resides.

Although a map phase30and a reduce phase32may run in two distinct phases, in some examples, they may be overseen by a common task tracker38d. Conversely, a map phase30and a reduce phase32may be overseen by two distinct task trackers32a,32d. Even though reducers52may not start until all mappers44are complete, shuffling may begin before all mappers44are complete.

Although at least one mapper44runs on each node42that has data24for the job, reducers52are not constrained to the nodes42on which they run by considerations of data locality. Reducers52may be assigned to run in the network50with flexibility. Hence, the job tracker36may assign tasks to less loaded task trackers38. For example, the load may be assigned toward the achievement of an evenly distributed load. One approach to evenly distribute the load may involve assigning a reducer52to run on all available nodes42if there are enough reducers52in the job. Also, the intermediate records/files48(mapper output) may avoid the overhead associated with replication associated with the data blocks24.

A reducer52may run on multiple intermediate records48to produce an output record54. An output record54generated by such a reducer52may group values associated with common keys to produce combined values. Picking up with the previous example, the counts for links to the various target pages in the various intermediate records48could, for example, be summed. In such a way, one or more reduce phases32may be used to combine the results from several different mappers44in different intermediate files48to create a result for a data-processing job. Due to the way in which individual mappers44and/or reducers52operate at individual nodes42/40, the term ‘mapper’ and/or ‘reducer’ may be used to refer to the nodes42at which individual instances of mappers44and/or reducers52are implemented. As can be appreciated, a MapReduce engine28may compose any different number of map phases30and/or reduce phases32in any different combination, where a reduce phase32follows one or more map phases30, to run a given data processing job.

The functions involved in implementing such an ADFS10, some other distributed filing system, a Hadoop engine28, some other approach for processing distributed data, and/or the innovations discussed herein may be handled by one or more subsets of modules. With respect to the modules discussed herein, aspects of the present invention may take the form of an entirely hardware embodiment, an entirely software embodiment (including firmware, resident software, micro-code, etc.), or an embodiment combining software and hardware aspects that may all generally be referred to herein as a “module.” Furthermore, aspects of the presently discussed subject matter may take the form of a computer program product embodied in any tangible medium of expression having computer-usable program code embodied in the medium.

With respect to software aspects, any combination of one or more computer-usable or computer-readable media may be utilized. For example, a computer-readable medium may include one or more of a portable computer diskette, a hard disk, a random access memory (RAM) device, a read-only memory (ROM) device, an erasable programmable read-only memory (EPROM or Flash memory) device, a portable compact disc read-only memory (CDROM), an optical storage device, and a magnetic storage device. In selected embodiments, a computer-readable medium may comprise any non-transitory medium that may contain, store, communicate, propagate, or transport the program for use by or in connection with the instruction execution system, apparatus, or device.

Computer program code for carrying out operations of the present invention may be written in any combination of one or more programming languages, including an object-oriented programming language such as Java, Smalltalk, C++, or the like and conventional procedural programming languages, such as the “C” programming language or similar programming languages. Aspects of a module, and possibly all of the module, that are implemented with software may be executed on a micro-processor, Central Processing Unit (CPU) and/or the like. Any hardware aspects of the module may be implemented to interact with software aspects of a module.

Referring toFIG. 3, an implementation of virtual clusters56consistent with a Hadoop approach is depicted. Although the virtual clusters56depicted inFIG. 3are discussed below in terms of virtual clusters56consistent with a Hadoop implementation, any type of virtual clusters56, such as open stack clusters56are possible. By adding one or more layers of abstraction, the abilities of the Hadoop approach discussed above may be extended through virtualization. Such innovations may provide a configuration platform58operable to configure any number of different virtual clusters56with any number of data nodes16, name nodes18, job trackers36, and task trackers38on a wide variety of physical hardware60.

Such additional layers of abstraction may involve a new name space62. The new name space62may be configured consistent with an abstract file system64, which may reside with the configuration platform58or elsewhere, operable to store data blocks/replicas24/26in a virtual cluster56. Additionally, safeguards and/or redundancies may be built into this new abstract file system64to address problems typical of flash memory, such as, without limitation, data retention problems, to allow data blocks/replicas24/26to be stored at the physical layer on Solid State Drives (SSD)66implementing such memory. Additional elements and/or abstractions provided to facilitate and configure virtual clusters56may be implemented on top of an existing Hadoop approach implementation in the user space, as opposed to the kernel space, of the Hadoop implementation.

The abstractions and/or additional elements discussed above may facilitate implementation and/or removal of virtual clusters56and/or additions to and/or removal of nodes16/18and/or trackers36/38from such virtual clusters56on the fly through the configuration platform58. Virtualization may facilitate the creation of several nodes40/42on a single physical node12. Additionally, the flexibility of virtualization may allow Hadoop implementations on more heterogeneous clusters20, involving SSDs66and/or Hard Disc Drives (HDD) in a Storage Area Network (SAN)68for data storage functions communicatively coupled over a physical network70and protocols such as Internet Small Computer System Interface (iSCSI). Additional details used to explain the implementation of virtual clusters56can be found in the provisional application mentioned above: U.S. Provisional Application Ser. No. 61/876,045, entitled Virtual Hadoop Clusters and filed on Sep. 10, 2013.

Such flexibility can pave the way for implementation of new technologies, such as cloud technologies in the Hadoop framework. Indeed such flexibility could literally lead to the use of physical resources anywhere in the world to support virtual hadoop clusters56. Such flexibility, however, may have the potential to increase latencies already involved in running data-processing jobs. For example, retrieving a data block/replica24/26from a SAN68or a cloud service may take time. Such latencies required to prepare a cluster20/56for operations may be classified under the category of “warm-up” operations.

Operations to warm up a cluster20/56generate latency issues whether the cluster20/56is a virtual cluster56or not. Such warm-up operations may include loading a data block/replica24/26to a cache for processing, which may take significant time for sizes involved in distributed processing, such as those from 64 megabytes to gigabytes. Where the requisite block/replica24/26does not reside on a node42/40with available processing logic, warm-up latencies may be incurred in transferring the block/replica24/26to the requisite node via a network protocol. Such transfers may involve warming up relevant network resources. Additionally, one or more task trackers38may require time to spawn one or more JVMs to implement mappers44and/or reducers52, which may also need to be provisioned to the requisite nodes42/40. The foregoing is not intended as an exhaustive list of potential warm-up operations, but is provided only by way of illustration. As can be appreciated, there may be several additional warm-up operations that may contribute to the latency of running a data-processing job. The discussion with respect to the following figures explains innovations to reduce and/or remove warm-up latencies.

Referring toFIG. 4, a job analyzer72, which may be implemented as a module, is depicted. The job analyzer72may be implemented outside of a Hadoop implementation, such as in the user space, while being operable to interface with a job queue74. The job queue74may be operable for queuing data-processing jobs76a-fto be run in a distributed fashion on multiple nodes42/40in a cluster of nodes20/56. In some examples, the job queue74may reside at a job tracker36pertaining to a MapReduce layer78sitting atop an ADFS layer80to which a name node18may pertain. Where a job tracker36maintains scheduler data structures and allows for a pluggable scheduler, the job analyzer72enhancement may be provided with the job tracker36without having to change a corresponding Hadoop specification.

Being communicatively coupled with the job queue74, the job analyzer72may be operable to read one or more data-processing jobs76in the job queue74. For example, the job analyzer72may utilize information about a file structure used to store such data-processing jobs76at the master node40to read the data-processing jobs76. Additionally, the job analyzer72may be operable to retrieve characterization information from one or more data-processing jobs76. The characterization information for a given data-processing job76may characterize one or more resources in a cluster20/56serviceable to run the data-processing job76and/or data units, such as one or more data blocks/replicas24/26, to be processed during the data-processing job76.

Referring toFIG. 5, a job analyzer72is depicted as operable to traverse82across data-processing jobs76a-ffor purposes of reviewing the data-processing jobs76a-fin the job queue74. By way of example and not limitation, the data-processing jobs76a-fin the job queue74may be processed in a First In First Out (FIFO) manner such that the data-processing job76fon the right side of the queue74may be interpreted as both the first data-processing job76into the job queue74and the next data-processing job76to be processed84. As can be appreciated, additional approaches, apart from FIFO, for determining when to run data-processing jobs in the queue may be implemented with the disclosure provided herein. Conversely, the data-processing job76aon the left may be interpreted as both the last data-processing job76into the job queue74and the last data-processing job76, currently in the queue74, that will be processed84.

The job analyzer72is depicted examining the second data-processing job76ein line for processing84, indicated by the horizontal-line pattern. The job analyzer72may be operable to extract characterization information from this data-processing job76e. The characterization information may include a set of mappers44a-cand/or a set of reducers52a-bfor the data-processing job76e.

The master node40on which the job queue74may reside may pertain to a cluster20bof nodes40/42a-n. Each node40/42in the cluster20bmay have data storage capability, in terms of a data node16, data processing capability, directed by a task tracker38, and/or software for coordinating18/36one or more mapping and/or reducing operations within the cluster20b. The overall data-processing system may include a distribution module (not depicted) operable to divide an input file22into multiple splits, or blocks,24and/or store at least one copy of each of the multiple splits24as data blocks24at multiple locations selected from the cluster20band/or a backend storage device68, or cloud service. The name node18on the master node40may receive locational awareness, or location information, identifying the multiple locations of the data blocks/replicas24/26.

As can be appreciated, the mappers44a-cand/or reducers52a-bextracted from the data-processing job76emay correspond to individual nodes42in the cluster20b. To illustrate this correspondence, the pattern of white, diagonal stripes on a black background used to illustrate the set of three mappers44a-cis echoed by the same pattern on three slave nodes42b,42g,42nthat correspond to these three mappers44a-c. The mappers44a-cmay correspond to these three slave nodes42b,42g,42ninasmuch as the mappers44a-cmay be implemented on these nodes42b,42g,42n.

Similarly, the pattern of vertical stripes used to illustrate the set of two reducers52a-bis echoed by the same pattern on two slave nodes42d,42hthat correspond to these two reducers52a-b. In some examples, the reducers52a-bmay be implemented on these nodes42d,42h. As the remaining nodes42a,42c,42e,42f,42kand42min the cluster20bmay not be utilized in the map phase30, or the reduce phase32, they are left blank.

In some examples, but not in all examples, a schedule control module86may reside at the job analyzer72and/or be communicatively coupled to the job analyzer. The schedule control module86may be operable to analyze control information, such as, but not limited to, mappers44and/or reducers52extracted from multiple data-processing jobs76in the job queue74. The schedule control module86may further be operable to optimize a pool of data-processing jobs76to be run concurrently. The schedule control module86may determine the optimization of data-processing jobs76to be run concurrently based on data availability. Data unit availability may be defined by a data block/replica24/26being available where the data block/replica24/26to be processed by one data-processing job76in the pool is loaded in a cache for processing by a node42/40for another data-processing job76.

Referring toFIG. 6, a conversion module88and a resource warmer90, which may be implemented as a module, are depicted in relation to the job analyzer72. The resource warmer may be communicatively coupled to the job analyzer72and to the cluster20b. The resource warmer90, which may be implemented separately from the master node40while being communicatively coupled to the master node40, may be operable to warm up a portion of the cluster20bto be used to run a data-processing job76e. The portion of the cluster that the resource warmer90may warm up may be a portion of the cluster20bidentified from the characterization information extracted from the data-processing job76eby the job analyzer72, which may be in communication with the resource warmer90. The resource warmer90may warm up a portion of the cluster20b/56for a single data-processing job76eand/or multiple data-processing jobs76from the job queue74. Where there are insufficient cache resources to fully warm up the cluster20b/56for all of the jobs76in the job queue76, a selection may be made of a subset of the jobs76, which may or may not be based on a processing order of the jobs76in the job queue74.

The conversion module88may assist in the identification of the portion of the cluster20bto be warmed-up. The conversion module88may be communicatively coupled to the job analyzer72and/or to the resource warmer90. Although the conversion module is depicted inFIG. 6as residing at the job analyzer72, the conversion module88may be located elsewhere. The conversion module88may be operable to convert the characterization information into a set of compute node identifiers92a-eat which a given data-processing job76eis to be run.

The set of compute node identifiers92a-emay correspond to nodes42b,42d,42g,42h,42nat which mapping and/or reducing functions for the data-processing job76emay occur. Inasmuch as the set of mappers44a-cmay correspond to certain slave nodes42b,42g,42nin the cluster20bof nodes at which the mappers44a-cmay be implemented, the conversion module88may be operable to determine these slave nodes42b,42g,42nand designate these as compute node identifiers92a-c. In some examples, the conversion module88may be operable to determine a proximate and available node42for a mapper44where the corresponding data block/replica24/26does not reside at a slave node42with the requisite processing logic available.

In some embodiments, the conversion module88may also be operable to determine slave nodes42d,42hat which the reducers52a-bmay be implemented and designate these as compute node identifiers92d-e. The set of compute node identifiers92a-emay carry additional information to indicate whether a given compute node identifier92pertains to a map operation, a reduce operation, or some other processing functionality. The set of compute node identifiers92a-emay be limited solely to compute node identifiers92a-cdesignated for map functions, solely limited to compute node identifiers92d,92edesignated for reduce functions, or compute node identifiers92a-edesignated solely for both. The set of compute node identifiers92may also include compute node identifiers92designated for other data-processing functionalities.

The conversion module88may be operable, in the alternative or additionally, to convert the characterization information into a set of data unit identifiers94a-dindicating blocks/replicas24/26to be processed during the given data-processing job76e. Either the conversion module88, the job analyzer72, the resource warmer90, some other module, or any combination of the foregoing may be operable to retrieve multiple locations at which a set of data blocks/replicas24/26to be processed in accordance with a given data-processing job76eare stored. These multiple locations may be retrieved from the name node18based on information in the characterization information, such as, but not limited to, various mappers44and/or reducers52.

The name node18may be operable to store location information about locations where data blocks/replicas24/26reside within the cluster20baccording to a distributed filing system, such as a system consistent with the ADFS10described with respect toFIG. 1. The location information may be generated and/or stored in the name node18by the distributed filing system. Such location information may be provided at one or more of various levels of granularity, such as, for example, a geographic, a data center, a sector of a data center, a rack, a machine, and/or a particular data node at which, or in which the data block/replica24/26is stored.

As stated above, either the conversion module88, the job analyzer72, the resource warmer90, some other module, or any combination of the foregoing may be operable to access the location information from the name node18. One or more of these entities may also be operable to apply information from the characterization information to the location information to determine where data blocks/replicas24/26to be processed during the data-processing job76ereside. These data blocks/replicas24/26may be indicated by the characterization information extracted from the data-processing job76e.

The job analyzer72, conversion module88, resource warmer90, or some other entity, may also be operable to select nearby nodes42at which to process the data blocks/replicas24/26. Where a data block/replica24/26resides at node42/40with the requisite processing logic available to process the block/replica24/26, that node42/40may be selected. If the requisite processing logic is not available, one or more algorithms or routines may be applied to select a proximate node with the requisite processing logic available. Status reports may be served from the various nodes42/40to determine availability of processing logic.

In some examples, one or more data blocks/replicas24/26, making up a portion of the data to be processed during the running of a data-processing job76e, or group of data-processing jobs76scheduled to be run concurrently, may reside on a backend storage device68, such as, without limitation, those discussed with respect toFIG. 3. Although the backend storage device68may be outside the cluster20b, it may also be communicatively coupled to the cluster20b. The routines and/or algorithms discussed above may be employed to select nodes42/40for processing such data blocks/replicas24/26within the cluster20b.

The job analyzer72may send a message96to the resource warmer90with characterization information extracted from one or more data-processing jobs76. The characterization information may include, without limitation, mappers44, reducers52, compute node identifiers92, and/or data unite identifiers94. In some examples, the resource warmer90may receive a set of compute node identifiers92and/or the set of data units identifier94.

As stated, the resource warmer90may be operable to warm up aspects of the cluster20bindicated by the characterization information as relevant to processing a given data-processing job76e, or group of data-processing jobs76to be run concurrently. The resource warmer90may be operable to warm up these aspects in advance of processing the given data-processing job76e, or group of data-processing jobs76. In this way, the resource warmer90may contribute to removing latencies associated with performing warm-up operations.

Referring toFIG. 7, various potential functionalities of the resource warmer90are depicted. In some examples, the resource warmer90may be operable to (1) warm up a portion of the cluster20b. InFIG. 7, the depicted portion only includes a single slave node42nfor purposes of illustration. However, the actual portion may include many more aspects for the cluster20b, such as slave nodes42band42gin the cluster20b.

The resource warmer90may warm up the portion of the cluster20b, at least in part, by (2) provisioning a data block/replica24/26to a node42nin the cluster20b. In other words, the warm up of the aspects of the cluster20bindicated by the characterization information may involve provisioning the set of data blocks/replicas24/26from the multiple locations at which they are stored to caches100at the selected slave nodes42b,42g,42n, which may correspond to the compute node identifiers92a-c, where the data-processing job76e, or pool of jobs76, runs. In some examples, where a data block/replica24/26does not reside at the node42/40where it is to be processed, provisioning the block/replica24/26may involve warming up a relevant network resource in the cluster20bused to relay block/replica24/26to the node42/40at which the processing will take place. In some examples, the warm-up process may involve bringing one or more blocks/replicas24/26into the cluster20bfrom the backend68. In certain examples, provisioning a block/replica24/26may further involve loading a data block/replica24/26to be processed during a data-processing job76e, or pool of jobs76, into a cache100for processing logic102ain the cluster20bfor processing. In certain examples, to avoid data transfer over the network50by the use of distributed read to warm up a data node16cache100, the resource warmer90may be provided with a cache warming Application Programming Interface (API). By way of example and not limitation, some examples of data-node reads may include consistency checks, health monitoring, and/or the like.

The resource warmer90may identify and/or receive identification of the data blocks/replicas24/26in the set to be processed based on the characterization information extracted by the job analyzer72. The resource warmer90and/or the warm-up module98therein may determine the nodes42, such as the enlarged slave node42ninFIG. 7, on which the set of data blocks/replicas24/26reside and may further select the relevant data block/replica24pfrom among several data blocks/replicas24/26in a storage volume42nof a relevant data node16n. In some examples, the resource warmer90may rely on a data unit94aidentified by the conversion module88to identify and/or select the appropriate data block/replica24/26. In some examples, raw mappers44and/or reducers52may be used. In other examples, a compute node identifier92amay be used and/or other characterization information. In some examples, once a data block/replica24phas been identified, the job analyzer72, conversion module88, resource warmer90, or some other entity, may consult a name node18to determine the location, or general location, of the data node16nat which the data block/replica24presides.

In some examples, where multiple mappers44and/or reducers52are stored at a given node42/40, information from an extracted mapper44aand/or a compute node identifier indicator92amay be used to select (3) the appropriate mapper44, indicated by the diagonal background pattern inFIG. 7, to apply at the node42n. Certain nodes42/40at which mappers may be implemented may (4) provide a status update when the requisite data block/replica24/26for a given data-processing job76ehas been provisioned and/or loaded into a cache100to the resource warmer90and/or the job analyzer. In some examples, the status update may indicate the status of other warm-up operations, such as those discussed above, and/or only indicate a partial completion of a warm-up operation.

The resource warmer90may send an update message104with status-update information to the job analyzer72, where the job analyzer does not receive such information directly. An advance module106, which may be termed a process module106, may reside with the job analyzer72and/or be communicatively coupled to the job analyzer72. The advance module106may be operable to receive one or more status updates.

Upon a determination that the portion of the cluster20brelevant to the data-processing job76e, or pool of data-processing jobs76, is warmed up, or has reached a level of warm-up that satisfies a predetermined threshold, depending on the example, the advance module106may (6) advance, or act to advance, the relevant data-processing job76e, or pool of data-processing jobs76selected by the scheduling module98, out of the queue74, regardless of its, or their, position with respect to a FIFO scheduling order, or some other approach for determining when to run data-processing jobs76in the queue74. In some examples, the advance module106may be operable to advance (6) the data-processing job76e, or pool of data-processing jobs76, for (7) processing upon the warm-up, of the aspects of the cluster20bindicated by the characterization information, reaching a completion threshold. The advance module106may further (7) initiate, or act to initiate, processing the data-processing job76e, or pool of data-processing jobs76, by the cluster20b.

In some examples, the running of one or more data-processing jobs76may be broken into various stages such that after a first warm-up and run for a map phase30, another warm-up and run phase may begin. To illustrate this concept, certain nodes42/40at which reducers52may be implemented are depicted (8) providing a status update with respect to the provisioning and/or loading of the requisite intermediate records48for a given data-processing job76e. The running of data-processing jobs76may be optimized accordingly.

As can be appreciated, the innovations discussed with respect to the previous figures, and to be discussed with respect to the figures that follow, may be implemented on a virtual cluster56. Such a virtual cluster56may include data nodes16configured on SSDs66. As discussed with respect toFIG. 3, such a virtual cluster56may use a file system with location awareness and/or with redundancies that addresses data retention issues typical of flash memory.

Referring toFIG. 8, methods110are depicted for warming-up a portion of a cluster20bof nodes in anticipation of running a data-processing job76, or jobs76. The flowchart inFIG. 8illustrates the architecture, functionality, and/or operation of possible implementations of systems, methods, and computer program products according to certain embodiments of the present invention. In this regard, each block in the flowcharts may represent a module, segment, or portion of code, which comprises one or more executable instructions for implementing the specified logical function(s). It will also be noted that each block of the flowchart illustrations, and combinations of blocks in the flowchart illustrations, may be implemented by special purpose hardware-based systems that perform the specified functions or acts, or combinations of special purpose hardware and computer instructions.

It should also be noted that, in some alternative implementations, the functions noted in the blocks may occur out of the order noted in the figure. In certain embodiments, two blocks shown in succession may, in fact, be executed substantially concurrently, or the blocks may sometimes be executed in the reverse order, depending upon the functionality involved. Alternatively, certain steps or functions may be omitted if not needed.

The methods110may utilize characterization information about a data-processing job76, or jobs76, to warm-up the appropriate portion of a cluster20/56. The method110may begin112by114reading one or more data-processing jobs76in a job queue74. The data-processing job76may be configured for distributed processing of a set of data blocks/replicas24/26on a cluster20/56of nodes42/40. The set of data blocks/replicas24/26may be stored at multiple locations either in the cluster20/56and/or a backend.

The methods110may continue by116extracting characterization information from the data-processing job76, or jobs76. The characterization information may have various types of information characterizing the data-processing job76, or jobs76, and/or data and/or cluster resources involved in running the76, or jobs76. In some examples, the methods110may proceed by120warming up aspects of the cluster20/56indicated by the characterization information as aspects to be utilized in running the data-processing job76, or jobs76.

In other examples, the warm-up step120may be proceeded by118analyzing the characterization information for identification information. The identification information may identify, without limitation, a data block/replica24/26in the set to be processed during the data-processing job76, jobs76. The identification information may include one or more locations at which a data block/replica24/26in the set is stored, one or more nodes40/42in the cluster20/56serviceable for processing a data block/replica24/26in the set, and/or network resources serviceable to provision a data block/replica24/26in the set to a node42/40in the cluster20/56for processing.

In some examples,118analyzing the characterization information for identification information may further involve converting a set of mappers44and/or a set of reducers52extracted with the characterization information into a set of compute node identifiers92serviceable for processing the data-processing job76, or jobs76. The analysis step118may further involve converting set of mappers44and/or a set of reducers52into a group of data units indicating data blocks/replicas24/26on which the data-processing job76, or jobs76runs. Additionally, the step118may involve determining one or more of the multiple locations at which the set of data blocks/replicas24/26are stored, whether inside the cluster20/56or outside the cluster20/56. These locations may be obtained, in some examples, from locational awareness information stored in a name node18.

The methods110may wait to proceed until a determination122that the relevant aspects of the cluster are warmed up, have warmed up to a predetermined threshold, or a sufficient portion of the aspects of the cluster are warmed up. When an affirmative warm-up determination122has been reached, the methods110may proceed with running the data-processing job76, or jobs76, after at least a portion of the aspects of the cluster20/56are warmed up. Running the data-processing job76, or jobs76, may involve124getting the relevant data-processing job76, or jobs76, from the queue74and initiating126processing of the data-processing job76, or jobs76. At this point in many examples, the method110may end.

However, in additional examples, an additional determination128may be made as to whether additional aspects of the cluster20/56warmed up previously for the data-processing job76, or jobs76, may be utilized for a second data-processing job76, or jobs76. If they are, the second data-processing job76, or jobs76, which utilize these aspects, may be selected from the job queue74to be run. In other examples involving the running of multiple data-processing jobs concurrently, the methods110may involve reading multiple data-processing jobs74in the job queue74.

Such methods110may further involve extracting characterization information for each of the multiple data-processing jobs76. This characterization information may be used for identifying overlapping resources in the cluster20/56to be used by more than one data-processing job76. Based on these overlapping resources, a pool of data-processing jobs may be selected to be run concurrently in the cluster20/56that may be used to process the pool of data-processing jobs76. The overlapping resources may be warmed-up, at which point, processing the pool of data-processing jobs76may begin.