Data shuffling with hierarchical tuple spaces

Methods and systems for shuffling data to generate a dataset are described. A first map module may generate first pair data, and a second map module may generate second pair data, from source data. The first map module may insert the first pair data into a first local tuple space accessible to the first map module. The second map module may insert the second pair data into a second local tuple space accessible to the second map module. A shuffle module may request pair data that includes a particular key. The first and second pair data may be inserted into a global tuple space accessible by the first and second map modules. The shuffle module may identify the requested pair data in the global tuple space, and may fetch the identified pair data from a memory. The shuffle module may shuffle the fetched pair data to generate the dataset.

FIELD

The present application relates generally to computers, and computer applications, and more particularly to computer-implemented methods and systems data management systems.

BACKGROUND

In data management systems, data shuffling is often used as a building block for various operations on data (e.g. sort, reduce, group), and is widely used in big data frameworks (e.g., Spark, MapReduce, Hadoop etc.). In some examples, data shuffling may be associated with the repartitioning and aggregation of data in an all-to-all operations.

SUMMARY

In some examples, methods for shuffling source data to generate a dataset are generally described. The methods may include generating, by a first map module of a processor, first pair data from the source data. The methods may further include generating, by a second map module of the processor, second pair data from the source data. Each pair data among the first pair data and the second pair data may include a key and a value associated with the key. The methods may further include inserting, by the first map module of the processor, the first pair data into a first local tuple space accessible by the first map module. The methods may further include inserting, by the second map module of the processor, the second pair data into a second local tuple space accessible by the second map module. The methods may further include activating, by the processor, a shuffle module of the processor to execute a shuffle operation on pair data that includes a particular key. The methods may further include inserting, by the processor, and upon the activation of the shuffle module, the first pair data into a global tuple space accessible by the first map module and the second map module. The methods may further include inserting, by the processor, and upon the activation of the shuffle module, the second pair data into the global tuple space. The methods may further include identifying, by the shuffle module of the processor, pair data including the particular key in the global tuple space. The methods may further include fetching, by the shuffle module of the processor, the identified pair data from a memory. The methods may further include executing, by the shuffle module of the processor, the shuffle operation on the fetched pair data to generate the dataset. The dataset may include the particular key and one or more values associated with the particular key.

In some examples, systems effective to shuffle source data to generate a dataset are generally described. An example system may include a memory configured to store the source data. The system may further include a processor configured to be in communication with the memory. The system may further include at least a first map module and a second map module configured to be in communication with the memory and the processor. The system may further include at least one shuffle module configured to be in communication with the memory and the processor. The first map module may be configured to generate first pair data from the source data. The first map module may be further configured to insert the first pair data into a first local tuple space accessible by the first map module. The second map module may be configured to generate second pair data from the source data. The second map module may be further configured to insert the second pair data into a second local tuple space accessible by the second map module. The processor may be configured to activate the shuffle module to execute a shuffle operation on pair data that includes a particular key. The processor may be further configured to insert, upon the activation of the shuffle module, the first pair data into a global tuple space accessible by the first map module and the second map module. The processor may be further configured to insert upon the activation of the shuffle module, the second pair data into the global tuple space. The shuffle module may be further configured to identify the pair data that includes the particular key in the global tuple space. The shuffle module may be further configured to fetch the identified pair data from the memory. The shuffle module may be further configured to execute the shuffle operation on the fetched pair data to generate the dataset. The dataset may include the particular key and one or more values associated with the particular key.

In some examples, computer program products for shuffling source data to generate a dataset are generally described. The computer program products may include a computer readable storage medium having program instructions embodied therewith. The program instructions may be executable by a device to cause the device to generate first pair data from the source data. The program instructions may be further executable by a device to cause the device to generate second pair data from the source data, wherein each pair data among the first pair data and the second pair data includes a key and a value associated with the key. The program instructions may be further executable by a device to cause the device to insert the first pair data into a first local tuple space accessible by a first map module of the device. The program instructions may be further executable by a device to cause the device to insert the second pair data into a second local tuple space accessible by a second map module of the device. The program instructions may be further executable by a device to cause the device to activate a shuffle phase indicated by the program instructions to execute a shuffle operation on pair data that includes a particular key. The program instructions may be further executable by a device to cause the device to insert, upon the activation of the shuffle phase, the first pair data into a global tuple space accessible by the first map module and the second map module. The program instructions may be further executable by a device to cause the device to insert, upon the activation of the shuffle phase, the second pair data into the global tuple space. The program instructions may be further executable by a device to cause the device to identify pair data including the particular key in the global tuple space. The program instructions may be further executable by a device to cause the device to fetch the identified pair data from a memory. The program instructions may be further executable by a device to cause the device to execute the shuffle operation on the fetched pair data to generate the dataset. The dataset may include the particular key and one or more values associated with the particular key.

