Addressing data skew using map-reduce

A system and method includes using a queue with map-reduce. The system includes a computer cluster that is to execute, by a first node, a first reduce operation on a first location of data to generate a first plurality of markers indicative of data at the first location of data and execute, by a second node, a second reduce operation on a second location of data to generate a second plurality of markers indicative of data at the second location of data. Responsive to generation of one or more markers, the computer cluster is to submit the one or more markers to a queue. Responsive to completing the first reduce operation by the first node, the computer cluster is to direct the first node to perform a first copy operation that copies first data identified by a first marker of the one or more markers in the queue.

TECHNICAL FIELD

This disclosure relates to the field of map-reduce and, in particular, addressing data skew with map-reduce.

BACKGROUND

Data can be copied from a first location of data to a second location of data. The first location of data may be associated with a first server and the second location of data may be associated with a second server.

DETAILED DESCRIPTION

Locations of data (e.g., databases) may be associated with a first server. For example, production databases may be associated with a production server that is used to support the operations of a platform, such as a collaboration platform. Performing analytics on the databases associated with a production server may affect the performance of the production server and the platform generally. In some cases, to perform analytics the production databases may be copied for use by a second server (e.g., an analytics server). The analytics server may perform analytics on the copied databases and not affect the performance of the production server.

In some systems, performing the processes to copy data from databases to other databases may be performed by one more computing devices, such as server computers or processing devices. For example, a single computing device may prepare to copy and copy each database serially, one after the other, which may take a prohibitively large amount of time. In another example, multiple computing devices may prepare to copy and copy data in parallel. Data of a database may be skewed. Data skew may refer to data that is distributed unevenly across databases. Data skew may cause inefficiencies in the performance of processes to copy data from databases. For example, data skew may cause some of the computing devices that perform the copy processes to go unused and remain idle. For instance, some computing devices that perform one or more of the copy processes may finish a particular process (e.g., on databases with a smaller amount of data) quickly and may sit idle and wait for other computing devices to finish the particular process (e.g., on other databases with larger amounts of data). Having idle computing devices is an inefficient use of computer resources (e.g., computer processing resources) and reduces the speed at with copy operations are performed.

Aspects of the present disclosure address the above-mentioned and other challenges by using computer nodes of a computer cluster that implement map-reduce operations to copy data of databases of a production server to the databases of another server, such as an analytics server. Identifiers of the databases to be copied are submitted to a node in the computer cluster. The node may perform a map operation on the identifiers where each of the identifiers is distributed to another computer node in the computer cluster. The other nodes may use the identifiers to perform a reduce operation on the respective databases identified by the identifiers. The other nodes performing the reduce operation generate output data, which may be markers that identify locations of data in the respective database. As the markers are generated, the markers can be stored in a queue associated with and accessible by the computer nodes of the computer cluster. Responsive to a computer node completing a reduce operation, rather than remain idle, the computer node can be repurposed to use the markers in the queue to perform a copy operation to copy data of databases of the production server to databases of another server. Other computer nodes in the computer cluster, such as computer nodes that perform reduce operations on databases with disproportionally more amounts of data, may continue to perform the reduce operation as other computer nodes that finish respective reduce operations are repurposed to perform copy operations.

Copying or other types of processing (e.g., any other operations on the data) of data using a computer cluster that implements a queue and map-reduce operations in accordance with aspects of the disclosure improves computer-related technology at least because aspects of the disclosure copy data (e.g., as data from databases of a first server to databases of another server) or process data, faster and more efficiently (e.g., more efficiently uses computer processing resources and reduces the amount of computer processing resources that idle) than other systems. The improvements in computer-related technology (e.g., speed and efficiency) are further enhanced when used to copy or process data of databases that have data skew.

It may be noted that a map operation and a reduce operation described as part of a map-reduce programming model is provided for purposes of illustration, rather than limitation. Distributing identifiers of databases to nodes of a cluster, generating markers, and copying data can take place in one or more programming models, such as hash partitioning, self-balancing tree data structure (e.g., B-tree), relational database management system (RDBMS), and so forth.

It may be noted that markers are described as being used to copy data from a first database associated with a first server to a second database associated with a second server for purposes of illustration, rather than limitation. In other implementations, the markers may be used in any number of ways, or other data processing techniques for example. In some implementations, the markers may identify location of data. In other implementations, the markers may include the data. It may further be noted that the disclosure describes copying data from databases to other databases for purposes of illustration, rather than limitation. Aspects of the present disclosure may be applied to copying data generally. For example, aspects of the present disclosure may be applied to copying data from a first location of data (e.g., one or more first data stores) to a second location of data (e.g., one or more second data stores). In may be further noted that the disclosure describes copying data for purposes of illustration, rather than limitation. Aspects of the present disclosure may be applied to any type of processing of data (e.g., any other operations on the data). For example, aspects of the present disclosure may be applied to converting of data (e.g., from a first form to a second form), validation of data, sorting of data, summarization of data, aggregation of data, analysis of data, reporting of data, etc.

FIG. 1illustrates an example system architecture100, in accordance with implementations of the disclosure. The system architecture100(also referred to as “system” herein) includes client devices110A and110B (generally referred to as “client device(s)110” herein), a network105, servers140A and140B (generally referred to as “servers140”), a collaboration platform120, and a computer cluster130. It may be noted that system architecture100is provided for illustration, rather than limitation. In implementations, the system architecture100may include the same, fewer, more, or different elements configured in the same or different manner.

In one implementation, network105may include a public network (e.g., the Internet), a private network (e.g., a local area network (LAN) or wide area network (WAN)), a wired network (e.g., Ethernet network), a wireless network (e.g., an 802.11 network, a Wi-Fi® network, or wireless LAN (WLAN)), a cellular network (e.g., a Long Term Evolution (LTE) network), routers, hubs, switches, server computers, or a combination thereof.

In implementations, each server140may be one or more computing devices (e.g., a rackmount server, a server computer, cluster of physical servers, etc.). In implementations, the servers140may be included in the collaboration platform120, be an independent system, or be part of another system or platform. Servers140may be associated with (e.g., include) a memory (e.g., random access memory), a cache, a drive (e.g., a hard drive), a flash drive, a database system, or another type of component or device capable of storing data. Servers140may be associated with (e.g., include) multiple storage components (e.g., multiple drives or multiple databases) that may also span multiple computing devices (e.g., multiple server computers). The server140A may be associated with (e.g., store) databases142A-142N (generally referred to as “databases142”). The server140B may be associated with (e.g., store) databases144A-144N (generally referred to as “databases144”). In implementations, databases142may be independent from first server140A and databases144may be independent from second server140B.

