Graph-processing techniques for a MapReduce engine

Systems, methods, and devices for sorting and processing various types of graph data are described herein. Partitioning graph data into master data and associated slave data allows for sorting of the graph data by sorting the master data. In another embodiment, promoting a data bucket having a first data bucket size to a data bucket having a second data bucket size greater than the first data bucket size upon reaching a memory limit allows for the reduction of temporary files output by the data bucket.

BACKGROUND

Computer users, both in the business and home settings, utilize many different types of computing devices to send electronic mail, generate work product, and process data. A plurality of computing devices, networks, and/or systems may be needed to process large amounts of data related to these and other computer uses. Programming models, such as MapReduce, may be used to process and generate these large amounts of data.

Utilizing a MapReduce process, a user may cause a map function to process and partition data, such as a key/value pair, to generate a set of partitioned data. The partitioned data may then be distributed to one or more nodes so that the partitioned data may be processed separately and independently. Subsequently, a reduce function may serve to merge the processed and partitioned data in order to output a merged data record.

There are at least two general assumptions when large data sets are processed utilizing a MapReduce process. First, the amount of data distributed to each of the one or more nodes will be approximately the same. Furthermore, the data input into the MapReduce process will be in key-value format, meaning that such data will take up about the same size in the memory of a particular computing device. However, when the data input into the MapReduce process is in a graphical form, the foregoing assumptions may no longer be valid.

SUMMARY

Described herein are techniques for the processing and/or sorting of interrelated data, such as graphical data. In one embodiment, interrelated data items partitioned from input data are distributed into one of a plurality of data buckets each having a master container and a slave container. The partitioned interrelated data items in each data bucket are further divided into master data items to be held in the master container and corresponding slave data items to be held in the slave container. Each master data item may be associated with one or more slave data items. The master data items in each data bucket may then be processed and/or sorted so that the master data items and their corresponding slave data items are output as a temporary file. Subsequently, the temporary files generated by each of the data buckets may be merged together and output as a single data record.

In another embodiment, input data may be partitioned and distributed into a plurality of data buckets. Each of the data buckets may initially be designated as having a first data bucket size. As each of the data buckets reaches a maximum storage capacity, the data buckets may output the data as one or more temporary files. Further, if a particular data bucket reaches its maximum storage capacity a number of times, that data bucket may be promoted to a data bucket having a second data bucket size that is greater than the first data bucket size. As a result, the data bucket having a second, greater data bucket size may output a larger, reduced amount of temporary files.

In at least one embodiment, one or more computing devices may partition graph data to preserve a locality property. For instance, graph data associated with other graph data may be maintained in the same partition such that the data will be processed together.

DETAILED DESCRIPTION

Described herein are techniques for processing and sorting interrelated data, such as graph data. Each record in the graph data may have a structure containing a plurality of source nodes and destination nodes. Although the source nodes of a particular graph record may be of a fixed size, the number of destination nodes for each graph record may be large. Accordingly, when processing one or more graph records, it may be difficult to sort the data due to the number of source nodes and associated destination nodes. For instance, a single graph record may contain five source nodes and each source node may also have a plurality of associated destination nodes that are either different or the same from destination nodes associated with the other source nodes. Therefore, to sort the foregoing graph record, the graph record may be divided into master data items corresponding to the source nodes and slave data items corresponding to the destination nodes. As the associated destination nodes are known for each source node, the entire graph record may be sorted by sorting the master data items without sorting the associated slave data items.

Also described herein are techniques for managing the processing of graph data in one or more data buckets. As data is distributed into one or more data buckets and a particular data bucket reaches its memory capacity, the data bucket may output or distribute the data to memory. Thus, the memory capacity of the data bucket may determine the amount of data output to memory. As mentioned previously, due to the variance in size of partitioned graph data, certain data buckets may reach maximum memory capacity before other data buckets. Therefore, assuming that the data buckets each have the same data bucket size, it follows that data buckets that receive greater amounts of graph data may likely generate and output a greater number of temporary files. To reduce the number of temporary files that may later be merged with temporary files output by other data buckets, a particular data bucket having a first size that reaches its memory capacity a predetermined number of times may be promoted to a data bucket having a second, greater size. Thus, since the size of the data bucket has increased, that particular data bucket may generate and output larger, but fewer, temporary files.

Various examples for sorting and processing data in accordance with the embodiments are described below with reference toFIGS. 1-7.

