WO2024254894A1 - Single unit-based large-scale graph data processing system - Google Patents

Abstract

The present application provides a single unit-based large-scale graph data processing system, comprising a data loading module, a data computing module, a data release module, a storage management module and a disk; the data loading module is used for acquiring sub-graphs in an active state from the disk, and transmitting the sub-graphs to the data computing module; the data computing module is used for updating the sub-graphs, and transmitting messages generated by the updating to the storage management module; the data computing module is further used for transmitting the sub-graphs to the data release module; the data release module is used for writing the sub-graphs into the disk; when the sub-graphs are written into the disk, the storage management module is used for setting the states of the sub-graphs to be convergent. Applying a sub-graph center-based computing model to a single unit system and establishing a set of unique pipeline processing architecture capable of overlapping data I/O and CPU operations reduce I/O costs of traditional vertex center computing models while increasing CPU utilization rate, and promote sequential access to the disk.

Description

A large-scale graph data processing system based on a single machine Technical Field

The present application relates to the field of data processing technology, and in particular to a large-scale graph data processing system based on a single machine.

Background Art

In recent years, graph data has become a highly valued topic in the field of data science and engineering, as it is easy to abstract entities and relationships in the real world. It has been widely used in many fields such as social network analysis, recommendation systems, financial fraud detection, and drug discovery. At the same time, graph data has high flexibility, and many problems that were originally modeled using matrices, relationships, or other data structures can also be converted to graph data processing, further highlighting the importance of graph data. With the enhancement of social media and mobile Internet applications, the scale of abstract graph data generated or collected by computer systems is growing rapidly. This growth in magnitude has posed an extremely sharp challenge to the large-scale data storage, analysis, and mining capabilities of modern computer systems.

Traditional large-scale graph computing systems use a parallel approach to data partitioning, which is to integrate multiple computer resources to complete graph computing tasks. Although these computing systems play an important role in the field of large graph processing, due to the high maintenance and construction costs, only a few companies with large-scale computer clusters can perform large-scale graph computing. In addition, distributed computing systems are usually based on an assumption that using more computing nodes will reduce computing time, but in fact this assumption is not always true. Adding computing nodes may result in greater communication costs, which cannot significantly improve system performance.

In response to the actual needs of large-scale graph analysis and resource-constrained usage scenarios, a series of large-scale graph processing systems based on single machines have been proposed. These systems use external memory as memory extension to process large graphs, and adopt a vertex-centric computing model (which limits the information in the computing process to be transmitted between nodes) to improve data locality and simplify the user's usage burden. Although the vertex-centric computing model is simple and easy to understand, it also has the problem of high communication overhead or I/O overhead.

Summary of the invention

In view of the above problems, the present application is proposed to provide a method for overcoming the above problems or at least partially solving the above problems. A large-scale graph data processing system based on a single machine to solve the above problem includes:

A large-scale graph data processing system based on a single machine, comprising a data loading module, a data calculation module, a data release module, a storage management module and a disk; the disk stores large-scale graph data consisting of a plurality of subgraphs; the storage management module stores state information corresponding to each of the subgraphs; in an initial state, the state of the subgraph is active;

The data loading module is used for acquiring the sub-graph in the active state from the disk, and transmitting the sub-graph to the data calculation module;

The data calculation module is used to update the subgraph and transmit the message generated by the update to the storage management module;

When there is a change in the updated sub-graph, the data calculation module is further used to transmit the sub-graph to the data release module;

When the sub-graph is not the last one in the current round of update, the data release module is used to write the sub-graph into the disk;

When the subgraph is written to the disk, the storage management module is used to set the state of the subgraph to converged.

Preferably, when there is no change in the updated sub-graph, the data calculation module is further used to write the sub-graph into the disk.

Preferably, when the sub-graph is the last one in the current round of updating, the data releasing module is further used to transmit the sub-graph to the data loading module.

Preferably, when the current round of updating is finished, the storage management module is further used to set the state of the subgraph that has received the message to active.

Preferably, when the current round of updates is completed and there is no message cache in the storage management module, the data calculation module is also used to aggregate all the sub-graphs to obtain updated large-scale graph data.

Preferably, the storage management module includes a message storage unit and a state management unit; the state management unit stores the state information;