In some implementations, server140A may be a production server and server140B may be an analytics server. Server140A may be part of the collaboration platform120. Databases142may be production databases and may include data146generated by one or more of server140A, collaboration platform120, or client device110. Databases144may be analytics databases and may include copies of data146from databases142. Data146may be copied from databases142to databases144and the copy of data146in databases144may be analyzed (e.g., to not affect performance of server140A).

In some implementations, the collaboration platform120may be one or more computing devices (such as a rackmount server, a router computer, a server computer, a personal computer, a mainframe computer, a laptop computer, a tablet computer, a desktop computer, etc.), data stores (e.g., hard disks, memories, databases), networks, software components, and/or hardware components that may be used to perform operations on the collaboration platform120and to provide a user with access to collaboration platform120. The collaboration platform120may also include a website (e.g., a webpage) or application back-end software that may be used to provide a user with access to content provided by collaboration platform120. For example, users may access collaboration platform120using collaboration application114on client devices110. It may be noted that collaboration application114A and114B may generally referred to as collaboration application(s)114herein. In some implementations, collaboration application114may be two instances of the same application.

As discussed herein, collaboration platform120(and/or collaboration application114) may generate (e.g., gather, process, etc.) data146associated with user access and interaction with collaboration platform120, games122, media items, game content, additional functionalities121, game engine124, creator module126, messaging module128, etc. The collaboration platform may store the data146in databases142. For example, collaboration platform120may generate data146(for a database142) including an identifier for a user, an identifier for a game122or media item with which the user is interacting, a start time that the user started interacting with the game122or media item, an end time that the user stopped interacting with the game122or media item, etc. In one implementation, collaboration platform120stores data146for one or more games122or media items in the same database142. In another implementation, collaboration platform120stores data146for one or more users in the same database142. In another implementation, collaboration platform120stores data146for one or more periods of time in the same database142. For example, database142A may include data146A from a first day and database142B may contain data146B from a second day.

In implementations, collaboration platform120may be a type of social network providing connections between users or a type of user-generated content system that allows users (e.g., end-users or consumers) to create content for the platform, where the created content may also be consumed by other users of the system. In implementations of the disclosure, a “user” may be represented as a single individual. However, other implementations of the disclosure encompass a “user” (e.g., creating user) being an entity controlled by a set of users or an automated source. For example, a set of individual users federated as a community or group in a user-generated content system may be considered a “user.” One or more of collaboration platform120or collaboration application114may generate (e.g., gather, process, etc.) data146associated with the user-generated content and may store the data146in database142.

In one implementation, collaboration platform120may be a gaming platform, such as an online gaming platform or a virtual gaming platform. For example, the gaming platform may provide single-player or multiplayer games to a community of users that may access or interact with the games122A-122Z using client devices110via network105. In implementations, games122(also referred to as “video game,” “online game,” or “virtual game” herein) may be two-dimensional (2D) games, three-dimensional (3D) games (e.g., 3D user-generated games using creator module126), virtual reality (VR) games, or augmented reality (AR) games, for example. In implementations, users, such as playing users may participate in gameplay with other playing users. In implementations, a game122may be played in real-time with other users of the game122.

In implementations, gameplay may refer to interaction of one or more players with a game (e.g., user inputs) or the presentation of the interaction on a display of a client device. In implementations, real-time or live gameplay may refer to the broadcast, presentation, or display of gameplay as the gameplay occurs, rather than past or recorded gameplay. For example, real-time gameplay may be displayed on a display of a client device110substantially concurrent with user interaction with a game (at least within the technical limitations of displaying real-time action, e.g., the user interaction with the game may be displayed within milliseconds of user input and may appear to the user as if in absolute real-time).

In implementations, a view of the gameplay may be presented on the client device110. The views on different client devices110may be the same or different. In implementations, a view (also referred to as “field of view” herein) may refer to the extent of the observable game world that may be seen at any given moment from the perspective of a game camera and that is presented in the display of a client device. For example, the view of the game camera may be from a first-person perspective or a third-person perspective or some combination thereof.

In some implementations, a game122can include an electronic file that can be executed or loaded using software, firmware or hardware configured to present the game content (e.g., digital media item) to an entity. In implementations, a game122may be executed and rendered using a game engine124. In some implementations, a game122may have a common set of rules or common goal, and the environments of a game122share the common set of rules or common goal. In implementations, different games may have different rules or goals from one another.

In some implementations, games may have one or more environments (also referred to as “gaming environments” or “virtual environments” herein) where multiple environments may be linked. An example of an environment may be a three-dimensional (3D) environment. The one or more environments of a game122may be collectively referred to a “world” or “gaming world” or “virtual world” or “universe” herein. An example of a world may be a 3D world of a game122. For example, a user may build a virtual environment that is linked to another virtual environment created by another user. A character of the virtual game may cross the virtual border to enter the adjacent virtual environment.

It may be noted that 3D environments or 3D worlds use graphics that use a three-dimensional representation of geometric data representative of game content (or at least present game content to appear as 3D content whether or not 3D representation of geometric data is used). 2D environments or 2D worlds use graphics that use two-dimensional representation of geometric data representative of game content.

In some implementations, collaboration platform120hosts games122and allows users to interact with the games122using collaboration application114of client devices110. Users of collaboration platform120may play, create, interact with, or build games122, or create and build objects (e.g., also referred to as “item(s)” or “game objects” or “virtual game item(s)” herein) of games122. For example, in generating user-generated virtual items, users may create characters, decoration for the characters, one or more virtual environments for an interactive game, or build structures used in a game122, among others. In implementations, users may buy, sell, or trade game virtual game objects, such as in-platform currency (e.g., virtual currency), with other users of the collaboration platform120. In implementations, collaboration platform120may transmit game content to collaboration applications114. In implementations, game content (also referred to as “content” herein) may refer to any data or software instructions (e.g., game objects, game, user information, video, images, commands, media item, etc.) associated with collaboration platform120or collaboration applications114. In implementations, game objects (e.g., also referred to as “item(s)” or “objects” or “virtual game item(s)” herein) may refer to objects that are used, created, shared or otherwise depicted in games122of the collaboration platform120. For example, game objects may include a part, model, character, tools, weapons, clothing, buildings, vehicles, currency, flora, fauna, components of the aforementioned (e.g., windows of a building), and so forth.