FIG. 1illustrates a diagram for the sorting and processing of data. In various embodiments, the processing begins with data input102, which may be any type of data, including interrelated data such as graph data. Subsequently, data input102may be divided or partitioned into extracted data104.FIG. 1illustrates that extracted data items104is shown as two different data records. However, any number of extracted data items104is contemplated. In an embodiment in which input data102comprises graph data, the graph data may be divided or partitioned into extracted data104by a MapReduce engine, which may comprise one or more graph records. Each graph record may be comprised of one or more source nodes and a plurality of destination nodes, with each source node being associated with one or more destination nodes of the graph record. As the source nodes and destination nodes are associated with one another, this data may be referred to as interrelated data items. Accordingly, each source node may have one or multiple associated destination nodes and two different source nodes may have the same or different destination nodes. The source nodes in a particular graph record may be fixed. On the other hand, the number of destination nodes for different graph records may vary greatly. For instance, a source node in data record A may have 3 associated destination nodes whereas a source node in data record B may have 15 associated destination nodes. Therefore, the size of each graph record may vary.

In various embodiments, the interrelated data items, represented by extracted data104, are directed into mapper106. Mapper106may be embodied in hardware, software, or a combination thereof. In the above embodiment, mapper106may include bucket judgment108and a plurality of data buckets110. As shown, data bucket110A is an example data bucket of data buckets110. Each of data buckets110, including data bucket110A, may include a master container112and a slave container114.

In the above embodiment, once extracted data104is directed into mapper106, bucket judgment may distribute each set of interrelated data items into one of the plurality of data buckets110, such as data bucket110A. Subsequently, the interrelated data items in each of data buckets110may be divided or partitioned into master data items116and slave data items118. The interrelated data items may be partitioned by a MapReduce engine and the master data items116and the associated slave data items118may be comprised of master data items116and associated slave data items118, respectively. In data bucket110A, the master data items116are held in master container112and the slave data items118associated with the master data items116are held in slave container114. In various embodiments, each master data item116may have unidirectional control over its associated slave data items118and/or may be aware of the status of each slave data item118. As shown in data bucket110A, the lines extending from each master data item116in master container112to a subset of slave data items118in slave container114represent specific slave data items118associated with a particular master data item116. Moreover, as mentioned previously, the master data items116and the associated slave data items118may comprise source nodes and associated destination nodes, respectively.

As stated above, the foregoing embodiment describes that the master container112may hold the master data items116and the slave container114may hold the corresponding slave data items118. Furthermore, in each of data buckets110, master container116includes one or more records. In the above embodiment, each record may serve as a wrapper for a particular master data item116whereby the wrapper may include information relating to that particular master data item116. Such information may include an index and length of slave data items118in the slave container114associated with that particular master data item116. That is, the wrapper describes information about the one or more slave data items118associated with a particular master data item116.

After the master data items116and corresponding slave data items118are distributed into master container112and slave container114, respectively, mapper106may proceed to sort and process the interrelated data items contained in each of data buckets110. In one embodiment, both the master data items116and the corresponding slave data items118may be sorted together. However, in another embodiment, the master data items116in data buckets110may be sorted without processing or sorting the slave data items118corresponding to the master data items116. Since each master data item116may be aware of its corresponding slave data items118, it may not be necessary to sort the slave data items118in addition to the master data item116. In other words, by sorting each master data item116in the master container112in data buckets110and110A, mapper106is also sorting the associated slave data items118in slave container114.

Furthermore, in an example embodiment, data input102may be any type of graph data and extracted data104may represent one or more graph records partitioned from the graph data. As stated above, the source nodes held in master container112may be sorted without sorting the associated destination nodes held in slave container114. Therefore, in this embodiment, mapper106need not sort or process the slave data items118, or destination nodes, provided that the master data items116, or source nodes, of the graph record are sorted. In other words, the master-slave relationship between the source nodes and the associated destination nodes may allow for the sorting of just a portion of the graph record.

In some embodiments, after sorting master data items116, each data bucket110may merge its master data items116and associated slave data items118into a single record. Subsequently, the single record for each of the data buckets110may be output as bucket output120such that each data bucket110has output a final bucket output126. For instance, with respect to data bucket110A, the bucket output120for data bucket110A may include the sorted master data items116and its associated slave data items118.

In one embodiment, each bucket output120may be sorted by source node identifiers. In this embodiment, each source node may be associated with and recognizable by a unique identifier. Accordingly, the final bucket output126for each data bucket110may be sorted such that each source node is followed by its associated destination nodes. That is, for example, if a particular data bucket contained multiple source nodes, the bucket output120may be sorted and ordered as follows: (1) source node having a first source node identifier, (2) destination nodes associated with first source node, (3) source node having a second source node identifier, (4) destination nodes associated with second source node, and so on.

