Methods and systems for processing large graphs using density-based processes using map-reduce

• US 8,521,782 B2
• Filed: 06/15/2012
• Issued: 08/27/2013
• Est. Priority Date: 07/20/2011
• Status: Active Grant

• Current Assignee: Salesforce.com, Inc.
• Original Assignee: Salesforce.com, Inc.
• Inventors: Gong, Nan, Koister, Jari
• Primary Examiner(s): Vy, Hung T
• Assistant Examiner(s): Pham, Tuan A
• Application Number:
  US13/524,280
• Publication Number:
  US 20130024479A1
• Time in Patent Office: 438 Days
• Field of Search: None
• US Class Current: 707/791
• CPC Class Codes: G06F 9/5066 Algorithms for mapping a plurality of inter-dependent sub-tasks onto a plurality of physical CPUs mappping at compile time, see G06F8/451

Abstract

Embodiments are directed to a density-based clustering algorithm that decomposes and reformulates the DBSCAN algorithm to facilitate its performance on the Map-Reduce model. The DBSCAN algorithm is reformulated into connectivity problem using a density filter method and a partial connectivity detector. The density-based clustering algorithm uses message passing and edge adding to increase the speed of result merging, it also uses message mining techniques to further decrease the number of iterations to process the input graph. The algorithm is scalable, and can be accelerated by using more machines in a distributed computer network implementing the Map-Reduce program.

Claims

1. A computer-implemented method for processing a graph comprising graph data in a network comprising a plurality of individual processor-based machines, the method comprising:

partitioning the graph into a plurality of partitions wherein the number of the plurality of partitions is a function of the number of the plurality of machines;

assigning each partition to a respective machine of the plurality of machines by applying a splitting function to the graph data;

applying a density-based filter to each partition to produce a plurality of partitions of a filtered graph, the density-based filter applies one or more density constraints to each partition such that vertices with at least a minimum number of neighboring nodes within a distance defined by the density constraints include nodes with edges pointing to neighboring nodes and vertices that do not have the minimum number of neighboring nodes within the distance defined by the density constraints do not include nodes with edges pointing to neighboring nodes;

applying a partial connectivity detector process that detects connectivity between nodes to each of the partitions of the filtered graph to produce sub-clusters of nodes of the filtered graph based on the edges between the nodes; and

merging the sub-clusters of nodes through a message based merge process.

2. The method of claim 1 wherein the splitting function uses a method selected from one of node ID and hash function.

3. The method of claim 1 wherein the step of applying a partial connectivity detector process comprises:

identifying all nodes in each partition that are connected to one another by at least one edge as a sub-cluster;

determining if each sub-cluster contains a core node; and

determining a belonging value for each sub-cluster, wherein the belonging value identifies which cluster a respective sub-cluster will be merged into.

4. The method of claim 3 further comprising performing an edge cutting operation that comprises:

identifying multiple edges between pairs of sub-clusters; and

removing all edges exceeding a single edge between a pair of sub-clusters if multiple edges exist between the pair of sub-clusters.

5. The method of claim 3 wherein the step of merging the sub-clusters further comprises:

establishing neighbor relationships among the sub-clusters based on the belonging value for each sub-cluster; and

assigning an active or halt state to each sub-cluster.

6. The method of claim 5 further comprising applying a message sending process that comprises:

broadcasting a message when a sub-cluster is in an active state, the message comprises the belonging value of a broadcasting sub-cluster, wherein a sub-cluster with at least one core node will broadcast a message to all neighbor sub-clusters, and wherein a sub-cluster no core node is an edge node that broadcasts a message indicating a need to be merged; and

removing sub-clusters without core nodes from the graph.

7. The method of claim 6 further comprising applying a message mining process changing a belonging value of at least one sub-cluster based upon a change of belonging value of at least one other sub-cluster.

8. The method of claim 7 wherein the message mining process comprises:

defining a bipartite graph, wherein a first portion of the bipartite graph includes sub-cluster ID'"'"'s and corresponding belonging values, and a second portion of the bipartite graph includes destination ID values in messages of the sub-cluster.