It may be noted that collaboration platform120hosting games122, is provided for purposes of illustration, rather than limitation. In some implementations, collaboration platform120may host one or more media items. Media items can include, but are not limited to, digital video, digital movies, digital photos, digital music, audio content, melodies, website content, social media updates, electronic books, electronic magazines, digital newspapers, digital audio books, electronic journals, web blogs, real simple syndication (RSS) feeds, electronic comic books, software applications, etc. In implementations, a media item may be an electronic file that can be executed or loaded using software, firmware or hardware configured to present the digital media item to an entity. One or more of collaboration platform120or collaboration application114may generate (e.g., gather, process, etc.) data146associated with user access to, creation of, and interaction with the content (e.g., game content, media items, etc.) and may store the data146in database142.

In some implementations, a game122may be associated with a particular user or a particular group of users (e.g., a private game), or made widely available to users of the collaboration platform120(e.g., a public game). In implementations, where collaboration platform120associates one or more games122with a specific user or group of users, collaboration platform120may associated the specific user(s) with a game122using user account information (e.g., a user account identifier such as username and password).

In some implementations, collaboration platform120or client devices110may include a game engine124. In implementations, game engine124may be used for the development or execution of games122. For example, game engine124may include a rendering engine (“renderer”) for 2D, 3D, VR, or AR graphics, a physics engine, a collision detection engine (and collision response), sound engine, scripting functionality, animation engine, artificial intelligence engine, networking functionality, streaming functionality, memory management functionality, threading functionality, scene graph functionality, or video support for cinematics, among other features. The components of the game engine124may generate commands that help compute and render the game (e.g., rendering commands, collision commands, physics commands, etc.) In some implementations, game engine124of client devices110may work independently, in collaboration with game engine124of collaboration platform120, or a combination of both.

In some implementations, both the collaboration platform120and client device110execute a game engine124. The collaboration platform120using game engine124may perform some or all the game engine functions (e.g., generate physics commands, rendering commands, etc.), or offload some or all the game engine functions to game engine124of client device110. In some implementations, each game122may have a different ratio between the game engine functions that are performed on the collaboration platform120and the game engine functions that are performed on the client device110. For example, the game engine124of the collaboration platform120may be used to generate physics commands in cases where there is a collision between at least two game objects, while the additional game engine functionality (e.g., generate rendering commands) may be offloaded to the client device110. In some implementations, the ratio of game engine functions performed on the collaboration platform120and client device110may be changed (e.g., dynamically) based on gameplay conditions. For example, if the number of users participating in gameplay of a particular game122exceeds a threshold number, the collaboration platform120may perform one or more game engine functions that were previously performed by client device110.

For example, playing users may be playing a game122on client devices110and may send control instructions (e.g., user inputs, such as right, left, up, down, user election, or character position and velocity information, etc.) to collaboration platform120. Subsequent to receiving control instructions from the client devices110, collaboration platform120may send gameplay instructions (e.g., position and velocity information of the characters participating in the group gameplay or commands, such as rendering commands, collision commands, etc.) to the client devices110based on control instructions. For instance, the collaboration platform120may perform one or more logical operations (e.g., using game engine124) on the control instructions to generate gameplay instruction for the client device110. In other instances, collaboration platform120may pass one or more or the control instructions from one client device110to other client devices participating in the game122. The client devices110may use the gameplay instructions and render the gameplay for presentation on the displays of client devices110.

In implementations, the control instructions may refer to instructions that are indicative of in-game actions of a user's character. For example, control instructions may include user input to control the in-game action, such as right, left, up, down, user selection, gyroscope position and orientation data, force sensor data, etc. The control instructions may include character position and velocity information. In implementations, the control instructions are sent directly to the collaboration platform120. In other implementations, the control instruction may be sent from a client device110A to another client device110B, where the other client device110B generates gameplay instructions using the local game engine124.

In implementations, gameplay instructions may refer to instructions that allow a client device110to render gameplay of a game, such as a multiplayer game. The gameplay instructions may include one or more of user input (e.g., control instructions), character position and velocity information, or commands (e.g., physics commands, rendering commands, collision commands, etc.).

In implementations, collaboration platform120may include a creator module126. In implementations, creator module126may allow users of the collaboration platform120to become creating user that design or create environments in an existing game122, create new games, or create new game objects within games or environments.

In implementations, creator module126may allow a user to create, modify, or customize characters. In implementations, characters (or game objects generally) are constructed from components, one or more of which may be selected by the user, that automatically join together to aid the user in editing. One or more characters (also referred to as an “avatar” or “model” herein) may be associated with a user (also referred to as a “playing user” herein) where the user may control the character to facilitate a user's interaction with the game122. In implementations, a character may include components such as body parts (e.g., hair, arms, legs, etc.) and accessories (e.g., t-shirt, glasses, decorative images, tools, etc.). In implementations, body parts of characters that are customizable include head type, body part types (arms, legs, torso, and hands), face types, hair types, and skin types, among others. In implementations, the accessories that are customizable include clothing (e.g., shirts, pants, hats, shoes, glasses, etc.), weapons, or other tools. In implementations, the user may also control the scale (e.g., height, width, or depth) of a character or the scale of components of a character. In implementations, the user may control the proportions of a character (e.g., blocky, anatomical, etc.). It may be noted that is some implementations, a character may not include a character game object (e.g., body parts, etc.) but the user may control the character (without the character game object) to facilitate the user's interaction with the game (e.g., a puzzle game where there is no rendered character game object, but the user still controls a character to control in-game action).

In some implementations, a component, such as a body part, may be a primitive geometrical shape such as a block, a cylinder, a sphere, etc., or some other primitive shape such as a wedge, a torus, a tube, a channel, etc. In implementations, creator module126may publish a user's character for view or use by other users of collaboration platform120. In some implementations, creating, modifying, or customizing characters, other game objects, games122, or game environments may be performed by a user using a user interface (e.g., developer interface) and with or without scripting (or with or without an application programming interface (API)). It may be noted that for purposes of illustration, rather than limitation, characters are described as having a humanoid form. In may further be noted that characters may have any form such as a vehicle, animal, inanimate object, or other creative form.