In an exemplary embodiment, bucket output120for each of the data buckets110may then be input into reducer122. Subsequently, the bucket outputs120for each of the data buckets110may be distributed into temporary data bucket124. In this embodiment, temporary bucket124may sort each of the bucket outputs120and subsequently generate and output a final output126. Final output126may represent data input102after it has been both sorted and processed.

In one embodiment, reducer122may reduce bucket outputs120into final output126utilizing the master-slave concept described above. More particularly, similarly to data bucket110A, temporary data bucket124may include both a master container and a slave container. Further, once the bucket output120for each data bucket110has been directed to temporary data bucket124, each bucket output120may be divided into master data items116and slave data items118, with each slave data item118being associated with one or more master data items116. As stated previously, provided that the bucket output120relates to a graph record, the source nodes and the associated destination nodes may be defined as the master data items116and the associated slave data items118, respectively. Moreover, the master container, which may hold the master data items116, may also include one or more records that represent a wrapper for a particular master data item116.

In the foregoing embodiment, since the master data items116in the temporary data bucket124may have control over, or at least be aware of, its associated slave data items118held in the slave container of temporary data bucket124, the reducer122may sort the master data items116without sorting the associated slave data items118. The temporary data bucket124may then merge the master data items116and the associated slave data items118from each of the bucket outputs120in order to generate a single record. The single record may be sorted in any manner, such as by sorting the record by a source node identifier, as previously described. Further, once the bucket output120for each of the data buckets110is sorted as a single record, reducer122may output the sorted record as final output126. As set forth herein, final output126may comprise data input102, such as graph data, sorted and reduced by mapper106and reducer122, respectively.

FIG. 2illustrates a diagram for the sorting and processing of data utilizing one or more data buckets. In one embodiment, data input202is partitioned into one or more extracted data204. Data input202may be any type of data, such as graph data. In an embodiment in which data input202is graph data, the graph data may be partitioned into one or more graph records. As stated above with respect toFIG. 1, each graph record may have one or more source nodes and a plurality of destination nodes associated with a particular source node.

In various embodiments, extracted data204is directed into mapper206. Mapper206may be embodied in hardware, software, or a combination thereof. In the above embodiment, mapper206may include bucket judgment208and a plurality of data buckets210. As shown, data buckets210A-D may serve as example data buckets of data buckets210. As mentioned previously in relation toFIG. 1, each of data buckets210, including data buckets210A-D, may include a master container and a slave container. In response to extracted data204being input into mapper206, bucket judgment208may direct or distribute the extracted data204into one of data buckets210. That is, in this embodiment, extracted data204may be distributed into one of data buckets210A-D.

In the above embodiment, extracted data204may be divided into master data items116and slave data items118associated with a particular master data item116once the extracted data is distributed into one of data buckets210. For any particular one of data buckets210, the master data items116may be held in the master container and the corresponding slave data items118may be held in the slave container. Furthermore, as described with respect toFIG. 1, as each data bucket210is sorted and/or processed by mapper206, the master data items116may be sorted without sorting the associated slave data items118. In some embodiments, as each master data item116has control over, or at least is aware of, its associated slave data items118, the interrelated data items in a particular data bucket210may be sorted by sorting just the master data items116. Sorting the master data items116without sorting the corresponding slave data items118may cause the sorting of a data bucket210to be more efficient. Subsequently, the master data items116and corresponding slave data items118may be merged into a single record and output by one of data buckets210.

In an example embodiment, as additional extracted data204is distributed into data buckets210, each of data buckets210may reach a memory limit or capacity. The memory limit for each data bucket210may be based on the total number of data buckets and/or the total memory of the machine or computing device in which the data buckets are embodied. In this embodiment, when each of data buckets210A-D reaches their respective memory limits, data buckets210A-D may output the data as a temporary file212. However, before outputting the data as a temporary file212, data buckets210A-D may sort the master data items116and the associated slave data items118utilizing the process described above. Upon outputting the master data items116and the associated slave data items118as a temporary file212, each of data buckets210A-D may receive additional extracted data204from bucket judgment208. Data buckets210A-D may then repeat the above process and output additional data as a second temporary file upon reaching a memory capacity. The above process may continue until no more data is being distributed into data buckets210A-D. Therefore, the memory capacity for a particular data bucket210determines the number of temporary files212output by that data bucket210.

That is, in the above embodiment, extracted data204may be distributed to one of data buckets210, divided into master data items116and associated slave data items118, the master data items116are sorted by one of data buckets210when the data bucket210reaches a maximum memory capacity, the master data items116and associated slave data items118are merged into a single record, and the single record is output as a temporary file212.