The data calculation module is used to transmit the updated message to the message storage unit;

When the current round of update is finished, the state management unit is used to update the subgraph received the message Set the status to Active.

Preferably, the data calculation module includes an aggregation calculation unit;

When the current round of updates is completed and there is no message cache in the message storage unit, the aggregation calculation unit is used to aggregate all the subgraphs to obtain updated large-scale graph data.

Preferably, when the subgraph is acquired by the data loading module, the storage management module is further used to set the state of the subgraph to waiting for calculation.

Preferably, when the subgraph is transmitted to the data calculation module, the storage management module is further used to set the state of the subgraph to being calculated.

Preferably, when the sub-graph is transmitted to the data release module, the storage management module is further used to set the state of the sub-graph to being released.

This application has the following advantages:

In the embodiments of the present application, with respect to the problem of high communication overhead or I/O overhead of existing large-scale graph processing systems based on a single machine, the present application provides a solution for applying a sub-graph-centric computing model to a single machine system and establishing a pipeline processing architecture, specifically: "A large-scale graph data processing system based on a single machine, comprising a data loading module, a data computing module, a data release module, a storage management module and a disk; the disk stores large-scale graph data consisting of a number of sub-graphs; the storage management module stores status information corresponding to each of the sub-graphs; in an initial state, the sub-graph is active; The data loading module is used to obtain the subgraph with active status from the disk and transmit the subgraph to the data calculation module; the data calculation module is used to update the subgraph and transmit the message generated by the update to the storage management module; when the updated subgraph has changes, the data calculation module is also used to transmit the subgraph to the data release module; when the subgraph is not the last one in the current round of updates, the data release module is used to write the subgraph to the disk; when the subgraph is written to the disk, the storage management module is used to set the state of the subgraph to convergence. By applying the subgraph-centric computing model to a single-machine system and establishing a unique pipeline processing architecture, the architecture can overlap data I/O and CPU operations, thereby reducing the I/O cost of the traditional vertex-centric computing model while improving CPU utilization and promoting sequential access to the disk. In addition, the architecture uses shared memory data structures for message passing and efficient synchronization, which can separate computing from memory management and scheduling. out, thus providing new opportunities for optimization.

BRIEF DESCRIPTION OF THE DRAWINGS

In order to more clearly illustrate the technical solution of the present application, the drawings required for use in the description of the present application will be briefly introduced below. Obviously, the drawings described below are only some embodiments of the present application. For ordinary technicians in this field, other drawings can be obtained based on these drawings without paying any creative work.

FIG1 is a schematic diagram of the connected component calculation process on the vertex center model and the subgraph center model;

FIG2 is a schematic diagram of a processing architecture of a large-scale graph data processing system provided by an embodiment of the present application;

FIG3 is a schematic diagram of a state management and optimization strategy for a large-scale graph data processing system provided in an embodiment of the present application.

DETAILED DESCRIPTION

In order to make the objects, features and advantages of the present application more obvious and understandable, the present application is further described in detail below in conjunction with the accompanying drawings and specific implementation methods. Obviously, the described embodiments are part of the embodiments of the present application, rather than all of the embodiments. Based on the embodiments in the present application, all other embodiments obtained by ordinary technicians in the field without creative work are within the scope of protection of the present application.

By analyzing the prior art, the inventors found that the vertex-centric computing model will undoubtedly bring additional communication overhead or I/O overhead due to the transmission of messages between vertices. As shown in Figure 1, when calculating the connected components on the input graph G, the subgraph-centric computing model (which allows information in the calculation process to be freely transmitted within the subgraph) has significantly fewer computing steps than the vertex-centric computing model.

However, the traditional subgraph-centric computing model is designed for multi-machine systems, and no work has yet introduced the subgraph-centric computing model into a single-machine environment. Therefore, some issues remain unclear. For example, when the communication cost is potentially converted into I/O cost, can the introduction of the subgraph-centric computing model systematically reduce the I/O cost of the out-of-core graph system and improve multi-core parallelism? The traditional subgraph-centric computing model requires a finer-grained partitioning of the graph to improve parallelism, but this The cost is more redundant control information, such as the mapping of global vertex IDs to local vertex IDs. This problem can be solved in a distributed environment by allocating enough memory to each computing node, but in a single-machine multi-core environment, finer-grained graph partitioning will occupy the already precious memory resources.