In implementations, collaboration platform120may store characters created by users in database142. In implementations, the collaboration platform120maintains a character catalog and game catalog that may be presented to users via collaboration application114. A user may select a game122from the game catalog to play. In implementations, the game catalog includes images of games stored on collaboration platform120. In addition, a user may select a character (created by the playing user or other user) from the character catalog to participate in the chosen game. The character catalog includes images of characters stored on the collaboration platform120. In implementations, one or more of the characters in the character catalog may have been created or customized by the user. In implementations, the chosen character may have character settings defining one or more of the components of the character.

In implementations, a user's character includes a configuration of components, where the configuration and appearance of components and more generally the appearance of the character may be defined by character settings. In implementations, the character settings of a user's character may at least in part be chosen by the user. In other implementations, a user may choose a character with default character settings or character setting chosen by other users. For example, a user may choose a default character from a character catalog that has predefined character settings, and the user may further customize the default character by changing some of the character settings (e.g., adding a shirt with a customized logo). The character settings may be associated with a particular character by collaboration platform120.

In implementations, collaboration platform120executing creator module126includes a user-interfacing website or application (e.g., collaboration application114) where users (also referred to as “creating users,” “creators,” “owners,” or “owning users” herein) may access online computational resources (e.g., cloud resources) hosted by collaboration platform120for the purposes of building, administrating, editing, and interacting with personally owned games122or gaming environments. In implementations, creator module126includes tools available to users for creating and instantiating three-dimensional virtual games or environments. In implementations, creator module126is available to users that wish to create and administer their own private virtual game122. In implementations, a user may access creator module126using collaboration application114. In implementations, creator module126may use a user interface (also referred to as a “developer interface” herein) via collaboration application114to allow users access the functionality of creator module126. In implementations, the developer interface may be part of collaboration application114. For example, a developer interface of collaboration application114may allow a user access to a library of game objects that may be selected by the user to build a game environment or build a game122. The user may publish their game objects via the developer interface so that the game is available to users of collaboration platform120.

In implementations, creator module126may provide control of created games122and environments to creating users who may set administrative policy regarding which users will be allowed to interact with the created game122or environment and which users have the capability of modifying, editing, or interacting with the game122or environment. Users with administrative control may also be referred to as administrative users herein, unless otherwise specified. In some implementations, administrative rules can be granular to the point of specifying various levels of interaction, modification, or editing capabilities for certain individuals who might be recruited or otherwise desire to interact with the game122or gaming environment. For example, individuals may be recruited from friends in collaboration platform120or friends from social networks or in other group associated online venues. In implementations, creator module126has administrative tools for setting such policies including or in addition to general rules for interaction within the game122or gaming environment. One or more of collaboration platform120or collaboration application114may generate (e.g., gather, process, etc.) data146associated with user access to, creation of, and interaction with the games122and may store the data146in database142.

In implementations, the client devices110A through110B may each include computing devices such as personal computers (PCs), mobile devices (e.g., laptops, mobile phones, smart phones, tablet computers, or netbook computers), network-connected televisions, gaming consoles, etc. In some implementations, client devices110A through110B may also be referred to as “user devices.” In implementations, one or more client devices110via collaboration application114may connect to collaboration platform120at any given moment. It may be noted that the number of client devices110is provided as illustration, rather than limitation. In implementations, any number of client devices110may be used.

The client devices110A-B may host various applications, including, for example, web applications, desktop applications, browser applications, etc. In one implementation, an application is a map-reduce skew module170(e.g., a map-reduce interface application). The map-reduce skew module170may allow a user of a client device110to send a map-reduce job to be processed and may receive a result of a map-reduce job. The map-reduce skew module170may allow the client device to interact with (e.g., submit map-reduce jobs to) computer cluster130(e.g., master node132).

In implementations, each client device110may include an instance of collaboration application114. In one implementation, the collaboration application114may be an application that allows users to use and interact with collaboration platform120, such as control a virtual character in a virtual game hosted by collaboration platform120, or view or upload content, such as games122, images, video items, web pages, documents, and so forth. In one example, the collaboration application114may be a web application (e.g., an application that operates in conjunction with a web browser) that can access, retrieve, present, or navigate content (e.g., virtual character in a virtual environment, etc.) served by a web server. In another example, collaboration application114may be a native application (e.g., a mobile application, app, or a gaming program) that is installed and executes local to client device110and allows users to interact with collaboration platform120. The collaboration application114may render, display, or present the content (e.g., a web page, a media viewer) to a user. In an implementation, the collaboration application114may also include an embedded media player (e.g., a Flash® player) that is embedded in a web page.

According to aspects of the disclosure, the collaboration application114may be a collaboration platform application for users to build, create, edit, upload content to the collaboration platform120as well as interact with collaboration platform120(e.g., play games122hosted by collaboration platform120). As such, the collaboration applications114may be provided to the client devices110A and110B by the collaboration platform120. In another example, the collaboration applications114may be applications that are downloaded from the collaboration platform120. In some implementations, collaboration application114of client device110may include game engine124. In some implementations, game engine124of client device110may be separate from collaboration application114.

In implementations, a user may login to collaboration platform120via collaboration application114. The user may access a user account by providing user account information (e.g., username and password) where the user account is associated with one or more characters available to participate in one or more games122of collaboration platform120.

In general, functions described in one implementation as being performed by the collaboration platform120can also be performed by the client devices110A through110B, computer cluster130, or servers140, in other implementations if appropriate. In addition, the functionality attributed to a particular component can be performed by different or multiple components operating together. The collaboration platform120can also be accessed as a service provided to other systems or devices through appropriate application programming interfaces (APIs), and thus is not limited to use in websites.

In implementations, collaboration platform120may include messaging module128. In implementations, messaging module128may be a system, application, or module that allows users to exchange electronic messages via a communication system, such as network105. Messaging module128may be associated with collaboration application114(e.g., module of collaboration application114or be a separate application). In implementations, users may interface with messaging module128and exchange electronic messages among users of collaboration platform120. Messaging module128may be, for example, an instant messaging application, a text messaging application, an email application, a voice messaging application, a video messaging application, or a combination thereof, among others.