In various embodiments, provided that one of data buckets210A-D has output more than one temporary file212, the temporary files212may be merged into a single file representing a final output for one of data buckets210A-D. Further, as described with respect toFIG. 1, the single records output for each data bucket210A-D may be sorted and merged in order to generate a final output for each of data buckets210. The number of data buckets shown inFIG. 2(data buckets210A-D) is for illustration purposes only and it is contemplated that more or less data buckets210may be present.

As stated previously, in various embodiments, data input202may be a particular type of graph data. The graph data may be divided into one or more graph records having source nodes and associated destination nodes. However, for large graph data sets, the graph data may be partitioned with high locality property. That is, for each partition, the source nodes and the associated destination nodes of the graph data may be highly overlapped. As a result, a particular data bucket, such as data bucket210A, may receive more and/or larger graph records than other data buckets, such as data buckets210B-C. In other words, there may be an unbalanced amount of graph data distributed into different data buckets210A-D. In this case, assuming that data buckets210A-D each had the same memory capacity, data bucket210A would likely reach its memory capacity prior to data buckets210B-D and, therefore, generate a greater number of temporary files212than data buckets210B-C. Furthermore, in large scale data processing, generating, processing, and/or sorting a great number of files may cause significant system performance inefficiencies, such as being unable to open a large number of files at one time.

To reduce the number of temporary files212generated by a data bucket210that receives larger and/or more graph data, the memory capacity of that data bucket210may be adjusted. In one embodiment, each of data buckets210may have the same storage capacity and may be designated as having a first data bucket size. There may also be a category of data buckets that have a larger storage capacity, which may be designated as data buckets having a second data bucket size that is greater than the first data bucket size. In some embodiments, the number of data buckets with a larger storage capacity may be limited based on the number of data buckets already having the second data bucket size and/or a total amount of available memory of the system. For example, there may be a quota representing a total number of data buckets having a second data bucket size that the system can accommodate. It is also contemplated that additional categories of data buckets may exist, such as data buckets having a third size, data buckets having a fourth size, and so forth.

In various embodiments, a computing device may track the number of times a particular data bucket210having a first data bucket size reaches its memory capacity. In one embodiment, the above tracking may be performed by a MapReduce engine of the computing device. As stated above, once a data bucket210reaches its memory capacity, the data bucket210may output a temporary file212. Upon the data bucket210reaching its memory capacity a predetermined number of times, that data bucket210may be promoted to a data bucket having a second data bucket size that is greater than the first data bucket size. In some instances, the data bucket210having a first data bucket size may be promoted into a data bucket having a second data bucket size if the memory capacity of the data bucket210is reached a predetermined number of times in a predefined time period. In above embodiment, the data bucket having a second data bucket size has a greater amount of memory than the data bucket210having a first data bucket size.

In various embodiments, promoting a data bucket210having a first data bucket size to a data bucket having a second data bucket size may be performed automatically as the data bucket210having a first data bucket size reaches its memory capacity a predetermined number of times. Furthermore, the promotion may also be based on whether a quota of data buckets having a second data bucket size has been reached. If not, the data bucket210may be promoted to a data bucket having a second data bucket size. However, if the above quota has been reached, there may not be enough memory to promote the data bucket210to a data bucket having a second data bucket size. In this case, the data bucket210will remain as a data bucket having the first data bucket size. As a result of the promotion, the data bucket210will have a larger memory capacity and may therefore generate a reduced amount of temporary files122.

FIG. 2further illustrates an example of the process described immediately above. In an example embodiment, data buckets210include at least data buckets210A-D. As shown, data buckets210A-210D are approximately the same size and, therefore, may be designated as data buckets having a first data bucket size. In other words, data buckets210A-D each has approximately the same amount of memory. In this embodiment, as data is distributed into each of data buckets210A-D, data buckets210A-210D generate one or more temporary files212in the manner previously described. Initially,FIG. 2illustrates that data bucket210A has output three temporary files212, data bucket210B has output two temporary files212, and each of data bucket210C and data bucket210D has output one temporary file212. That is, data bucket210A has reached its memory capacity three times, data bucket210B has reached its memory capacity two times, and data buckets210C and210D have each reached their respective memory capacities one time.

In the above embodiment, the system will promote a data bucket210having a first data bucket size to a data bucket having a second data bucket size if a particular data bucket210reaches its memory capacity three times. However, in other embodiments, it is contemplated that such a promotion may also occur if the number of times of reaching memory capacity is either less than or greater than three times. As shown, since data bucket210A has output three temporary files212, it is detected that data bucket210A has reached its memory capacity three times. Therefore, data bucket210A is promoted to a data bucket having a second data bucket size, which is greater than data buckets210B-C, which each have a first data bucket size.