9. The method of claim 6 wherein the network executes a Map-Reduce program, and wherein the message sending process is implemented as Map function of an edge cut and first merge job of the Map-Reduce program, and the merging process is implemented as a Reduce process of the edge cut and first merge job.

10. The method of claim 7 wherein the network executes a Map-Reduce program, and wherein the message sending process is implemented as Map function of merge job of the Map-Reduce program, and the message mining process is implemented as a Reduce process of the merge job.

11. The method of claim 1 wherein the network executes a Map-Reduce program, and wherein the density-based filter is implemented as a Map function of a partial density-based clustering job of the Map-Reduce program, and the partial connectivity detector process is implemented as a Reduce process of the partial density-based clustering job.

12. A system for identifying clusters in a large graph using a network of coupled machines, each machine executing a Map-Reduce process, the system comprising:

a processor; and

one or more stored sequences of instructions which, when executed by the processor, cause the processor to carry out the steps of;

a splitting function component partitioning the graph into a plurality of partitions by applying a splitting function to the graph and assigning each partition to a respective machine of the plurality of machines;

applying a density-based filter applied to each partition to produce a plurality of partitions of a filtered graph, the density-based filter applies one or more density constraints to each partition such that vertices with at least a minimum number of neighboring nodes within a distance defined by the density constraints include nodes with edges pointing to neighboring nodes and vertices that do not have the minimum number of neighboring nodes within the distance defined by the density constraints do not include nodes with edges pointing to neighboring nodes;

applying a partial connectivity detector applied to each of the partitions of the filtered graph to produce sub-clusters of nodes of the filtered graph based on the edges between the nodes; and

a merger merging the sub-clusters of nodes through a message based merge process.

13. The system of claim 12 wherein the partial connectivity detector is configured to:

identify all nodes in each partition that are connected to one another by at least one edge as a sub-cluster;

determine if each sub-cluster contains a core node; and

determine a belonging value for each sub-cluster, wherein the belonging value identifies which cluster a respective sub-cluster will be merged into.

14. The system of claim 12 further comprising an edge cutting component identifying multiple edges between pairs of sub-clusters, and removing all edges exceeding a single edge between a pair of sub-clusters if multiple edges exist between the pair of sub-clusters.

15. The system of claim 14 wherein the merger is configured to:

establish neighbor relationships among the sub-clusters based on the belonging value for each sub-cluster; and

assign an active or halt state to each sub-cluster.

16. The system of claim 15 further comprising a message sending component configured to:

broadcast a message when a sub-cluster is in an active state, the message comprises the belonging value of a broadcasting sub-cluster, wherein a sub-cluster with at least one core node will broadcast a message to all neighbor sub-clusters, and wherein a sub-cluster no core node is an edge node that broadcasts a message indicating a need to be merged; and

remove sub-clusters without core nodes from the graph.

17. The system of claim 16 further comprising a message miner configured to change a belonging value of at least one sub-cluster based upon a change of belonging value of at least one other sub-cluster.

18. A non-volatile, non-transitory machine-readable medium containing one or more sequences of instructions for controlling access to an application program on a computer network, the instructions configured to cause a processor to:

partition a graph to be processed by a network of coupled computers into a plurality of partitions wherein the number of the plurality of partitions is a function of the number of the plurality of machines;

assign each partition to a respective computer of a plurality of computers comprising the network by applying a splitting function to the graph;

apply a density-based filter to each partition to produce a plurality of partitions of a filtered graph, the density-based filter applies one or more density constraints to each partition such that vertices with at least a minimum number of neighboring nodes within a distance defined by the density constraints include nodes with edges pointing to neighboring nodes and vertices that do not have the minimum number of neighboring nodes within the distance defined by the density constraints do not include nodes with edges pointing to neighboring nodes;

apply a partial connectivity detector process that detects connectivity between nodes to each of the partitions of the filtered graph to produce sub-clusters of nodes of the filtered graph based on the edges between the nodes; and

merge the sub-clusters of nodes through a message based merge process.