In implementations, messaging module128may facilitate the exchange of electronic messages between users. For example, a user may be logged into a messaging application on client device110A, and another user may be logged into a messaging application on client device110B. The two users may start a conversation, such as an instant messaging conversation. Messaging module128may help facilitate the messaging conversation by sending and receiving the electronic messages between the users of collaboration platform120. In another implementation, two users may use respective messaging applications to participate in in-game dialog with one another where the dialog may be part of the view that includes the gameplay. One or more of collaboration platform120or collaboration application114may generate (e.g., gather, process, etc.) data146associated with messaging other users via the messaging module128and may store the data146in database142(e.g., a chat database).

In implementations, users of a collaboration platform120may be linked to other users and be recognized by the collaboration platform120as “friends.” A “friend” on the collaboration platform120may refer to a connection to another user via the collaboration platform120. “Friendship” may refer the connection between at least two users via the collaboration platform120. In some implementations, a user and their friends may have special access to one another such as the ability to see information about a friend's user account, send electronic messages to one another, trade items, as well as others. In some implementations, establishing friendships may be performed by sending a friend request to another user and having the other user accept the friend request to establish the friendship. A friend request may be a message (or alternative) sent to a user requesting that the user establish a friendship with another user on the collaboration platform120.

In some implementations, users of collaboration platform120may have base functionalities to interact with other users of collaboration platform120. By establishing a friendship with another user, a user may be allowed to access additional functionalities (also referred to as “privileges” or “additional privileges” herein) of the collaboration platform120that allow the friends greater opportunity to interact with one another. In some implementations, users of collaboration platform120that are friends may be able to use the base functionalities and the additional functionalities in response to establishing a friendship.

In some implementations, the additional functionality may include messaging functionality that allows user A (e.g., user of client device110A) and user B (e.g., user of client device110B) to communicate using a messaging service (e.g., messaging module128) via the collaboration platform120(e.g., posts, chats, personal messages, public messages, messaging in a virtual gaming environment, video, etc.). For example, the messaging functionality may allow users to transmit messages to each other. In another example, the messaging functionality allows users to transmit certain types of information (e.g., sensitive information, contact information, name, picture, voice recording, current location, unique identifiers (device ID), etc.) to each other that would otherwise be filtered by messaging module128. It may be noted that user A and user B are used for purposes of illustration, rather than limitation. For example, collaboration platform120may include one or more users capable of establishing friendships that cause collaboration platform120to grant additional functionality.

In some implementations, an additional functionality may include a sharing functionality that allows user A and user B to share items with each other via the collaboration platform120. For example, users of a gaming platform may have a sharing functionality that allows users to purchase, trade, or transfer virtual items, such as virtual currency, in a virtual gaming environment.

In some implementations, an additional functionality may include a following functionality that allows user A and user B to follow each other into a game122or into a specific gaming environment of a game122hosted by the collaboration platform120. For example, user A may be participating in a game122A. The following functionality may allow user B to navigate to a profile page of user A and select an option, “Join Game,” on user A's profile page, which places user B in the game122A (e.g., the specific gaming environment of game122A where user A is participating). In implementations, the following functionality may also include a notification feature that for example, notifies user B of the game in which user A has moved or is currently participating.

In some implementations, an additional functionality may include an invitation functionality that allows user A and user B to invite each other to a private game hosted by the collaboration platform120. For example, user A may create a private game and the invitation functionality may allow user A to invite user B to participate in the private game.

In some implementations, an additional functionality may include a group functionality that allows user A and user B to join a group hosted by the collaboration platform120. For example, the additional functionality may allow user A to invite user B to a group where user A is a group member or group creator. Members of a group may participate with each other to compete against a different group in a game122, may participate with each other in building (e.g., environments, structures, etc.) within a game122, may participate with each other to create a game122(e.g., via creator module126), etc.

In some implementations, an additional functionality includes an interaction functionality that allows user A and user B to interact with each other via the collaboration platform120. For example, the base functionality may allow a user to participate in single-player games (but not multi-player games) and the interaction functionality may allow the user to participate in multi-player games. In another example, the base functionality may allow user A to compete against other users in a game122and the interaction functionality may allow user A to work together with user B to compete against other users in a game122. It may be noted that the additional functionalities described herein are provided for purposes of illustration, rather than limitation. In other implementations, other features of collaboration platform120may be included in the additional functionalities that are provided in view of establishing a friendship.

In implementations, the computer cluster130includes a set of interconnected nodes132and134(e.g., computing devices) to perform a common task so that the computer cluster130may be viewed as a single computer system. For example, computer cluster130includes master node132and worker nodes134A-134N. Each node132and134of computer cluster130may include, but is not limited to, any data processing device, such as a desktop computer, a laptop computer, a mainframe computer, a personal digital assistant, a server computer, a hand-held device or any other device configured to process data. The nodes132and134of the computer cluster130may be connected to each other through a network, such as network105. Each node132and134may be running its own instance of an operating system.

Each node132and134of computer cluster130may have its own physical or virtual memory. Memory may include, but is not limited to, main memory such as, read-only memory (ROM), flash memory, dynamic random access memory (DRAM) (e.g., synchronous DRAM (SDRAM) or DRAM (RDRAM), etc.), and static memory (e.g., flash memory, static random access memory (SRAM), etc.). Each node of computer cluster130may have data stored on local storage (not shown), such as local storage disks.

Computer cluster130, and each node132and134of computer cluster130, can further implement various network-accessible server-based functionalities (not shown) or include other data processing equipment.