Data bucket210A may also be promoted to a data bucket of a second data bucket size under limited circumstances. Such circumstances include promoting data bucket210A when it reaches its memory capacity a predetermined number of times in a predefined time period and/or when there is available memory for such a promotion. In addition, the promotion of data bucket210A from a first data bucket size to a second data bucket size may be performed either automatically or manually upon reaching its memory capacity a predetermined number of times. Furthermore, as shown inFIG. 2, data buckets210B-D will remain as data buckets having a first data bucket size as they each have not reached a memory capacity three different times.

FIG. 2further illustrates data buckets210A-D in which data bucket210A has a second data bucket size greater than the first data bucket size. On the contrary, data buckets210B-D each have a first data bucket size. In this embodiment, since data bucket210A has been promoted to a data bucket having a second data bucket size, it now may have a greater memory capacity than data buckets210B-D. Accordingly, data bucket210A may generate temporary files212that are larger than temporary files output by data buckets210B-D. Further, data bucket210A may generate fewer temporary files212than when data bucket210A had a first data bucket size. Therefore, once a data bucket210has been promoted to a second, larger data bucket size, the amount of output temporary files212may decrease. Consequently, fewer temporary files212will need to be merged in a subsequent sorting.

FIG. 3illustrates a process for sorting and processing graph data. Initially, graph data may be partitioned into multiple graph partitions, such as graph partition304and graph partition316. For the purposes of this embodiment, graph partition304and graph partition316may be any type of graph data. Moreover graph partition304and graph partition316may each be comprised of source nodes and one or more destination nodes each being associated with a particular source node. Further, the one or more destination nodes may also be associated with more than one source node.

In an example embodiment, with respect to computer device302, mapper306may divide graph partition304into a plurality of records308. Each record308, such as records308A,308B,308C, and308N, relates to destination nodes for a different graph partition. For instance, record308A refers to destination nodes corresponding to one or more source nodes in graph partition304, record308B refers to destination nodes corresponding to one or more source nodes in graph partition316, and records308C and308N correspond to one or more source nodes for graph partitions not displayed inFIG. 3. Record308N is shown to indicate that any number of records and corresponding graph partitions may be processed by mapper306.

Similarly, with respect to computing device314, mapper318may divide graph partition316into a plurality of records320. Records320A,320B,320C, and320N each relate to destination nodes for a different graph partition. As with records308N, record320N illustrates that any number of records may be included in computing device314. In this embodiment, record320A refers to destination nodes corresponding to the one or more source nodes in graph partition304, record320B refers to destination nodes corresponding to the one or more source nodes in graph partition316, and records320C and320N correspond to the one or more source nodes for graph partitions not shown inFIG. 3.

Subsequently, records having destination nodes corresponding to a source node for a particular graph partition are distributed to the same reducer. In other words, destination nodes associated with the same source nodes may be merged together into a resulting data record. For instance, as stated above, records308A and320A both contain destination nodes relating to one or more source nodes corresponding to graph partition304. Therefore, records308A and320A may be distributed to the same reducer, such as reducer310. In response, reducer310may merge records308A and320A to generate a single file represented by result312. Result312will contain source nodes and associated destination nodes for graph partition304. Likewise, records308B and320B both contain destination nodes relating to one or more source node corresponding to graph partition316. Accordingly, records308B and320B may be distributed to a different reducer, such as reducer322. Reducer322may then merge records308B and320B to generate a single file represented by result324. Result324will contain one or more source nodes and associated destination nodes for graph partition316. Furthermore, a process similar to the above may also be conducted for records308C and320C, and so forth. Therefore, each result, such as results312and324, contains source nodes from a graph partition and its associated destination nodes.

FIGS. 4-6describe various example processes for processing and/or sorting various types of data. The example processes are described in the context of the environment ofFIGS. 1-3, but are not limited to those environments. The order in which the operations are described in each example process is not intended to be construed as a limitation, and any number of the described blocks can be combined in any order and/or in parallel to implement each process. Moreover, the blocks in theFIGS. 4-6may be operations that can be implemented in hardware, software, and a combination thereof. In the context of software, the blocks represent computer-executable instructions that, when executed by one or more processors, cause one or more processors to perform the recited operations. Generally, the computer-executable instructions may include routines, programs, objects, components, data structures, and the like that cause the particular functions to be performed or particular abstract data types to be implemented.