The inventors believe that extending the subgraph-centric computing model to a stand-alone system will face the following improvement requirements: when the input graph exceeds the memory capacity, the stand-alone system needs to use auxiliary storage (such as hard disk, SSD, etc.) as memory extension for calculation, so it is necessary to reasonably manage the scheduling of the graph between memory and disk; the traditional subgraph-centric computing model transmits messages between computing units through a computer network for synchronization, but in the case of shared memory, the synchronization logic of the stand-alone system has changed, so it is necessary to implement message synchronization more effectively; the subgraph-centric computing model only utilizes data partition parallelism, which may lead to insufficient CPU core utilization or excessive graph fragmentation when the memory capacity is limited, so it is necessary to balance the parallel computing between subgraphs and the parallel computing within the subgraph; due to the shared memory architecture of the stand-alone system, the cost of work migration between the cores of a single machine is very low, so flexible resource scheduling is needed to improve system performance.

In this embodiment, a large-scale graph data processing system based on a single machine is provided, comprising a data loading module, a data calculation module, a data release module, a storage management module and a disk; the disk stores large-scale graph data consisting of a plurality of subgraphs; the storage management module stores state information corresponding to each of the subgraphs; in an initial state, the state of the subgraph is active;

The data loading module is used for acquiring the sub-graph in the active state from the disk, and transmitting the sub-graph to the data calculation module;

The data calculation module is used to update the subgraph and transmit the message generated by the update to the storage management module;

When there is a change in the updated sub-graph, the data calculation module is further used to transmit the sub-graph to the data release module;

When the sub-graph is not the last one in the current round of update, the data release module is used to write the sub-graph into the disk;

When the subgraph is written to the disk, the storage management module is used to set the state of the subgraph to converged.

In the embodiment of the present application, compared with the existing large-scale graph processing system based on a single machine, In order to solve the problem of high pin or I/O overhead, the present application applies the subgraph-centric computing model to a single-machine system and establishes a pipeline processing architecture. Referring to Figure 2, given a large graph G (large graph G is initially stored on disk), the pipeline processing architecture uses the subgraph {F0, F1, F2, F3, ..., Fn-1} of graph G as the minimum input and output unit, and updates the large graph G iteratively with a pipeline. Specifically, the architecture decomposes the out-of-core processing of the subgraph Fi into three consecutive stages: reading Fi into memory, calculating and updating Fi, and if necessary, writing the updated Fi back to external memory. These stages are completed through three modules, namely, a data loading module, a data calculation module, and a data release module. In the pipeline processing architecture, these modules work asynchronously through two task queues "input queue" and "output queue".

The pipeline processing architecture effectively overlaps subgraph I/O and CPU operations, performs calculations on memory subgraphs while loading suspended subgraphs from disk, which can reduce the I/O cost of traditional vertex-centric computing models, while improving CPU utilization by reducing idle waits and enabling continuous access to disks; in addition, the architecture uses shared memory data structures for message passing and efficient synchronization, which can separate computing from memory management and scheduling, thereby providing new opportunities for optimization.

Next, a large-scale graph data processing system based on a single machine in this exemplary embodiment will be further described.

In this embodiment, the system adopts APIs based on a hybrid computing model. The APIs adopt a unified PIE+ interface, which integrates vertex-centric and subgraph-centric programming models. Users can not only parallelize sequential graph algorithms under the subgraph-centric computing model to simplify parallel programming (inter-subgraph parallelism), but also further explore the parallelism within the subgraph based on the vertex-centric computing model through new interfaces. It should be noted that the hybrid model supports both inter-subgraph parallelism of the "subgraph-centric computing model" and intra-subgraph parallelism of the "vertex-centric computing model". Under limited memory, it can better utilize multi-core resources and avoid fragmentation of the input graph; in addition, it provides a unified interface from which users can choose the interface that best suits their applications and graphics.