DETAILED DESCRIPTION

Briefly stated, methods and systems for aggregating data to generate a dataset are described. A first map module may generate first pair data, and a second map module may generate second pair data, from source data. The first map module may insert the first pair data into a first local tuple space accessible by the first map module. The second map module may insert the second pair data into a second local tuple space accessible by the second map module. A processor may activate a shuffle module to request pair data that includes a particular key. The first and second pair data may be inserted into a global tuple space accessible by the first and second map modules. The shuffle module may identify the requested pair data in the global tuple space, and may fetch the identified pair data from a memory. The shuffle module may execute a shuffle operation on the fetched pair data to generate the dataset.

FIG. 1illustrates an example computer system100that can be utilized to implement data shuffling with hierarchical tuple spaces, arranged in accordance with at least some embodiments described herein. In some examples, system100may be a distributed system including a plurality of processing nodes. System100may be a computer system, and may include a processor120, a memory controller121, a memory122, one or more map modules132(including map modules132a,132b, etc.), and one or more shuffle modules140(including shuffle modules140a,140b, etc.). Processor120, memory controller121, memory122, map modules132, and shuffle modules140may be configured to be in communication with each other. In some examples, processor120, memory controller121, memory122, map modules132, and shuffle modules140may be housed, or distributed, in a same housing and/or a computer device. In some examples, processor120, memory controller121, memory122, and map modules132, and shuffle modules140may be housed, or distributed, in two or more different housings and/or computer devices. For example, processor120, memory controller121and memory122may be distributed in a first device and map modules132, and shuffle modules140may be distributed in a second device different from the first device. In some examples, more than two map modules and more than two shuffle modules, may be included in system100depending on a desired implementation. For example, system100may be designed to optimize shuffle operations between two or more shuffle modules such that a distributed nature of data shuffling on multiple computing nodes may be captured, and multiple instances of the shuffle modules may exist simultaneously.

In another embodiment, processor120, memory controller121, memory122, and map modules133, and shuffle modules140may each be hardware components or hardware modules of system100. In some examples, map modules132, and shuffle modules140may each be a hardware component, or hardware modules, of processor120. In some examples, processor120may be a central processing unit of a computer device. In some examples, processor120may control operations of map modules132, and shuffle modules140. In some examples, each map modules132, and shuffle modules140may each include electronic components, such as integrated circuits. In some examples, each map module132and each shuffle module140may be software modules that may be implemented with processor120, or may be software modules that may be implemented with processor120to execute respective threads (e.g., map threads, reduce threads, shuffle threads, etc.). In some examples, processor120may be configured to control operations of memory controller121. In some examples, processor120may be configured to run an operating system that includes instructions to manage map modules132, and shuffle modules140and memory122. In some examples, memory controller121may be integrated as a chip on processor120. Memory controller121may be configured to manage a flow of data to and from memory122.

Memory122may be configured to store a data shuffling instruction124. Data shuffling instruction124may include one or more set of instructions to facilitate implementation of system100. In some examples, data shuffling instruction124may include instructions executable by an operating system running on processor120to manage virtual memory operations and mappings between virtual memory and memory122. In some examples, data shuffling instructions124may be implemented using other methods, such as being implemented in runtime feature, and may be accessed via application programming interface (API) calls. In some examples, memory122may be a main memory of a device configured to implement system100. In some examples, memory122may include persistent storage components, or may include dynamic random access memory (DRAM) components.

In an example, system100may receive a query110indicating an inquiry to process data stored in memory122to generate a dataset170. Generation of dataset170may include reorganizing, sorting, grouping, filtering, joining, word counting, etc. based on an index or key of each piece of data among source data126. For example, query110may inquire a number of occurrences of each unique word among source data126stored in memory122. System100may be implemented to generate a dataset170, where dataset170may be a piece of data including a response to query110. In an example associated with word counting, dataset170may include data indicating a number of occurrence for each unique word among source data126. In an example associated with grouping, dataset170may include one or more groups of data, where each group may correspond to a respective key.