FIG. 4is flowchart illustrating a method of sorting and/or processing a plurality of interrelated data items. More particularly, block402illustrates directing a plurality of interrelated data items into one of a plurality of data buckets. The plurality of interrelated data items may comprise data input102and/or extracted data104. Moreover, the interrelated data items may be any type of data including, for example, graph data. The interrelated items may also be distributed to one of data buckets110by mapper106and/or bucket judgment108. Further, each of data buckets110may include both a master container112to hold master data items116and a slave container to hold slave data items118associated with one or more master data items116.

Block404illustrates partitioning the plurality of interrelated data items into master data items and slave data items. Moreover, the above master data items and the slave data items may comprise master data items116and associated slave data items118, respectively. Provided that the interrelated data items comprise graph data, the master data items116may correspond to source nodes and the associated slave data items118may correspond to destination nodes associated with the source nodes.

Block406illustrates distributing the master data items into a master container of the data bucket and the slave data items into a slave container in the data bucket. In particular, once distributed into one of data buckets110, master data items116may be distributed into master container112and the associated slave data items118may be distributed into slave container114. In one embodiment, the distributing may be performed by a MapReduce engine of a computing device.

Block408illustrates sorting the interrelated data items by sorting the master data items. More particularly, as a particular master data item116may either have control over or at least be aware of its associated slave data items118, the master data items116held in the master container112of each data bucket110may be sorted without sorting the slave data items118associated with that master data item116. Sorting the interrelated data items by sorting just the master data items116may decrease the time required for sorting and, therefore, increase efficiency.

Block410illustrates outputting the master data items and associated slave data items for a particular one of the data buckets as a temporary file. Moreover, the master data items116and associated slave data items118for a particular one of data buckets110may be output as a temporary file, such as bucket output120, in response to the data bucket110reaching a maximum storage capacity. That is, as each data bucket110receives data and reaches a maximum storage capacity, the sorted data, which includes the master data items116and the associated slave data items118, may be output as a temporary file for that data bucket110. Each data bucket110may sort and output one or more temporary files based on the number of times each data bucket110reaches a maximum storage capacity.

Block412illustrates reducing the master data items and the associated slave data items from each of the data buckets into a single file for output. As described above, the temporary files may be referred to as bucket output120. Further, once each data bucket110has sorted and output its stored data as a temporary file, the temporary files may be distributed to reducer122, which contains temporary data bucket124. The temporary files, or bucket output120, may then be distributed into temporary data bucket124.

In various embodiments, Temporary data bucket124may sort the master data items116and the associated slave data items118by sorting the master data items116without sorting the associated slave data items118. Following such sorting, the temporary data bucket124may merge the temporary files output by each data bucket110into a single file for output. The single file may be referred to as final output126.

FIG. 5is flowchart illustrating tracking a number of times that a data bucket having a first data bucket size reaches a memory capacity.FIG. 5further illustrates promoting the data bucket to a second data bucket size that is greater than the first data bucket size based on the number of times the data bucket reaches its memory capacity.

Block502illustrates tracking a number of times that a data bucket having a first data bucket size reaches a memory capacity. In one embodiment the tracking may be performed by a MapReduce engine of a computing device. More particularly, data bucket may be any one of data buckets210A-210D. In one embodiment, each of data buckets210A-210D may have a first data bucket size and a memory limit. The memory limit may be based on the total memory available for a particular computing device. Moreover, as each of data buckets210A-D may output a temporary file212when its maximum memory capacity is reached, tracking the number of times one of data buckets210A-D reaches a memory capacity may be determined by tracking the number of temporary files212output for each data bucket210.

Block504illustrates outputting a temporary file when the data bucket reaches the memory capacity. As stated above, data buckets210A-D may output one or more temporary files212when any one of data buckets210A-D reaches its maximum memory capacity.

Block506illustrates promoting the data bucket to a second data bucket size that is greater than the first data bucket size based on the number of times. For example, data bucket210A having a first data bucket size may be promoted to a data bucket having a second data bucket size that is greater than the first data bucket size. The promoting may be based on a predetermined number of times any one of data buckets210A-D reaches its memory capacity. Further, in order for a particular data bucket210to be promoted to a greater data bucket size, the data bucket210may need to reach its memory capacity a predetermined number of times within a predefined time period. The promoting set forth in block506may also be based on a quota defined as a number of data buckets having a second data bucket size that are available or be based on the total memory available for a particular system or computing device. Promoting a data bucket210to a data bucket having a greater data bucket size may also reduce the number of temporary files212output by that data bucket.

Block508illustrates sorting data in the data bucket prior to outputting the temporary file. As mentioned previously with respect toFIGS. 1,2, and4, interrelated data items, such as master data items116and associated slave data items118, may be sorted prior to outputting the master data items116and the associated slave data items118to a temporary file212.