In this embodiment, the system further includes a scheduler. The scheduler is used to track and allocate threads in a thread pool, where each thread corresponds to a physical CPU core. It decides to allocate physical threads to virtual worker threads to perform (parallel) calculations on subgraphs. It also actively adjusts to support two levels of parallelism: when threads are available, the scheduler allocates them to new computing units by consuming the "input queue" to speed up the parallelism between subgraphs, or through the running worker cores. To improve the parallelism within a subgraph.

In this embodiment, the storage management module includes a message storage unit; the data calculation module is used to transmit the updated messages to the message storage unit. The message storage unit is used to realize message synchronization between parallel computing units. Specifically, the message storage unit is implemented as a data structure in memory. In order to improve space efficiency, it can be implemented as a compact variable-length array. It should be noted that the space complexity of the message storage unit is closely related to the partitioning strategy. The more boundary vertices/edges there are, the larger the space consumed by the message storage unit. Compared with the message passing strategy of a multi-machine system, the use of a message storage unit is more efficient in a shared memory environment.

In this embodiment, the storage management module further includes a state management unit; the state management unit stores the state information and can update the state information at a specific time. The state management unit is used to maintain a state machine to model the state of the subgraph. Specifically, the state management unit is implemented as a lightweight data structure, and each subgraph only maintains a few states, and the memory space occupied can be ignored.

Using the state management unit to record the state information of the subgraph is a low-cost convergence detection method. It only requires a tag list M to help track the message exchange between computing units, and a lightweight state machine to model the work progress of each computing unit. Specifically, the state management unit constructs a tag list M, and each subgraph corresponds to a tag to indicate whether it has received any messages in this round of iteration. If a subgraph has at least one pending update to be extracted from the message storage unit, its corresponding M[i] is true, otherwise, M[i] is false. In actual operation, a finite state machine can be used to model the progress of each subgraph, and the flag M[i] can be used to trigger the state transition of the subgraph.

As shown in Figure 3, the states of the subgraphs include "active", "waiting for calculation", "calculating", "releasing" and "converging". At any time, the subgraph is in one of the five states, where the first two states indicate that the subgraph is on the disk, and the remaining states indicate that the subgraph is in the memory. The initial state of each subgraph is "active", which means that the subgraph is waiting for the data loading module to load it into the memory; when the subgraph is obtained by the data loading module, the state management unit is used to set the state of the subgraph to "waiting for calculation", which means It means that the subgraph has been resident in the memory and the subgraph is waiting to be assigned a processing core; when the subgraph is transmitted to the data computing module, the state management unit is used to set the state of the subgraph to "computing", which means that the subgraph is being processed by the processing core; when the subgraph is transmitted to the data release module (that is, the subgraph generates a message that needs to be sent to other subgraphs in the current round of update), the state management unit is used to set the state of the subgraph to "releasing"; when the subgraph is written to the disk, the state management unit is used to set the state of the subgraph to "converging"; when the current round of update ends, the state management unit is also used to set the state of the subgraph participating in the next round of update to active, so that this part of the subgraph starts the next round of update; if and only if the current round of update ends and there is no message cache in the message storage unit, the entire system stops updating.

Under certain conditions, the system can skip certain states in a round of computation without affecting correctness, that is, it can take some "shortcuts" in state transitions and reduce unnecessary computation and I/O.

As shown in FIG3 , in this embodiment, when the current round of updates is completed, the storage management module is used to set the state of the subgraph that received the message to "active" ("shortcut A"). In order to start a new round of incremental calculations, the state of the subgraph with a "convergence" state needs to be reset to "active". If M[i] corresponding to a subgraph is true at this time, the subgraph can be kept in the "convergence" state so that it does not participate in the next round of updates. This allows the processing of subgraphs that do not need to be updated to be skipped without affecting the correctness of the program. "Shortcut A" is often used when the input graph is not well connected and a subgraph is "isolated", and effectively reduces I/O costs.

In this embodiment, when the subgraph is the last one in the current round of updates, the data release module is used to transfer the subgraph to the data loading module ("shortcut B"). After all the subgraphs have completed the current round of updates, a new round of updates will begin. If a subgraph is still in the "releasing" state, that is, it has not been completely saved to the disk, the state of the subgraph can be directly set to "waiting for calculation", thereby starting a new round of updates to the subgraph without going through the disk. "Shortcut B" can be used at the end of each round and effectively reduces I/O costs.