In another example, query110may indicate an inquiry to generate dataset170including information of a plurality, and/or a significantly large amount, of different users (e.g., one million, two million, etc.) of a set of social network platforms. Each key may be an identification of a user, such as a username, a name, an ID number, etc. A particular user may have provided a name on a first and second social network platforms, but may have provided an age on the first social network platform and a location on a second social network platform. System100may be implemented to combine the information of the particular user such that the generated dataset170may include key-value data indicating the name of the particular user, and also the age and location of the particular user, as one piece of data.

Processor120may partition source data126into one or more partitions, such as partitions128a,128b. In an example, source data126may be a database including a significantly large amount of data, and each partition may correspond to a portion of the database such as a number of rows. Processor120may activate a map phase of the implementation of system100, such as by activating one or more map modules132. Processor120may assign a map module to generate pair data, or a set of key-value pairs, for each partition. For example, processor120may assign map module132ato generate pair data134from partition128a, and may assign map module132bto generate pair data135from partition128b. As will be described in more detail below, pair data134,135may each include one or more key-value pairs, and each key-value pair may include a key and a value associated with the key. For example, if query110is an inquiry regarding a number of occurrences of words, a key may be a word and a value may be a number of occurrence of the word, or may be a value to indicate a singular presence of the word (e.g., “1” being present). Each map module may store respective generated pair data in memory122at a respective set of memory addresses. For example, map module132amay store the generated pair data134in memory122at memory addresses150, and map module132bmay store the generated pair data135in memory122at memory addresses152. Memory122may include one or more different sets of memory addresses assigned, or allocated to, different map modules. For example, memory addresses150may be assigned to map module132aand memory addresses152may be assigned to map module132b. In another example, processor120may generate pair data134,135and may send pair data134,135to respective map modules132a,132bfor subsequent processing.

Processor120may, for example, run an operating system to create virtual memory spaces, such as a local tuple space160, a local tuple space162, and a global tuple space164. Local tuple space160may be assigned to map module132aand local tuple space162may be assigned to map module132b. In some examples, an example tuple space may be associated with a concept of a computation environment implementing an associative memory model for distributed/parallel programming. Tuple spaces may also be associated with fundamental mechanisms of various programming languages.

In an example, local tuple space160may be assigned to map module132asuch that other map modules (e.g., map modules132b), may not have access to local tuple space160. In an example, when map module132bdoes not have access to local tuple space160, map module132bmay fail to determine a storage location of pair data134generated and/or stored by map module132a. Similarly, local tuple space162may be assigned to map module132band may be inaccessible by map module132a.

Map modules132may each insert respective generated pair data into an assigned local tuple space. For example, map module132amay insert pair data134into local tuple space160and map module132bmay insert pair data134into location tuple space162. Insertion of a piece of pair data into a local tuple space may include populating an entry of the local tuple space with an indication, or identification, of the piece of pair data (further described below).

Processor120may map memory addresses of memory122to one or more locations, or entries of local tuple spaces160,162based on the insertion of pair data134,135in local tuple spaces160,162. For example, processor120may map a memory address storing pair data134to a location among local tuple space160, and may generate one or more page tables, such as mappings158, to indicate the mappings between memory addresses150,152of memory122and local tuple spaces160,162. Processor120may store mappings158in memory122.

In an example, data shuffling instructions124may include instructions indicating a need to activate a shuffle phase of an implementation of system100subsequent to a map phase of the implementation (e.g., map phase may include, for example, generation of pair data and insertion of pair data into local tuple spaces). A shuffle phase may correspond to, for example, a reduce phase in a mapreduce framework, a key-based sorting phase, a key-based grouping phase, etc. Thus, upon a completion of generating pair data134,135, and mapping memory addresses of memory122to local tuple spaces, processor120may activate one or more shuffle modules140to execute data shuffling operations (e.g., aggregate, sort, filter, group, etc.) in order to generate dataset170. For example, processor120may activate shuffle module140ato aggregate pair data including a first key, and may activate shuffle module140bto aggregate pair data that includes a second key (further described below). In some examples, processor120may activate shuffle modules140to sort, join, group, organize, pair data based on one or more keys. Shuffle modules140may be configured to execute key-based shuffle operations associated with data shuffling such as sorting, joining, grouping, etc. pair data based on one or more keys of the pair data.