Block510illustrates merging the temporary file with a plurality of additional temporary files output by one or more data buckets. As stated above, as each of data buckets210A-D reaches a maximum storage capacity, each data bucket may output a temporary file212. Moreover, these temporary files212may be merged together. For instance, the temporary file(s)212output by data bucket210A may be merged with the temporary files212output by data buckets210B-D.

Block512illustrates generating a single file including the temporary file and the plurality of additional temporary files for output to a storage device. As mentioned above with respect to block510, temporary files212output by each of data buckets210may be merged together. A single file or record may then be generated that includes the temporary files212from each data bucket210. For example, provided that the temporary files212output by data buckets210A-D were previously merged, the temporary files212for data buckets210A-D may be output as a single file to a storage device. Such a file may comprise the master data items116and the associated slave data items118from each temporary file212output by data buckets210A-D. Further, the storage device may be any type of device having memory to store the single file.

FIG. 6is flowchart illustrating sorting graph data having source nodes and associated destination nodes. Block602illustrates partitioning a plurality of graph records into source nodes and associated destination nodes. In one embodiment, a particular source node is associated with one or more destination nodes.

Block604illustrates distributing the source nodes into a master container of a data bucket and the destination nodes into a slave container of the data bucket. In one embodiment, the distributing may be performed by a MapReduce engine of a computing device. More particularly, the source nodes may be distributed into master container112in a data bucket110and the associated destination nodes may be distributed into slave container114of the same data bucket110. In one embodiment, each source node held in the master container112is associated with one or more associated destination nodes held in the slave container114.

Block606illustrates sorting the graph records by sorting the source nodes without sorting the associated destination nodes. In various embodiments, as a particular source node either has control over, or at least is aware of, the destination nodes associated with that source node, both the source nodes and the associated nodes may be sorted by sorting just the source nodes. As shown, sorting of the source nodes of data bucket110may occur in the master container112.

Further, in response to sorting the source nodes, one of data bucket110may output the source nodes and the associated destination nodes as one or more temporary files212in response to reaching a memory capacity of the data bucket110. For example, once data bucket210A reaches its memory capacity, data bucket210A may output the source nodes and associated destination nodes as a temporary file212. Moreover, data bucket210A may continue to output one or more temporary files212if data bucket210A reaches its maximum memory capacity one or more times. Subsequently, provided that a particular data bucket210, such as data bucket210A, output more than one temporary file212, the one or more temporary files212may be merged to generate a single record for output by the data bucket210to a storage device. For instance, as shown inFIG. 2, data bucket210A has output three temporary files212. Therefore, the three temporary files212may be merged together to generate a single record for data bucket210A and the single record may be output to a storage device.

Block608illustrates reducing the source nodes and the associated destination nodes from each of the data buckets into a single file for output. More particularly, the source nodes and associated destination nodes included in temporary files212output by data buckets210A-D may be reduced to a single file for output. Accordingly, the single file may include the source nodes and associated destination nodes for each of the temporary files212output by data buckets210. Moreover, as stated above, the single file for output may be sorted based on source node identifiers.

Block610illustrates reducing the single record for each of the data buckets into a single record for output. With respect toFIG. 1, the files output by each of data buckets110may be reduced to a single record that comprises the source nodes and the associated destination nodes for data buckets110. Moreover, the single record for output may be referred to as bucket output120and the reducing may be performed by reducer122.

In various embodiments, the reducing may comprise partitioning the single record for each of data buckets110into source nodes and destination nodes. As mentioned previously, each destination node may be associated with at least one source node. Subsequently, the source nodes may be distributed into a master container of a data bucket, such as temporary data bucket124, and the destination nodes may be distributed into a slave container of temporary data bucket124. Therefore, in the above embodiment, the master container of temporary data bucket124may hold the source nodes output by data buckets110and the slave container of temporary data bucket124may hold the destination nodes associated with the source nodes output by data buckets110.

In addition, the single record may be sorted by sorting the source nodes held in master container of temporary data bucket124. In one embodiment, temporary data bucket124may sort the single record by sorting the source nodes held in the master container without independently sorting the destination nodes held in the slave container. In this embodiment, the temporary data bucket124may sort the single record more efficiently by sorting just the source nodes since it may take additional time to also independently sort the associated destination nodes. Upon the single record being sorted, the temporary data bucket124may reduce the source nodes and the associated destination nodes held in the master container and the slave container, respectively, as a final output126for output. In this embodiment, final output126comprises the sorted source nodes and associated destination nodes from each of data buckets110.