In this embodiment, when there is no change in the updated sub-graph, the data calculation module is used to write the sub-graph to the disk ("shortcut C"). There is no change compared to before calculation. You can skip the "releasing" state and directly set it to "converging", which can effectively reduce redundant disk writes.

In this embodiment, the data calculation module includes an aggregation calculation unit. When the current round of update is completed and there is no message cache in the message storage unit, the aggregation calculation unit is used to call a preset aggregation function to aggregate all the sub-graphs to obtain updated large-scale graph data.

Although the preferred embodiments of the present application have been described, those skilled in the art may make additional changes and modifications to these embodiments once they have learned the basic creative concept. Therefore, the appended claims are intended to be interpreted as including the preferred embodiments and all changes and modifications that fall within the scope of the embodiments of the present application.

Finally, it should be noted that, in this article, relational terms such as first and second, etc. are only used to distinguish one entity or operation from another entity or operation, and do not necessarily require or imply any such actual relationship or order between these entities or operations. Moreover, the terms "include", "comprise" or any other variants thereof are intended to cover non-exclusive inclusion, so that a process, method, article or terminal device including a series of elements includes not only those elements, but also other elements not explicitly listed, or also includes elements inherent to such process, method, article or terminal device. In the absence of further restrictions, the elements defined by the sentence "comprise a ..." do not exclude the existence of other identical elements in the process, method, article or terminal device including the elements.

The above is a detailed introduction to a large-scale graph data processing system based on a single machine provided by the present application. This article uses specific examples to illustrate the principles and implementation methods of the present application. The description of the above embodiments is only used to help understand the method of the present application and its core idea; at the same time, for general technical personnel in this field, according to the idea of the present application, there will be changes in the specific implementation method and application scope. In summary, the content of this specification should not be understood as a limitation on the present application.

Claims (10)

A large-scale graph data processing system based on a single machine, characterized by comprising a data loading module, a data calculation module, a data release module, a storage management module and a disk; the disk stores large-scale graph data consisting of a plurality of subgraphs; the storage management module stores state information corresponding to each of the subgraphs; in an initial state, the state of the subgraph is active; The data loading module is used for acquiring the sub-graph in the active state from the disk, and transmitting the sub-graph to the data calculation module; The data calculation module is used to update the subgraph and transmit the message generated by the update to the storage management module; When there is a change in the updated sub-graph, the data calculation module is further used to transmit the sub-graph to the data release module; When the sub-graph is not the last one in the current round of update, the data release module is used to write the sub-graph into the disk; When the subgraph is written to the disk, the storage management module is used to set the state of the subgraph to converged. The system according to claim 1 is characterized in that when there is no change in the updated sub-graph, the data calculation module is also used to write the sub-graph to the disk. The system according to claim 1 is characterized in that when the sub-graph is the last one in the current round of updates, the data release module is also used to transfer the sub-graph to the data loading module. The system according to claim 1 is characterized in that when the current round of update is completed, the storage management module is also used to set the state of the subgraph that received the message to active. The system according to claim 1 is characterized in that when the current round of updates is completed and there is no message cache in the storage management module, the data calculation module is also used to aggregate all the sub-graphs to obtain updated large-scale graph data. The system according to claim 1, characterized in that the storage management module comprises a message storage unit and a state management unit; the state management unit stores the state information; The data calculation module is used to transmit the updated message to the message storage unit; When the current round of update is finished, the state management unit is used to update the subgraph received the message Set the status to Active. The system according to claim 6, characterized in that the data calculation module includes an aggregation calculation unit; When the current round of updates is completed and there is no message cache in the message storage unit, the aggregation calculation unit is used to aggregate all the subgraphs to obtain updated large-scale graph data. The system according to claim 1 is characterized in that when the subgraph is acquired by the data loading module, the storage management module is also used to set the state of the subgraph to waiting for calculation. The system according to claim 1 is characterized in that when the subgraph is transmitted to the data computing module, the storage management module is also used to set the state of the subgraph to being calculated. The system according to claim 1 is characterized in that when the sub-graph is transmitted to the data release module, the storage management module is also used to set the state of the sub-graph to being released.