The computer cluster130may be associated with (e.g., include) a queue136. A queue136can include a data structure that stores elements (e.g., markers138that identify units of data146, markers that are units of data146). The elements in the queue136may be kept in an order and operations on the data structure may include the addition of elements to the data structure and removal of elements from the data structure. For example, the queue136may be a First-In-First-Out (FIFO) queue where the first element added to the queue will be the first element to be removed from the queue (e.g., operations on the data structure include addition of elements to the rear terminal position of the queue and removal of elements from the front terminal position of the queue). In some implementations, the queue136is hosted by the computer cluster130. In some implementations, the queue hosted external to the computer cluster130. The queue136may be stored on one or more of master node132, a worker node134, a server140, etc. In one implementation, the queue136may be stored in a memory (e.g., random access memory), a cache, a drive (e.g., a hard drive), a flash drive, a database system, or another type of component or device capable of storing data. The queue136may be stored on any number of mass storage devices, such as magnetic or optical storage based disks, solid-state drives (SSDs) or hard drives, coupled to nodes132and134. The queue136may be stored on multiple storage components (e.g., multiple drives or multiple databases) that may also span multiple computing devices (e.g., multiple server computers). In implementations, the queue136may be stored on a distributed file system. For example, the queue136may be stored on a cloud storage service. The queue136may include markers138. Each marker138may be associated with a portion (e.g., a set of rows) of data146in database142. Each marker may identify a database142, a beginning address of data (e.g., a beginning row in the database142), and an ending address of data (e.g., an ending row in the database142). For example, a marker138may be associated with rows one through 10,000 in database142A. Each marker138may identify a fixed size address range. For example, a first marker138may identify rows 1-10,000 of database142A and a second marker138may identify rows 10,001-20,000 of database142A.

In implementations, computer cluster130may copy data146associated with server140A to server140B (e.g., from databases142to databases144). In implementations, the server140A receives data146from collaboration platform120and stores the data146in databases142. Related data146may be stored in the same database142.

Computer cluster130may include a master node132and worker nodes134. Master node132may control the processing of the map-reduce job. The master node132may determine how many map operations to use, how many reduce operations to use, which processes and processing devices (e.g., nodes) to use to perform the operations, where to store the intermediate data and the output data (e.g., markers138), how to respond to processing failures, etc.

Master node132may contain map-reduce skew module170. Map-reduce skew module170operating on the master node132may receive one or more map-reduce jobs from the client devices110A-B (e.g., a client device110may submit a map-reduce job that is received by master node132). As described in more detail below, map-reduce skew module170executing on master node132may cause worker node134A to execute a map operation on identifiers of the databases142to distribute databases to worker nodes134B-C, cause worker node134B to execute a first reduce operation on database142A to generate markers138, cause worker node134C to execute a second reduce operation on database142B to generate markers138, submit markers138to queue136, and direct a worker node134to perform a copy operation based on a marker138in the queue136responsive to the worker node134completing the respective reduce operation.

In implementations, once the map-reduce job is sent to cluster130, the map-reduce skew module170executing on the master node132may monitor the progress of the map-reduce job being executed on the cluster130. In implementations, map-reduce skew module170executing on the master node132may receive a result of the map-reduce job (e.g., markers138) and may store the result in the queue136.

In implementations, each node132and134of computer cluster130may execute all or part of a map-reduce job. The execution of all or part of one or more map-reduce jobs is illustrated by map-reduce skew module170running on their respective nodes. It may be noted that a single map-reduce job may run on one or more nodes of cluster130in parallel. Each node132and134may have memory (e.g., memory hardware) and may also communicate with the queue136or local storage (not shown).

In implementations, the computer cluster130may run a map-reduce framework. Each node132and134of cluster130is configured to run a specific map-reduce framework. Different frameworks include frameworks such as, Apache™ Hadoop®, Hazelcast®, MongoDB®, Infinispan, and Apache™ Spark™.

In implementations, map-reduce skew module170receives map-reduce jobs submitted by client devices110A. Map-reduce skew module170may also receive information from computer cluster130and servers140(e.g., databases142, databases144). The information may include input data (e.g., identifiers of databases142), intermediate data (e.g., identifiers of databases142), and output data (e.g., markers138) of the map-reduce job.

It may be noted that a map-reduce job may include multiple functions. It may be noted that a map-reduce job may describe a complete execution of the functions and may include an input data, intermediate data, and output data. Alternatively, a map-reduce job may indicate one or more functions or operations in executing a map-reduce job. For example, a map-reduce job may refer to the map function or the reduce function.

For the sake of illustration, rather than limitation, map-reduce skew module170is described as implemented on master node132. In other implementations, map-reduce skew module170may in part or wholly be implemented on collaboration platform120. In other implementations, map-reduce skew module170may in part or wholly be implemented on client devices110. In other implementations, map-reduce skew module170operating on one or more of client device110, computer cluster130, or collaboration platform120may work in conjunction to perform the operations described herein. Although implementations of the disclosure are discussed in terms of collaboration platforms, implementations may also be generally applied to any type of platform generating data146. The map-reduce skew module170may help facilitate the operations described herein, such as operations described with respect toFIGS. 2-3. In some implementations, the map-reduce skew module170may be part of another application (e.g., collaboration application114), such as a plug-in. In some implementations, map-reduce skew module170may be a separate application executing on a device.

FIG. 2is a block diagram illustrating copying data146using a queue136, in accordance with some implementations. As shown inFIG. 2, a set of input data210(e.g., identifiers270of databases142) is processed by a first set of processes (e.g., mapping operations220) to distribute a set of intermediate data230(e.g., identifiers270of databases142). The intermediate data230is processed by a second set of processes (e.g., reduce operations240) to generate output data250(e.g., markers138of data146of databases142). The output data250may be placed in a queue136and then processed by a third set of processes (e.g., copy operations260) to generate a copy of data146. The master node132controls the set of processing tasks, determines how many map tasks to use, how many reduce tasks to use, which processes and processors (e.g., nodes) to use to perform the tasks, where to store the input data210, intermediate data230, output data250, and copy of data146, how to respond to any processing failures, etc. Master node132assigns tasks to processes and multiple processes may be executed by each of the nodes134(e.g., worker node134A may execute a map operation220and a reduce operation). The processes controlled by the master node132may be a subset of the full set of processes executed by the system100and the set of nodes134available to do the work assigned by the master node132may be fewer than the set of nodes134of the system100.

In some implementations, the mapping operation220is a pass-through operation and the intermediate data230is the same as the input data210. The worker node134A (via the mapping operation220) may distribute the intermediate data230to the worker nodes134for the reduce operations240. The worker node134A in mapping operation220may distribute a respective identifier270of a database142to each worker node134B-C in the reduce operations240so that each worker node134B-C executes a respective reduce operation on a respective database142. The worker nodes134B-C may execute reduce operations240to output markers138of data146of the respective database142. Each marker138may correspond to database142, a beginning address of data146, and an ending address of data146. Each marker138may identify a fixed size address range (e.g., 10,000 rows of data).