Upon the activation of shuffle modules140a,140b, each shuffle module140may request a particular key from the global tuple space, such as by communicating with memory controller121and/or processor120to identify memory addresses storing pair data that includes particular keys. For example, shuffle module140amay generate a request141afor pair data including the first key, and shuffle module140bmay generate a request141bfor pair data including the second key. In response to activation of shuffle modules140a,140b, and/or in response to requests141a,141b, processor120may insert pair data among each local tuple space to global tuple space164, where global tuple space164may be accessible by all map modules including map modules132a,132b. Insertion of a piece of pair data into global tuple space164may include populating an entry of the global tuple space with an indication, or identification, of the piece of pair data (further described below). In some examples, processor120may append metadata to a key of each pair data prior to inserting the pair data into global tuple space164(further described below). Processor120may map memory addresses of memory122that stored pair data134,135to entries, or locations, of global tuple space164. Processor120may update mappings158to include mappings between memory122and global tuple space164.

Similarly, shuffle module140bmay identify pair data142bin global tuple space164, where pair tuple142bmay include key-value pairs including the second key requested by shuffle module140b. Shuffle module140b, based on mappings158, may identify memory addresses of memory122that stored pair data142b, where the identified memory addresses may include memory addresses among both memory addresses150,152. Shuffle module140bmay fetch pair data142bfrom the identified set of memory addresses.

Shuffle module140amay aggregate the fetched pair data142athat includes the first key to generate a piece of pair data that may be a part of dataset170. For example, shuffle module140amay generate a piece of pair data including the first key and one or more aggregated values associated with the first key (further described below). Similarly, shuffle module140bmay aggregate the fetched pair data142bthat includes the second key to generate a piece of pair data that may be a part of dataset170. Processor120may further combine the pair data generated by shuffle module142a, shuffle module142b, and/or additional shuffle modules, to complete a generation of dataset170. As a result, dataset170may be include a plurality of key-value pairs, where each key-value pair includes a key, and includes one or more aggregated values associated with the corresponding key (further described below).

FIG. 2illustrates the example system ofFIG. 1with additional details relating to data shuffling with hierarchical tuple spaces, arranged in accordance with at least some embodiments described herein.FIG. 2is substantially similar to computer system100ofFIG. 1, with additional details. Those components inFIG. 2that are labeled identically to components ofFIG. 1will not be described again for the purposes of clarity.

In an example shown inFIG. 2, system100may receive a query110indicating an inquiry to count a number of occurrences of each unique word among source data126stored in memory122. Source data126may include one or more occurrences of words “k1”, “k2”, “k3”. Processor120may partition source data126into partitions128a,128b, where partition128aincludes words “k1”, “k3”, “k3”, “k1”, and partition128bincludes words “k2”, “k3”, “k1”, “k3”. Processor120may assign map module132ato generate pair data134from partition128a, and may assign map module132bto generate pair data135from partition128b.

In an example, map module132amay convert each word among partition128ainto a key-value pair, such as by counting an occurrence of each word. As shown in the example, a key-value pair (k1,1) may represent a singular occurrence of the word “k1”. In the example, pair data134generated from partition128amay include key-value pairs (k1,1), (k3,1), (k3,1), (k1,1). Map module132amay store pair data134in memory122at a set of memory addresses150a,150b,150c,150d. Similarly, pair data135generated from partition128bmay include key-value pairs (k2,1), (k3,1), (k1,1), (k3,1). Map module132bmay store pair data135in memory122at a set of memory addresses152. Map module132amay insert pair data134into local tuple space160, where local tuple space160may be accessible by map module132aand may be inaccessible by map module132b. Map module132amay insert an indication, such as a key-value pair, of pair data134into locations of local tuple space160. For example, map module132amay insert the key-value pair (k1,1) into location160aof local tuple space160, and may insert the key-value pair (k1,1) into location160dof local tuple space160. Similarly, map module132bmay insert pair data135into local tuple space162, where local tuple space162may be accessible by map module132band may be inaccessible by map module132a.