Block612illustrates tracking a number of times the data bucket reaches a memory capacity. In one embodiment, as a data bucket210will output a temporary file212upon its memory limit being reached, the number of times a data bucket210reaches its memory capacity may be determined by tracking the number of temporary files212output by a particular data bucket210. For instance, one could determine that data bucket210A has reached its memory capacity three different instances since data bucket210A has output three temporary files212.

Block614illustrates promoting the data bucket to a data bucket having a greater data bucket size based on the number of times in a predefined time period. Initially, each data bucket210may have a first data bucket size. However, if one of data buckets210reaches a maximum memory limit a predetermined number of times in a predefined period of time, that data bucket may be promoted to a data bucket having a second data bucket size that is greater than the first data bucket size. An increase in data bucket size may cause that particular data bucket to output larger, but a reduced number, of temporary files212.

In an example embodiment, data bucket210A may be promoted from a data bucket having a first data bucket size to a data bucket having a second data bucket size that is greater than the first data bucket size. In this embodiment, the predetermined number of times is three. As shown, data bucket210A has output three temporary files212, which indicates that data bucket210A has reached its memory capacity three different times. Therefore, after being promoted to a data bucket having a second size, data bucket210outputs larger but possibly fewer temporary files212. Since data buckets210B-D have each output less than three temporary files212, data buckets210B-D have not reached their respective memory limits the predetermined number of times. Accordingly, data buckets210B-D will continue to have a first data bucket size.

FIG. 7illustrates a representative computing device700that may implement the sorting of various types of data, such as graph data, in accordance with various embodiments. For example, the computing device700may be implemented as any of the computing devices described previously. However, it will be readily appreciated that the techniques and mechanisms may be implemented in other computing devices, systems, and environments. The computing device700shown inFIG. 7is only one example of a computing device and is not intended to suggest any limitation as to the scope of use or functionality of any computing device utilized to perform the processes and/or procedures described above.

In at least one configuration, computing device700includes at least one processing unit702and system memory704. Processing unit702may execute one or more modules and/or processes to cause computing device700to perform a variety of functions. In some embodiments, the processing unit702is a central processing unit (CPU), a graphics processing unit (GPU), both CPU and GPU, or other processing units or components known in the art.

Depending on the exact configuration and type of computing device, system memory704may be volatile (such as RAM), non-volatile (such as ROM, flash memory, miniature hard drive, memory card, or the like) or some combination thereof. System memory704may include an operating system706, one or more program modules708, and may include program data710. The computing device700is of a very basic configuration demarcated by a dashed line712. Again, a terminal may have fewer components but may interact with a computing device that may have such a basic configuration.

Computing device700may have additional features and/or functionality. For example, computing device700may also include additional data storage devices (removable and/or non-removable) such as, for example, magnetic disks, optical disks, or tape. Such additional storage is illustrated inFIG. 7by removable storage714and non-removable storage716. Computer storage media may include volatile and nonvolatile, removable, and non-removable media implemented in any method or technology for storage of information, such as computer readable instructions, data structures, program modules, or other data. System memory704, removable storage714and non-removable storage716are all examples of computer storage media. Computer storage media includes, but is not limited to, RAM, ROM, EEPROM, flash memory or other memory technology, CD-ROM, digital versatile disks (DVD), or other optical storage, magnetic cassettes, magnetic tape, magnetic disk storage or other magnetic storage devices, or any other medium which can be used to store the desired information and which can be accessed by computing device700. Any such computer storage media may be part of computing device700. Computing device700may also have input device(s)718such as a keyboard, a mouse, a pen, a voice input device, a touch input device, etc. Output device(s)720, such as a display, speakers, a printer, etc. may also be included.

Computing device700may also contain communication connections722that allow the device to communicate with other computing devices724, such as over a network. These networks may include wired networks as well as wireless networks. Communication connections722are some examples of communication media. Communication media may typically be embodied by computer readable instructions, data structures, program modules, etc.

It is appreciated that the illustrated computing device700is only one example of a suitable device and is not intended to suggest any limitation as to the scope of use or functionality of the various embodiments described. Other well-known computing devices, systems, environments and/or configurations that may be suitable for use with the embodiments include, but are not limited to personal computers, server computers, hand-held or laptop devices, multiprocessor systems, microprocessor-base systems, set top boxes, game consoles, programmable consumer electronics, network PCs, minicomputers, mainframe computers, distributed computing environments that include any of the above systems or devices, and/or the like.

The implementation and administration of a shared resource computing environment on a single computing device may enable multiple computer users to concurrently collaborate on the same computing task or share in the same computing experience without reliance on networking hardware such as, but not limited to, network interface cards, hubs, routers, servers, bridges, switches, and other components commonly associated with communications over the Internet, as well without reliance on the software applications and protocols for communication over the Internet.