As each marker138is generated by a worker node134and may be submitted to a queue136associated with the computer cluster130. Markers138may be submitted (e.g., continuously) as the markers138are generated (e.g., not waiting for the reduce operation240to complete). In some implementations, a worker node134outputs (e.g., submits) the marker138to the queue136. In some implementations, the master node132determines that a marker138has been generated by the worker node134and the master node132submits the marker138to the queue136. Responsive to submitting a marker138to the queue136, the master node132may generate a first message indicating that the marker138is in the queue136and broadcast the first message to the worker nodes134B-C.

In one implementation, database142A is larger (e.g., has more rows of data146) than database142B. Worker node134C may complete a respective reduce operation240B that generated markers138for database142B while worker node134B is still performing a respective reduce operation240A on database142A. The master node132may receive a second message from the worker node134C indicating that the worker node134C has completed the respective reduce operation on the database142B.

The master node132may direct the worker node134C to perform a copy operation260that copies data146identified by a marker138(in queue136) to database144. In some implementations, the master node132directs the worker node134C to perform a copy operation260that copies data146B identified by a marker138B, where the worker node134C generated the marker138B. In some implementations, the master node132directs the worker node134C to perform a copy operation260that copies data146A identified by a marker138A, where the worker node134B generated the marker138A. To perform the copy operation260, the worker node134C may copy the data146identified by a marker138from a database142associated with server140A to another database144associated with server140B.

Subsequent to directing the worker node134C to perform the copy operation260, the master node132may receive a third message from the worker node134C that the worker node134C has completed copying the data146identified by the marker138(e.g., from database142associated with server140A to database144associated with server140B). In some implementations, responsive to receiving the third message from the worker node134C, the master node132may remove the marker138from the queue136. In some implementations, responsive to directing the worker node134C to perform the copy operation, the master node132may remove the marker138from the queue136. In some implementations, the master node132may monitor the copy operation260and responsive to determining the copy operation260corresponding to the marker138has completed, the master node132may remove the marker138from the queue136.

Responsive to determining that worker node134B has completed a respective reduce operation240on database142A, the master node132may direct the worker node134B to perform a copy operation260that copies data146identified by another marker138in the queue136. The worker nodes134A-N may start copy operations at different points in time (e.g., once they are available) so that the worker nodes134A-N are not idle. The master node132may distribute the markers138to the different worker nodes134A-N for execution of copy operations as each of the worker nodes134becomes available (e.g., completes a respective reduce operation240, completes a respective mapping operation220).

After all reduce operations240are completed by the worker nodes134, a final record may be sent (e.g., by the master node132) to the queue136indicating reduce operation240completion. After all copy operations260are completed, a final record may be sent (e.g., by the master node132) to the queue136indicating completion of copying data from the databases142.

FIG. 3is a flow diagram illustrating a method300for copying data using a queue, in accordance with implementations of the disclosure. Method300may be performed by processing logic that includes hardware (e.g., circuitry, dedicated logic, programmable logic, microcode), software (e.g., instructions run on a processing device to perform hardware simulation), or a combination thereof. In some implementations, map-reduce skew module170executing at master node132may perform some or all the operations. In other implementations, map-reduce skew module170executing at collaboration platform120, client device110A, client device110B, master node132, worker node134, or combination thereof may perform some or all the operations. Elements ofFIGS. 1-2may be used to help illustrated method300. It may be noted that the in some implementations, method300may include the same, different, fewer, or a greater number of operations performed in any order.

At block305of method300, processing logic performing method300causes identifiers270of databases142to be sent to a first node134A of a computer cluster130. The processing logic may receive the identifiers270of databases142that are used by a first server140A to cause a copy of the databases142for use by a second server140B. In some implementations, processing logic receives a map-reduce job from a client device110or collaboration platform120. The map-reduce job may indicate databases142to be copied. In some implementations, the processing logic generates identifiers270of databases142(e.g., one identifier270per database142). In some implementations, the processing logic receives the identifiers270of databases142(e.g., as part of the map-reduce job, in conjunction with the map-reduce job).

At block310, processing logic causes the first node134A to execute a map operation220on the identifiers270to distribute a first database142A to a second node134B and to distribute a second database142B to a third node134C. The map operation220may distribute the identifiers270to nodes134. The map operation220may distribute one identifier270per node134.

At block315, processing logic causes a second node134B of the computer cluster130to execute a first reduce operation240A on the first database142A to generate first markers138. In some implementations, the markers138identify locations of data146of a respective database (e.g., data146A of the first database142A). In some implementations, the markers138include the data146(e.g., units of data) of the respective database142. In some implementations, the identifier270includes information about the database142(e.g., indicates the amount of rows of data146in the database142). In some implementations, the worker node134uses the identifier270to identify the database142associated with the first server140A and obtains information about the database142from the database142or from the first server140A.

At block320, processing logic causes a third node134C of the computer cluster130to execute a second reduce operation240B on the second database142B to generate second markers138(e.g., that identify locations of data146of the second database142B, that are the units of data of the second database142B). In some implementations, the second database142B is smaller (e.g., has less rows) than the first database142A and the third node134C completes the second reduce operation240B before the second node134B completes the first reduce operation240A.

At block325, processing logic, responsive to generation of one or more markers138, causes a submission of the one or more markers138to a queue136associated with the computer cluster130. In some implementations, the processing logic monitors the nodes134placing the markers138in the queue136. Each marker138is submitted to the queue136as the respective marker138is generated (e.g., marker138is immediately sent to the queue once the marker138is generated even if the reduce operation240executing on the worker node134has not completed). The queue136receives markers138as the markers138are generated (e.g., without waiting for reduce operations240executing on worker nodes134to finish). It can be noted that the queue136can part of the computer cluster130on which the reduce operation240is performed, or on other hardware.

At block330, processing logic determines that the second node134B has completed the first reduce operation220A that generated the first markers138. In some implementations, processing logic monitors the reduce operations240to determine that a node134has completed a reduce operation220. In some implementations, a node134transmits a final record to the queue136responsive to finishing the respective reduce operation240.

At block335, processing logic directs the second node134B to perform a first copy operation260A that copies first data146corresponding to a marker138of the queue136. In some implementations, the second node134B generated the marker138in the reduce operation240A. In some implementations, the third node134C generated the marker138in the reduce operation240B.