Upon storing pair data134,135in memory122, and inserting pair data134,135into local tuple spaces160,162, processor120may map memory addresses150a,150b,150c,150dto one or more locations of local tuple space160. In the example shown inFIG. 2, memory address150amay be mapped to location160aof local tuple space160(based on (k1,1) being stored at memory address150aand inserted in location160a), and memory address150dmay be mapped to location160bof local tuple space160(based on (k1,1) being stored at memory address150dand inserted in location160d). A selection of mapping locations to map memory addresses150may be performed by processor120, or by an operating system being executed by processor120, or memory controller121, based on various memory management algorithms. In response to mapping memory addresses150to local tuple space160, processor120may update mappings158stored in memory122.

Processor120may activate and assign a shuffle module to aggregate pair data based on a same key. For example, if the query received at system100relates to word counting, processor120may assign a set of shuffle modules to count words, where each shuffle module may be responsible to count one particular word. In the example, processor120may assign shuffle module140ato determine a number of occurrences of word “k1” among source data126, and may assign shuffle module140bto determine a number of occurrences of word “k2” among source data126.

Shuffle module140amay generate a request141afor pair data including word “k1”, and shuffle module140bmay generate a request141afor pair data including word “k2”. In response to requests141a,141b, processor120may insert pair data134,135into global tuple space164. Upon the insertion of pair data134,135into global tuple space164, global tuple space164may include indications of all key-value pairs among pair data134,135, as shown inFIG. 2. Processor120may map memory addresses150,152, which stored pair data134,135, to locations of global tuple space164. Processor120may update mappings158to include mappings between memory122and global tuple space164.

In some examples, processor120may append metadata to a key of each pair data prior to inserting the pair data into global tuple space164. For example, processor120may append an indicator to each key among pair data134, where the indicator may indicate locality information such as an identification of a processor/node/executor (e.g., map module132a, or a process configured to run a map tasks), an identification of partition128a, and a memory address storing the pair data with the appended metadata. In some examples, global keys may be generated by appending metadata to each piece of pair data, such that insertion of the pair data into global tuple space164includes insertion of pair data including the global keys.

Shuffle module140amay aggregate fetched pair data142athat includes “k1” to generate a piece of pair data (k1,3) that may be a part of dataset170. For example, in a word counting example, shuffle module140amay sum the values among pair data142ato conclude that “k1” occurred three times, and output the key-value pair (k1,3). Shuffle module140bmay aggregate fetched pair data142bthat includes “k2” to generate a piece of pair data (k2,1) that may be a part of dataset170. In an example with information consolidation from social network platforms, shuffle modules may aggregate fetched pair data by eliminating duplicated values, merging similar values, identifying values that only appeared once, etc. In some examples, the aggregation operations by shuffle modules140a,140b, may be performed in parallel.

Processor120may combine the pair data generated by shuffle module142a, shuffle module142b, and/or additional shuffle modules (e.g., another shuffle module may generate (k3,4)), to complete a generation of dataset170. As a result, dataset170may be include a plurality of key-value pairs, where each key-value pair includes a key, and includes one or more aggregated values associated with the corresponding key. As shown in the example, dataset170may include key-value pairs (k1,3), (k2,1), and (k3,4) to indicate that there are three occurrences of word “k1”, one occurrence of word “k2”, and four occurrences of word “k3”, in source data126. Processor120may return the output as a response to a device that sent query110to system100. In some examples, upon a completion of generating dataset170, processor120may remove pair data that have been inserted in local tuple spaces160,162, and global tuple space164. Processor120may further clear mappings158to remove all mappings among memory122, local tuple spaces160,162, and global tuple space164.

FIG. 3illustrates the example system ofFIG. 1with additional details relating to data shuffling with hierarchical tuple spaces, arranged in accordance with at least some embodiments described herein.FIG. 3is substantially similar to computer system100ofFIG. 1andFIG. 2, with additional details. Those components inFIG. 3that are labeled identically to components ofFIG. 1andFIG. 2will not be described again for the purposes of clarity.

In an example diagram shown inFIG. 3, three processor initiated calls, or instructions, may be executed by system100—namely “put( )”, “get( )”, and “read( )”, to implement data shuffling with hierarchical tuple spaces. The call “put( )” may cause key-value pairs of data to be transferred into a local tuple space or into global tuple space164. For example, system100, or a processor (e.g., processor120inFIGS. 1, 2) may execute an API call to instruct map modules132a,132b, to perform the put( ) call to insert pair data134,135into local tuple spaces160,162, respectively. System100may instruct map modules132a,132b, to perform the put( ) call to insert pair data134,135from local tuple spaces160,162to global tuple space164.

The call “get( )” may cause system100to fetch/remove key-value pairs of data from global tuple space into local tuple space. For example, system100, or a processor (e.g., processor120inFIGS. 1, 2) may execute an operating system to instruct shuffle modules140a,140b, to perform the get( ) call to fetch pair data142a,142bfrom global tuple space164, and subsequently, remove pair data142a,142bfrom global tuple space164. In some examples, a blocking call may occur during the get( ) call, such that threads being executed among system100may be suspended until shuffle modules140complete fetching and/or removing pair data from global tuple space164. In some examples, keys requested by shuffle modules140may need to exist in global tuple space164in order for get( ) call to be executed successfully. For example, each shuffle module140may search for a respective key among global tuple space164prior to executing the get( ) call, and may fetch pair data from global tuple space164if the keys exists in global tuple space164.

The call “read( )” may fetch/copy key-value pairs of data from global tuple space into local tuple space. For example, system100, or a processor (e.g., processor120inFIGS. 1, 2) may execute an operating system to instruct shuffle modules140a,140b, to perform the read( ) call to copy pair data142a,142bfrom global tuple space164to another location (e.g., memory address, or local tuple spaces assigned to shuffle modules that may be different from local tuple spaces160,162), without removing pair data from global tuple space164. In some examples, a blocking call may occur during the read( ) call, such that threads being executed among system100may be suspended until shuffle modules140complete fetching and/or copying pair data from global tuple space164. In some examples, keys requested by shuffle modules140may need to exist in global tuple space164in order for read( ) call to be executed successfully. For example, each shuffle module140may search for a respective key among global tuple space164prior to executing the read( ) call, and may fetch and/or copy pair data from global tuple space164if the keys exists in global tuple space164.

In summary, key-value pairs may be loaded from a persistent storage (e.g., memory122), or created by a computation (e.g., map module132) in local memory for each worker thread performed by each map module. Initially, each key-value pair may be added to an assigned local tuple space. At the beginning of a reduce stage, map modules may use a “put( )” call to put requested blocks into the global tuple space. As such, local key value pairs stored in local tuple space may be exposed to the global tuple space, without an actual data transfer. Shuffle modules on the reduce stage may fetch key-value pairs from the global tuple space into respective local tuple space using either a “get( )” or a “read( )” call. The use of “get( )” or “read( )” calls (e.g., move vs. copy, respectively) may be defined by application level hints for persistency (e.g., subsequent use of the data).

A system in accordance with the present disclosure may facilitate an improvement in data shuffle mechanisms by establishing an in-memory hierarchical tuple spaces for key-value pairs as generated from the mappers. The tuple spaces may be hierarchical with local and global spaces, and mappers may transfer local key-value pairs from local to the global tuple space that also resides in a distributed fashion within the system memory. Reducers may fetch key-value pairs from the global tuple space, and the transfer action may be triggered by the fetch request from reducers. As a result, disk I/O overhead may be prevented by utilizing hierarchical in-memory tuple spaces as described above. Further, a system in accordance with the present disclosure may intercept a shuffle write process to avoid file generation for key-value pairs, and may collect all generated key-value pairs inside a local tuple space. Each local tuple space corresponding to a physical computing node may gather key-value pairs from all executors/threads inside the physical computing node. Each key may be appended with a specific executor/partition meta data such that a global key may be generate and data locality may be improved.

In conventional data shuffling schemes, all-to-all communication may be required and thus, may incur major performance cost, may be complex, and may cause problems such as bandwidth and latency issues. Further, typical data shuffling techniques may depend on filesystem and may include I/O limitations. Many current shuffle implementations may store data in blocks on local or distributed disk I/O for data shuffling, which may cause major overhead on an operating system, and both the source and the destination side may require many file and network I/O operations. Some existing data aggregation techniques used for filesystem and communication optimization may add extra computation overheads. For example, techniques that merges files into buckets then reduce the total number of files may use many files, or techniques that utilize sort-based shuffle such that each mapping task may generate one shuffle data file and one index file, may use file I/O operations to store and manage shuffle files.