At block340, processing logic identifies that the second node134B has completed copying the first data146corresponding to the marker138from a respective database142associated with the first server140A to another database144associated with the second server140B. In some implementations, the processing logic monitors the copy operations260to determine a node134has completed copying data146corresponding to a marker138. In some implementations, the node134transmits a message to the processing logic responsive to completing copying data146corresponding to a marker138. In implementations, after the second node134B completes copying the first data146corresponding to the marker138, the second node134B can return to the queue136to retrieve the next marker in the queue and perform another copy operation. The second node134B can continue copying data until no markers remain in queue136.

At block345, processing logic removes the marker138from the queue136. In some implementations, the processing logic removes the marker138from the queue136responsive to determining a node134completed copying the data146corresponding to the marker138from a database142to a database144. In some implementations, the processing logic removes the marker138from the queue136responsive receiving an indication from a node134that the node134has completed copying the data146corresponding to the marker138from the database142to the database144.

In implementations, as other nodes134of computer cluster130complete respective reduce operations, the other nodes134can be repurposed to perform copy operations using the queue136(similar as described above). The processing logic continues directing nodes134to perform copy operations to copy data146corresponding to markers138of the queue136(in the order that the queue provides) until all of the data146from the databases142has been copied to the databases144. In some implementations, copying of data146from databases142to databases144may be performed in parallel. For instance, copying of database142A and142B can be copied in parallel.

Method300describes copying data146from databases142to other databases144for purposes of illustration, rather than limitation. In other implementations, aspects of the present disclosure can be applied to efficiently copying stored data generally, such as structured data (e.g., databases) or unstructured data. For example, data can be copied from one or more data stores of a first group of data stores to one or more data stores of a second group of data stores. A data store can be a repository to persistently store collections of data, and can include databases or other types of data such as files, emails, picture, etc. An illustration is provided below.

In some implementations, at block305of method300, processing logic performing method300causes identifiers270of a first group of data stores to be sent to a first node134A of a computer cluster130. At block305processing logic can cause the first node134A to execute a map operation220on the identifiers270to distribute a first data store of a first group of data stores to a second node134B and to distribute a second data store of the first group of data stores to a third node134C. At block315, processing logic causes a second node134B of the computer cluster130to execute a first reduce operation240A on the first data store to generate first markers138. In some implementations, the markers138identify locations of data146at the data store. In some implementations, the markers138include the data146(e.g., units of data) of the data store. At block320, processing logic causes a third node134C of the computer cluster130to execute a second reduce operation240B on the second data store of the first group of data stores to generate second markers138. At block325, processing logic, responsive to generation of one or more markers138, causes a submission of the one or more markers138to a queue136associated with the computer cluster130. At block330, processing logic determines that the second node134B has completed the first reduce operation220A that generated the first markers138. At block335, processing logic directs the second node134B to perform a first copy operation260A that copies first data146corresponding to a marker138of the queue136. At block340, processing logic identifies that the second node134B has completed copying the first data146corresponding to the marker138from a respective data store of the first group of data stores to another data store of a second group of data stores. At block345, processing logic removes the marker138from the queue136.

FIG. 4is a block diagram illustrating an exemplary computer system400, in accordance with implementations. The computer system400executes one or more sets of instructions that cause the machine to perform any one or more of the methodologies discussed herein. Set of instructions, instructions, and the like may refer to instructions that, when executed computer system400, cause computer system400to perform one or more operations of map-reduce skew module170. The machine may operate in the capacity of a server or a client device in client-server network environment, or as a peer machine in a peer-to-peer (or distributed) network environment. The machine may be a personal computer (PC), a tablet PC, a set-top box (STB), a personal digital assistant (PDA), a mobile telephone, a web appliance, a server, a network router, switch or bridge, or any machine capable of executing a set of instructions (sequential or otherwise) that specify actions to be taken by that machine. Further, while only a single machine is illustrated, the term “machine” shall also be taken to include any collection of machines that individually or jointly execute the sets of instructions to perform any one or more of the methodologies discussed herein.

The computer system400may further include a network interface device422that provides communication with other machines over a network418, such as a local area network (LAN), an intranet, an extranet, or the Internet. The computer system400also may include a display device410(e.g., a liquid crystal display (LCD) or a cathode ray tube (CRT)), an alphanumeric input device412(e.g., a keyboard), a cursor control device414(e.g., a mouse), and a signal generation device420(e.g., a speaker).

The data storage device416may include a non-transitory computer-readable storage medium424on which is stored the sets of instructions of the system architecture100and map-reduce skew module170embodying any one or more of the methodologies or operations described herein. The sets of instructions of the system architecture100and map-reduce skew module170may also reside, completely or at least partially, within the main memory404and/or within the processing device402during execution thereof by the computer system400, the main memory404and the processing device402also constituting computer-readable storage media. The sets of instructions may further be transmitted or received over the network418via the network interface device422.

While the example of the computer-readable storage medium424is shown as a single medium, the term “computer-readable storage medium” can include a single medium or multiple media (e.g., a centralized or distributed database, and/or associated caches and servers) that store the sets of instructions. The term “computer-readable storage medium” can include any medium that is capable of storing, encoding or carrying a set of instructions for execution by the machine and that cause the machine to perform any one or more of the methodologies of the disclosure. The term “computer-readable storage medium” can include, but not be limited to, solid-state memories, optical media, and magnetic media.

The disclosure also relates to an apparatus for performing the operations herein. This apparatus may be specially constructed for the required purposes, or it may include a general purpose computer selectively activated or reconfigured by a computer program stored in the computer. Such a computer program may be stored in a computer readable storage medium, such as, but not limited to, any type of disk including a floppy disk, an optical disk, a compact disc read-only memory (CD-ROM), a magnetic-optical disk, a read-only memory (ROM), a random access memory (RAM), an erasable programmable read-only memory (EPROM), an electrically erasable programmable read-only memory (EEPROM), a magnetic or optical card, or any type of media suitable for storing electronic instructions.

In additional implementations, one or more processing devices for performing the operations of the above described implementations are disclosed. Additionally, in implementations of the disclosure, a non-transitory computer-readable storage medium stores instructions for performing the operations of the described implementations. Also in other implementations, systems for performing the operations of the described implementations are also disclosed.