A system in accordance with the present disclosure may be used to avoid dependency on files for distributed data representation, and eliminate filesystem or disk I/O operations and extra computation to prepare data partitions in the shuffle write stage. A burden on the operating system for managing filesystem and I/O operations may also be mitigated. By eliminating the dependency on the filesystem, a system in accordance with the present disclosure may be used to establish a more efficient full in-memory shuffle mechanism.

FIG. 4illustrates a flow diagram for an example process to implement data shuffling with hierarchical tuple spaces, arranged in accordance with at least some embodiments presented herein. The process inFIG. 4could be implemented using, for example, computer system100discussed above. An example process may include one or more operations, actions, or functions as illustrated by one or more of blocks401,402,403,404,405,406, and/or407. Although illustrated as discrete blocks, various blocks may be divided into additional blocks, combined into fewer blocks, eliminated, or performed in parallel, depending on the desired implementation.

Processing may begin at blocks401(including blocks401a,401b), where a first map module of a processor may generate first pair data from source data, and a second map module of the processor may generate second pair data from the source data. In some examples, generation of the first pair data and the second pair data may happen in parallel. Each pair data may include a set of key-value pairs, and each key-value pair may include a key and one or more values associated with the key. For example, a key may be a word, and a value of the key may be a number of occurrences of the word. In another example, a key may be an identification of a person (e.g., name, ID number, etc.) and one or more values associated with the person may be an age, an address, an occupation, education background, hobbies, etc.

Processing may continue from blocks401to blocks402(including blocks402a,402b), where the first map module may insert the first pair data into a first local tuple space, and the second map module may insert the second pair data into a second local tuple space. The first local tuple space may be accessible by the first map module and may be inaccessible by the second map module. The second local tuple space may be accessible by the second map module and may be inaccessible by the first map module.

At blocks403, the processor may activate a first shuffle module of the processor may activate a second shuffle module of the processor. Upon activation, the first shuffle module may request pair data including a first key. Similarly, upon activation, the second shuffle module may request pair data including a second key.

At blocks404, in response to the requests from blocks403, the processor may insert the first pair data and the second pair data into a global tuple space. The global tuple space may be accessible by both the first map module and the second map module.

Processing may continue from blocks404to blocks405, where the first shuffle module may fetch pair data including the first key from the global tuple space, and the second shuffle module may fetch pair data including the second key from the global tuple space. In some examples, the first shuffle module and the second shuffle module may perform a search for each respective requested key prior to performing a fetch operation. If a requested key exists in the global tuple space, then the fetch operations may be performed. In some examples, the processor may block calls from other threads or operations such that the key search and fetch operations may be performed by the shuffle modules.

Processing may continue from blocks405to block406, where the first shuffle module may generate first output pair data and the second shuffle module may generate second output pair data. The first output pair data may include the first key and a first aggregated value, where the first aggregated value may include one or more values associated with the first key. The second output pair data may include the second key and a second aggregated value, there the second aggregated value may include one or more values associated with the second key. For example, if the first key is a name of a first person, the fetched pair data may include one or more key-value pairs such as (name, age), (name, age, location), (name, location, education), etc. collected from one or more social media platforms. The first aggregated value may be a combination of the fetched pair data such that the first output pair data may be (name, age, location, education).

Processing may continue from blocks406to blocks407, where the processor may generate a dataset based on the first output pair data and the second output pair data. The generated dataset may include the first output pair data and the second output pair data.

FIG. 5illustrates a schematic of an example computer or processing system that may implement any portion of computer system100, processor120, memory122, map modules132, shuffle modules140, systems, methods, and computer program products described herein in one embodiment of the present disclosure. The computer system is only one example of a suitable processing system and is not intended to suggest any limitation as to the scope of use or functionality of embodiments of the methodology described herein. The processing system shown may be operational with numerous other general purpose or special purpose computer system environments or configurations. Examples of well-known computer systems, environments, and/or configurations that may be suitable for use with the processing system may include, but are not limited to, personal computer systems, server computer systems, thin clients, thick clients, handheld or laptop devices, multiprocessor systems, microprocessor-based systems, set top boxes, programmable consumer electronics, network PCs, minicomputer systems, mainframe computer systems, and distributed cloud computing environments that include any of the above systems or devices, and the like.