WO2023168937A1 - Data processing method and apparatus, computer device, and readable medium - Google Patents

Abstract

Provided are a data processing method, a data processing apparatus, a computer device, and a readable medium. The data processing method comprises: in response to receiving a policy update request message sent by a working node and carrying network throughput information of the working node, updating global network throughput information according to the network throughput information (S21); when policy update request messages sent by all working nodes are received, determining data allocation information according to the global network throughput information (S22); and in response to determining that a preset policy update condition is satisfied according to a preset first threshold, the data allocation information, and historical data allocation information, sending a policy response message carrying data allocation information to each working node (S23).

Description

Data processing methods, devices, computer equipment and readable media

Cross-references to related applications

This application claims the priority of Chinese patent application CN202210232264.4 titled "Data processing method, device, computer equipment and readable medium" submitted on March 9, 2022, the entire content of which is incorporated into this application by reference.

Technical field

The present disclosure relates to the field of communication technology, and specifically relates to a data processing method, device, computer equipment and readable medium.

Background technique

Distributed machine learning is a typical business in the computing network scenario. Its model aggregation and update process is a representative example of data aggregation and distribution in the computing network. The transmission scheduling of communication during the aggregation and update process of distributed machine learning models is optimized. The focus of related work is a specific and effective perspective to explore the optimization of communication primitives in the computing network scenario. Distributed machine learning transmission scheduling optimization work can be carried out from the perspective of overlapping computing and communication. A priority-based transmission scheme has been proposed in related technologies, which mainly takes into account the contradiction between the gradient generation order in machine learning and the demand order of global gradients (parameters) in forward propagation, and considers giving higher priority to the front-layer model parameters. So that it can be obtained by each node earlier, thus starting forward propagation earlier. In this process, this solution also considers the application of fine-grained transmission to improve the parallelism of uplink and downlink utilization. The core idea is that smaller granularity can effectively reduce the waiting delay of the merge operation during the data aggregation process, but the specific design The above method is only limited to fixed-value segmentation of larger-granularity data blocks, and does not consider how to process data blocks with too small a granularity. Related technologies focus on the fact that the parameter sizes of many layers in deep learning neural networks are actually very small, especially CNN (Convolutional Neural Networks, convolutional neural network) networks. Considering other additional overheads in each transmission, in this case if When parameters or gradients are transmitted at layer granularity, the transmission efficiency will become very low.

Regarding the load balancing of data aggregation and distribution tasks under multi-center nodes, related technologies mainly focus on the load balancing problem of multi-servers under distributed machine learning PS architecture. The performance of each server is determined through statistical and predictive methods, and then selected Some servers are assigned a certain number of parameters with reference to their performance. Considering the randomness of dynamic changes in the network, in fact, the prediction methods are often not consistent with the actual situation, resulting in the network information that has been mastered not being used well. In addition, this solution does not discuss in depth the issues of fine-grained transmission and merging of fine-grained data during transmission.

Computing power networks usually require wide area networks to open up communication channels between cross-domain distributed computing power nodes. However, the inherent heterogeneity, scarcity and dynamic characteristics of wide area network bandwidth lead to the frequent occurrence of bottleneck links, causing the computing power network to carry The communication process of distributed services is significantly blocked and slowed down. In the related technology, a solution using multiple central nodes cannot sense the dynamically changing network status, causing the communication tasks undertaken by each central node to not match its communication capabilities, resulting in low overall communication efficiency.

Contents of the invention

The present disclosure provides a data processing method, device, computer equipment and readable medium.

In a first aspect, embodiments of the present disclosure provide a data processing method, which is applied to a scheduling node in a computing power network distributed cluster system. The method includes: responding to receiving a network message sent by a working node that carries the working node. The policy update request message of the throughput information updates the global network throughput information according to the network throughput information. The global network throughput information is used to record the network throughput between any two working nodes; after receiving all the work In the case of a policy update request message sent by a node, data allocation information is determined based on the global network throughput information, and the data allocation information is used to record the data allocation ratio of each working node; in response to the first preset The threshold, the data allocation information and the historical data allocation information determine that the preset policy update conditions are met, and a policy response message carrying the data allocation information is sent to each of the working nodes.

On the other hand, embodiments of the present disclosure also provide a data processing method applied to working nodes in a computing power network distributed cluster system. The method includes: dividing the data to be processed into multiple data shards, wherein, The data amount of each data fragment is less than or equal to the second threshold; the destination receiving node of the data fragment to be allocated is determined according to the locally stored data allocation information, and the locally stored data allocation information is the policy response sent by the scheduling node Data allocation information carried in the message; sending a second notice message and a partial data message carrying the data fragments to be allocated to each of the destination receiving nodes.

On the other hand, embodiments of the present disclosure also provide a data processing device applied to a scheduling node in a computing power network distributed cluster system, including a throughput information sensing module, a policy formulation module and a policy publishing module; the throughput information The sensing module is configured to, in response to receiving a policy update request message carrying the network throughput information of the working node sent by the working node, update the global network throughput information according to the network throughput information, the global network throughput The information is used to record the network throughput between any two working nodes; the policy formulation module is used to determine data based on the global network throughput information when receiving policy update request messages sent by all working nodes. Allocation information, the data allocation information is used to record the data allocation ratio of each of the working nodes; the policy issuance module is used to respond to a determination based on the preset first threshold, the data allocation information and historical data allocation information. If the preset policy update conditions are met, a policy response message carrying the data allocation information is sent to each working node.

On the other hand, embodiments of the present disclosure also provide a data processing device applied to working nodes in a computing power network distributed cluster system, including a data segmentation module, a destination receiving node determination module and a data distribution module. The data segmentation module The dividing module is used to divide the data to be processed into multiple data fragments, wherein the data amount of each data fragment is less than or equal to the second threshold; the destination receiving node determination module is used to determine the destination according to the locally stored data. The data allocation information determines the destination receiving node of the data fragments to be allocated, and the locally stored data allocation information is the data allocation information carried in the policy response message sent by the scheduling node; the data distribution module is used to provide each The destination receiving node sends a second notice message and a partial data message carrying the data fragments to be allocated.

In another aspect, an embodiment of the present disclosure also provides a computer device, including: one or more processors; a storage device on which one or more programs are stored; when the one or more programs are executed by the one or more When executed by multiple processors, the one or more processors implement the data processing method as described above.

In another aspect, embodiments of the present disclosure also provide a computer-readable medium on which a computer program is stored, wherein when the program is executed, the data processing method as described above is implemented.

Description of the drawings

Figure 1 is a schematic diagram of the system architecture of an embodiment of the present disclosure;

Figure 2 is a schematic flowchart of a data processing method with a scheduling node as the execution subject provided by an embodiment of the present disclosure;

Figure 3 is a schematic flowchart of determining data allocation information provided by an embodiment of the present disclosure;

Figure 4 is a schematic flowchart of a data processing method with working nodes as execution subjects provided by an embodiment of the present disclosure;

Figure 5 is a schematic diagram 1 of the process of determining the destination receiving node of the data fragments to be allocated according to an embodiment of the present disclosure;

Figure 6 is a schematic diagram 2 of the process of determining the destination receiving node of the data fragments to be allocated according to an embodiment of the present disclosure;

Figure 7 is a schematic flowchart of calculating network throughput between working nodes provided by an embodiment of the present disclosure;

Figure 8 is a schematic flowchart of data merging provided by an embodiment of the present disclosure;

Figure 9 is a schematic flowchart of a data processing method with a working node as the execution subject in a specific example provided by the embodiment of the present disclosure;

Figure 10 is a schematic structural diagram of a data processing device (scheduling node) provided by an embodiment of the present disclosure;

Figure 11 is a schematic structural diagram of a data processing device (working node) provided by an embodiment of the present disclosure;

Figure 12 is a schematic structural diagram 2 of a data processing device (working node) provided by an embodiment of the present disclosure;

Figure 13 is a schematic structural diagram three of a data processing device (working node) provided by an embodiment of the present disclosure;

Figure 14 is a schematic structural diagram 4 of a data processing device (working node) provided by an embodiment of the present disclosure.

Detailed ways

Example embodiments will be described more fully below with reference to the accompanying drawings, which may, however, be embodied in different forms and should not be construed as limited to the embodiments set forth herein. Rather, these embodiments are provided so that this disclosure will be thorough and complete, and will fully understand the scope of the disclosure to those skilled in the art.

As used herein, the term "and/or" includes any and all combinations of one or more of the associated listed items.

The terminology used herein is used to describe particular embodiments only and is not intended to limit the disclosure. As used herein, the singular forms "a," "an" and "the" are intended to include the plural forms as well, unless the context clearly indicates otherwise. It will also be understood that when the terms "comprising" and/or "made of" are used in this specification, the presence of stated features, integers, steps, operations, elements and/or components is specified but does not exclude the presence or Add one or more other features, entities, steps, operations, elements, components, and/or groups thereof.

Embodiments described herein may be described with reference to plan and/or cross-sectional illustrations, with the aid of idealized schematic illustrations of the present disclosure. Accordingly, example illustrations may be modified based on manufacturing techniques and/or tolerances. Therefore, the embodiments are not limited to those shown in the drawings but include modifications of configurations formed based on the manufacturing process. Accordingly, the regions illustrated in the figures are of a schematic nature and the shapes of the regions shown in the figures are illustrative of the specific shapes of regions of the element and are not intended to be limiting.

Unless otherwise defined, all terms (including technical and scientific terms) used herein have the same meaning as commonly understood by one of ordinary skill in the art. It will also be understood that terms such as those defined in commonly used dictionaries should be construed to have meanings consistent with their meanings in the context of the relevant art and the present disclosure, and will not be construed as having idealized or excessive formal meanings, Unless expressly so limited herein.

Embodiments of the present disclosure provide a data processing method, which is applied to a computing power network distributed cluster system. As shown in Figure 1, the system includes a scheduling node and multiple working nodes. The scheduling node is responsible for sensing network throughput information. And formulate and publish policies based on network throughput information; worker nodes are responsible for executing specific data transmission control based on policy information. In terms of system deployment, since the scheduling node needs to make decisions based on the throughput information of all other working nodes, the scheduling node is deployed separately on a network device. Worker nodes are deployed to all network devices that actually perform computing tasks. The scheduling node forms a star-shaped logical network with all working nodes, and exchanges scheduling information through the WAN to ensure the smooth progress of global decision-making and control; a logically fully connected network is formed between each working node, and data information is transmitted through the WAN to meet the computing requirements. The data requirements of the task. In the embodiment of the present disclosure, working nodes need to send calculated local data on the one hand, and receive global data sent by other working nodes on the other hand. In addition, they also need to perform data aggregation and distribution of local data of other working nodes.

The scheduling node includes three functional divisions: network throughput information perception, policy formulation and policy release. The network throughput information-aware partition is responsible for sensing network throughput information between worker nodes that perform distributed tasks. The network throughput information is determined by the worker nodes during data transmission and notified to the scheduling node in the policy update request message. The network throughput information aware partition maintains global network throughput information. The global network throughput information is used to record the network throughput between any two working nodes. When the scheduling node receives the policy update request message from the working node, the network throughput information The aware partition uses the latest network throughput information received from the worker node to update the value of the corresponding position in the global network throughput information. The policy formulation partition is the core work area of the scheduling node. Based on the network throughput information provided by the current global network throughput information, it quantitatively evaluates the communication performance of all working nodes according to the performance evaluation algorithm, and then compares the evaluation results with the previous round of evaluation results. , if the performance of the working node changes significantly, the policy reset is triggered and a new policy needs to be released. The role of the policy release partition is to distribute the performance evaluation results to all worker nodes. In distributed business, in order to ensure the synchronization of the policies of all working nodes, the working nodes need to first send a policy update request message to the scheduling node. After the scheduling node receives the policy update request messages from all working nodes, the latest policy will be written. Incoming policy response message is sent to the worker node.

The functions of the worker node mainly include policy local update and data transmission control. When the working node needs to start a new round of data sending, it needs to send a policy update request message to the scheduling node to ask whether the local sending policy needs to be updated. After receiving the policy response message from the scheduling node, the working node policy locally updates the partition and reads the policy in the policy response message, and then re-divides the size of the data blocks sent to different working nodes to minimize the transmission delay and provide future data. Prepare for transmission control. The data transmission control partition is responsible for optimizing the granularity of data fragmentation according to the current policy, assigning a sending priority to each fragment according to the internal relationship of the data, and then sending and receiving data. Therefore, create sending threads, receiving threads and service threads on the working nodes respectively to perform this part of the function. For the sending thread of the working node, the data transmission control partition is responsible for fragmenting the data and assigning priorities to each fragment, and then sends local data messages (one or more data messages) of different sizes to different working nodes according to established strategies. Fragmentation). For the receiving thread, the data transmission control partition is responsible for receiving the global data messages merged by the working nodes and performing local subsequent processing. For the service thread of the worker node, the data transmission control partition is responsible for receiving local partial data from other worker nodes and distributing global data to other worker nodes, where the global data is usually merged from the received local data.

Embodiments of the present disclosure provide a data processing method, which is applied to scheduling nodes in a computing power network distributed cluster system. As shown in Figure 2, the data processing method includes steps S21 to S23.

Step S21: In response to receiving the policy update request message sent by the working node and carrying the network throughput information of the working node, update the global network throughput information according to the network throughput information.

Global network throughput information (THRUPUT_TABLE) is used to record the network throughput between any two working nodes. The network throughput information between working nodes can reflect the communication performance and network status of the working nodes. In this step, the policy update request message sent by each working node to the scheduling node carries the network throughput information of the working node, and the scheduling node adjusts the corresponding position of the global network throughput information based on the network throughput information of each working node. renew.

Step S22: Upon receiving policy update request messages sent by all working nodes, determine data allocation information based on global network throughput information.

Data allocation information (ROPORTION_LIST) is used to record the data allocation ratio of each working node. The data allocation ratio of a working node is the proportion of the data size that the working node is responsible for aggregating and distributing to the total data size.

Step S23: In response to determining that the preset policy update conditions are met based on the preset first threshold, data allocation information and historical data allocation information, send a policy response message carrying data allocation information to each working node.

The data distribution information (ROPORTION_LIST) is the data processing strategy of the entire computing network, and the historical data distribution information (DICISION_LOG) is used to record the historical data distribution ratio of each working node. In this step, the scheduling node determines whether the preset policy update conditions are met based on the preset first threshold (DISTRIBUTION_THRESHOLD), data allocation information (ROPORTION_LIST), and historical data allocation information (DICISION_LOG). When it is determined that a policy update is required, The scheduling node publishes a new strategy to each working node, and the new strategy is the data allocation information (ROPORTION_LIST) obtained in step 22. It should be noted that when it is determined that no policy update is required, the scheduling node returns an empty policy response message to each working node.

A data processing method provided by an embodiment of the present disclosure, the method includes: in response to receiving a policy update request message sent by a working node carrying network throughput information of the working node, updating the global network throughput according to the network throughput information information; upon receiving the policy update request message sent by all working nodes, determine the data allocation information based on the global network throughput information; in response to determining that the data distribution information satisfies the requirements based on the preset first threshold, data allocation information, and historical data allocation information. Preset policy update conditions and send policy response messages carrying data allocation information to each working node. The disclosed embodiments can significantly accelerate the data aggregation and distribution process of distributed applications carried by WAN-based computing power networks, thereby effectively reducing the communication time for data aggregation and distribution and improving communication efficiency; based on network throughput sensing technology, through nodes The performance evaluation algorithm dynamically allocates data volumes of different sizes to different aggregation nodes, which can effectively avoid bottleneck links and more effectively utilize high-bandwidth resource links, thereby achieving efficient use of current WAN bandwidth resources.

In some embodiments, the global network throughput information (THRUPUT_TABLE) is an n*n matrix, and the elements a _ij of the matrix are the network throughput from node i to node j, 1≤i≤n, 1≤j≤n , n is the total number of working nodes in the distributed cluster system, and the matrix can be expressed as:

Among them, a _ij (i=j)=0 means that node i itself has no network throughput, and a _ij (i≠j) is random initially.

As shown in Figure 3, determining data allocation information based on global network throughput information includes steps S221 to S223.

Step S221: Calculate the sum R _i of the row elements in the matrix and the sum C _j of the column elements in the matrix respectively.

并计算矩阵中各列元素之和

示例性的，R ₁＝a ₁₁+a ₁₂+…+a _1n,R ₂＝a ₂₁+a ₂₂+…+a _2n；C _1＝a ₁₁+a ₂₁+…+a _n1,C ₂＝a ₁₂+a ₂₂+…+a _n2。 In this step, the sum R _i of each row element in the matrix is calculated,

and calculate the sum of the elements in each column of the matrix

For example, R ₁ =a ₁₁ +a ₁₂ +…+a _1n , R ₂ =a ₂₁ +a ₂₂ +…+a _2n ; C _{1 =} a ₁₁ +a ₂₁ +…+a _n1 ,C ₂ =a ₁₂ +a ₂₂ +…+a _n2 .

Step S222: Determine the effective throughput of each working node based on the sum of row elements _Ri and the sum C _j of column elements in the matrix.

The effective throughput of worker node i is SUM_C[i], SUM_C[i]=MIN(R _i , C _j ), where j=i. That is to say, the minimum value among R _i and C _j (i=j) is taken as the effective throughput SUM_C[i] of the working node i. For example, the effective throughput of working node 1 is MIN(R ₁ , C ₁ ), and the effective throughput of working node 2 is MIN(R ₂ , C ₂ ).

Step S223: Determine data allocation information based on the effective throughput of each working node and the minimum value of the effective throughput of each working node.

The data allocation information (ROPORTION_LIST) is a set of data allocation proportions b _i of each working node. b _i represents the proportion of the data scale that the working node i should be responsible for aggregating and distributing to the total data scale. Correspondingly, the data allocation information (ROPORTION_LIST) can Expressed as: [b ₁ b ₂ b ₃ ... b _n ].

The data allocation ratio b _i of worker node i can be calculated by the following formula (1):

Among them, MIN(SUM_C[j]) represents the minimum effective throughput of each working node, ROUND represents rounding to the nearest integer, and SUM_C[i] represents the effective throughput of each working node.

After obtaining the data allocation ratio of each working node according to formula (1), the data allocation information (ROPORTION_LIST) can be obtained.

In some embodiments, satisfying the preset policy update conditions includes: the change rate (CHANGE_RATE) of the data allocation ratio of any working node in the data allocation information (ROPORTION_LIST) is greater than the preset first threshold (DISTRIBUTION_THRESHOLD). Among them, the change rate (CHANGE_RATE) of the data allocation ratio of the working node is based on the data allocation ratio (PROPORTION_LIST[i]) of the working node in the data allocation information and the data allocation ratio (DICISION_LOG[i] of the working node in the historical data allocation information (DICISION_LOG) ]) is calculated.

That is to say, according to the traversal order, the change rate CHANGE_RATE[i] of the data allocation ratio of worker node i is compared with the set first threshold (DISTRIBUTION_THRESHOLD) for triggering policy release. As long as there is a value of CHANGE_RATE[i] greater than The first threshold triggers policy update and release.

The change rate CHANGE_RATE[i] of the data allocation ratio of worker node i can be calculated by the following formula (2):

Among them, DICISION_LOG[i] represents the historical data distribution information of working node i (that is, the historical data distribution ratio of working node i), and PROPORTION_LIST[i] represents the data distribution ratio of working node i.

In some embodiments, after determining the data allocation information according to the global network throughput information (ie, step 22), the data processing method may further include the following steps: responding to the data allocation information (ROPORTION_LIST) according to the first threshold (DISTRIBUTION_THRESHOLD), ) and historical data allocation information (DICISION_LOG) determine that the preset policy update conditions are met, and for the working nodes whose change rate (CHANGE_RATE) is greater than the first threshold, the historical data allocation information (DICISION_LOG) is updated according to the data allocation information (ROPORTION_LIST). That is to say, when it is determined that a policy update is required, the scheduling node locally updates its historical data allocation information for working nodes with large changes in communication performance.

Embodiments of the present disclosure also provide a data processing method, which is applied to working nodes in a computing power network distributed cluster system. As shown in Figure 4, the data processing method includes steps S41 to S43.

Step S41: Divide the data to be processed into multiple data fragments, where the data amount of each data fragment is less than or equal to the second threshold.

M为工作节点i(工作节点i即为进行数据切分的工作节点)待处理数据的数据总量，N为算力网络分布式集群系统中工作节点的总数量，α为系数，α＝1.2×10 ⁵。 The working node divides the data generated by the calculation based on the second threshold (SLICE_SIZE) to obtain data fragments. The second threshold (SLICE_SIZE) can be determined based on the total amount of data to be processed by the working node and the total number of working nodes. In some embodiments,

M is the total amount of data to be processed by working node i (working node i is the working node that performs data segmentation), N is the total number of working nodes in the computing power network distributed cluster system, α is the coefficient, α = 1.2 ×10 ⁵ .

In this step, different data segmentation schemes are used for different data types. In the case where the data to be processed is data with an internal structure, the internal structure in the data to be processed is divided into multiple data blocks with a data amount greater than 1/4 of the second threshold, wherein the data amount of each data block is less than or equal to 1/4 the second threshold, and according to the generation order of each data block, each data block is merged to generate data fragments. When the data to be processed is data without an internal structure, the data to be processed is divided into multiple data fragments according to the second threshold.

Step S42: Determine the destination receiving node of the data fragments to be allocated based on locally stored data allocation information. The locally stored data allocation information is the data allocation information carried in the policy response message sent by the scheduling node.

When a policy update occurs, the scheduling node publishes a new policy to each working node, that is, the current data allocation information (ROPORTION_LIST) is sent to each working node through a policy response message, and each working node stores the data allocation information (ROPORTION_LIST) locally. ). In this step, each working node determines the destination receiving node of the data fragments to be allocated based on the locally stored data allocation information (ROPORTION_LIST). The specific implementation method will be described in detail later.

Step S43: Send the second notice message and the partial data message carrying the data fragments to be allocated to each destination receiving node.

In this step, each working node sends all the data fragments that need to be allocated to each destination receiving node to achieve data distribution. For each data fragment to be allocated, the sending thread determines the destination receiving node of the data fragment. If the destination receiving node is itself, it sends the second notice message and the local data message to the service thread of the working node in sequence. If the destination If the receiving node is another working node, the second notice message and the local data message are sent to each destination receiving node in sequence.

The data processing method provided by the embodiment of the present disclosure takes into account the certain independence within the distributed application data, and performs better granular transmission by segmenting larger data blocks before data transmission, so that each optimized granularity can Data sharding performs subsequent data operations independently after being collected by each convergence node, thereby alleviating the problem of long-term blocking of all data operations and accelerating the acquisition and utilization of data by applications.

In some embodiments, after dividing the data to be processed into multiple data fragments (ie step S41), the data processing method may also include the following steps: step S42', assigning priorities to each data fragment, And generate a priority list, which includes each of the data fragments arranged from high to low.

In this step, add a priority label to each data fragment. The working node can assign priority to each data fragment according to the order in which the data fragments are generated (it can also be determined according to specific needs). The data fragment generated first The priority is the lowest, and the last data fragment generated has the highest priority. The worker node maintains a priority queue, and the data fragments marked with priority labels are pushed into the priority queue and enter the waiting state. By adding different priority labels to different data fragments, more important data can be sent first in the sending queue, further accelerating the acquisition and utilization of important data by applications.

Steps S41 to S43 and step S42' are processed by the sending thread of the working node. After all the data fragments to be distributed are distributed, the sending thread ends.

Two specific implementation methods for determining the destination receiving node of the data fragments to be distributed will be described below with reference to Figures 5 and 6 respectively.

In some embodiments, as shown in Figure 5, determining the destination receiving node of the data fragments to be allocated based on locally stored data allocation information (ie, step S42) includes steps S421 to S424.

Step S421: Calculate the sum of data allocation proportions of the working nodes based on the locally stored data allocation information.

In this step, the sending thread of the working node calculates the sum of each element in the data allocation information (ROPORTION_LIST) (that is, the data allocation ratio of each working node), which is represented by SUM_P.

Step S422: Determine whether the sum of the data allocation ratios of the working nodes is greater than or equal to the number of data shards. If so, execute step 423; otherwise, execute step S424.

In this step, determine the relationship between SUM_P and the number S of local data shards. If SUM_P ≥ S, it means there are fewer data shards, and some working nodes can be selected to distribute data to them; if SUM_P < S, it means the data shards are small. More, distribute data to all working nodes.

Step S423: Determine the working node whose data distribution ratio is greater than or equal to 1 in the data distribution information as the destination receiving node.

In the case of SUM_P≥S, according to the data allocation information (ROPORTION_LIST), determine the working nodes with data allocation ratio > 1, and use these working nodes as the destination receiving nodes.

Step S424, determine all working nodes as destination receiving nodes.

In the case of SUM_P<S, all working nodes are used as destination receiving nodes.

In some embodiments, when the sum of data allocation proportions of working nodes (SUM_P) is greater than or equal to the number of data fragments (S), the number of data fragments to be received by each destination receiving node is the first number, The first amount is the difference between the data allocation proportion of each destination receiving node (PROPORTION_LIST[i]) and the data fragmentation adjustment amount (REDUCTION). The data fragmentation adjustment amount (REDUCTION) is based on the sum of the data allocation proportions of the working nodes (SUM_P ), the number of data fragments (S) and the data distribution ratio of each destination receiving node are determined.

In some embodiments, the difference (P_S) between the sum of the data allocation proportions of each working node (SUM_P) and the number of data fragments (S) is calculated, and the difference (P_S) and the data allocation proportion of each destination receiving node are calculated. The sum of (∑ _i PROPORTION_LIST[i]) is represented by SUM_MORE_1, where each destination receiving node is a working node with a data allocation ratio greater than or equal to 1. Based on the difference (P_S), the data allocation ratio of each destination receiving node (PROPORTION_LIST[i]) and SUM_MORE_1, the data fragmentation adjustment amount (REDUCTION) of each destination receiving node is calculated respectively. The first number of data fragments to be received by each destination receiving node is the difference between the data allocation ratio of each destination receiving node (PROPORTION_LIST[i]) and the data fragmentation adjustment amount of each destination receiving node (REDUCTION[i]), That is, the first number of data fragments to be received by the destination receiving node i = PROPORTION_LIST[i]-REDUCTION[i].

Among them, the data fragmentation adjustment amount (REDUCTION[i]) of the destination receiving node i can be calculated according to the following formula (3):

REDUCTION[i]＝CEIL(P_S*PROPORTION_LIST[i]/SUM_MORE_1) (3)

In some embodiments, when the sum of data allocation proportions (SUM_P) of the working nodes is less than the number of data fragments (S), the number of data fragments to be received by each destination receiving node is the second number, and the second The quantity is determined based on the data distribution ratio of each destination receiving node. It should be noted that when all working nodes are used as destination receiving nodes and all the data fragments to be allocated cannot be fully allocated, the data allocation ratio of each working node in the data allocation information (ROPORTION_LIST) can be calculated again. Data fragments that have not yet been allocated continue to be allocated until all data fragments to be allocated are allocated.

For example, the destination receiving node lookup table LOOKUP_LIST[INDEX _i ]=i can be set, where INDEX _i∈ [ _BEGINi , _ENDi ], i∈(1,n).

BEGIN＝{0,PROPORTION_LIST[0],PROPORTION_LIST[:1],…,PROPORTION_LIST[:i-1],…,PROPORTION_LIST[:n-1]};

END＝{PROPORTION_LIST[0]-1,PROPORTION_LIST[:1]-1,…,PROPORTION_LIST[:i]-1,…,PROPORTION_LIST[:n]-1}.

Among them, PROPORTION_LIST[:j]=PROPORTION_LIST[0]+PROPORTION_LIST[1]+…+PROPORTION_LIST[j].

The sending thread uses a circular traversal method to allocate the corresponding data fragments to be allocated to the elements in LOOKUP_LIST (that is, all working nodes) in the order in which the data fragments to be allocated are generated (the number of allocated data fragments is Allocation ratio value for the data of the current working node) until all data shards to be allocated are allocated.

In some embodiments, as shown in Figure 6, determining the destination receiving node of the data fragments to be allocated based on locally stored data allocation information (ie, step S42) includes steps S421' and S422'.

Step S421', calculate the load of each working node based on the total amount of data to be processed by each working node and the locally stored data allocation information.

其中，M为工作节点i待处理数据的数据总量，PROPORTION_LIST[i]为工作节点i的数据分配比例，i∈(1,n)。 In this step, the load of worker node i can be calculated according to the following formula:

Among them, M is the total amount of data to be processed by working node i, PROPORTION_LIST[i] is the data allocation ratio of working node i, i∈(1,n).

Step S422', determine the destination receiving node of the data fragments to be allocated based on the load of each working node and the data volume of the data fragments to be allocated.

In this step, calculate the difference between the load of each working node and the data volume of the current data shards to be allocated, and determine the maximum value MAX (LOAD _i -PARAS _INDEX ) of each difference, and put the maximum value corresponding to The working node serves as the destination receiving node for the data fragments to be distributed.

The difference between the load of working node i and the current data volume of the data shards to be allocated can be calculated according to the following formula (4):

Difference _i =LOAD _i -PARAS _INDEX (4)

Among them, INDEX represents the identification of the data fragment currently to be allocated, PARAS _INDEX is the data amount of the data fragment INDEX, and LOAD _i is the load of working node i.

In this disclosed embodiment, when determining the destination receiving node of each local data fragment, the working node adopts the cycle traversal determination method guided by the idea of bandwidth maximization under minimum competition, which can ensure that the data generation speed is not much faster than the data transmission. speed, so that data can be distributed as early as possible. However, if the actual data generation speed is much greater than the data sending speed, in fact, a simple simple loop arrangement method can be used for simplicity (first determine the number of shards that each working node can be responsible for based on the total number of shards, and then use a single The sequential loop method selects each node as the destination receiving node for the current fragment. When the cumulative number of scheduled fragments of a node reaches the number of fragments it can be responsible for, the node will be skipped in the next cycle), and Target node arrangement method in sequence (first determine the number of data shards that each worker node can be responsible for based on the total number of data shards, and then allocate sufficient data shards to each worker node in the order of the worker nodes).

The disclosed embodiment proposes a data fragmentation allocation scheme based on maximizing bandwidth under minimum competition, and periodically allocates a proportional amount of data fragmentation to each working node, which can avoid early blocking of nodes with low communication capabilities under the naive round-robin arrangement method. It can also avoid the problem that the bandwidth resources of each working node cannot be utilized at the same time under the sequential arrangement method of the destination receiving node, so that the overall utilization rate of the network is higher and the transmission process can be completed faster and earlier.

For each worker node, in addition to creating a sending thread, a receiving thread and a service thread are also created. The sending thread, receiving thread and service thread can be executed in parallel.

The processing process of the receiving thread is described in detail below. As shown in Figure 7, the data processing method may also include steps S71 to S72.

Step S71: In response to receiving the first notice message and the global data message sent by other working nodes, obtain the global data carried in the global data message. The global data is the data obtained after data processing on the data fragments distributed to the other working nodes. the data obtained.

In this step, the receiving thread of the working node receives the first notice message and the global data message in sequence, records the reception time t1 of the first notice message and the reception time t2 of the global data message, and obtains the global data in the global data message. It should be noted that the working node performs aggregation and merging processing on the distributed data fragments (local data) to obtain global data, and sends the global data to other working nodes through global data messages.

Step S72: Calculate the network throughput between the working node and other working nodes based on the data volume of the global data, the receiving time of the first notice message and the receiving time of the global data message.

The receiving thread of the working node determines the network throughput between the working node i and the working node j that sends the global data message according to the following formula (5):

Among them, S _ij is the data amount of global data sent by working node j to working node i.

It should be noted that the network throughput between the same nodes needs to be accumulated and averaged.

In the embodiment of the present disclosure, the measurement of network throughput between nodes is determined by locating the start and end time of data transmission combined with the amount of data contained in the data message. Among them, the starting time of data transmission is determined by the time when the working node receiving the message receives the notice message, and the completion time of data transmission is determined by the time when the working node receiving the message receives the data message. This arrangement is for the upper layer timing operation. convenient. A feasible and better solution is to time the timing directly based on the transport layer protocol. Taking TCP (Transmission Control Protocol) as an example, the timestamp time timing can be located based on the ACK (response) confirmation that the entire data stream transmission is completed. . However, the implementation process of this solution will be more complex and difficult, involving modification of the underlying basic library, making compatibility difficult to achieve, and inconvenient for system integration.

In some embodiments, the data processing method further includes the following step: in response to receiving all data fragments distributed to this working node, processing all data fragments. In this step, the receiving thread of the working node determines whether it has received all data fragments assigned to the working node. If all data fragments are received, all data fragments are processed, that is, subsequent operations on local data are performed, and then the receiving thread End; if not all are received, execute step S71.

The following describes the processing process of the service thread in detail.

The data processing method may further include the following steps: in response to receiving the second preview message and the partial data message, obtaining the data fragments carried in the partial data message. The data fragments carried in the partial data message are partial data, which are distributed data fragments. According to the data amount S _ij ′ of the data fragmentation, the reception time t1' of the second notice message and the reception time t2' of the local data message, the network throughput between the corresponding working nodes is calculated. The specific implementation method of calculating the network throughput by the service thread of the working node is similar to the specific implementation method of calculating the network throughput by the receiving thread. The difference is that the data amount S _ij ′ of the local data is used instead of the data amount S _ij of the global data, and the second data amount S ij is used. The reception time t1' of the preview message replaces the reception time t1 of the first preview message, and the reception time t2' of the local data message is used to replace the reception time t2 of the global data message.

In some embodiments, as shown in Figure 8, the data processing method further includes steps S81 to S82.

Step S81: For the same data fragment, after receiving the data fragments sent by all working nodes, perform data merging on the data fragments to obtain global data of the data fragments.

In this step, when the service thread of the working node receives the same data fragment sent by all working nodes, it performs data aggregation and other data merging processing on the data fragment to obtain the global data of the data fragment.

After receiving the data, the service thread of the working node usually needs to perform a data merging operation to merge the same data shards from all working nodes. It should be noted that this merging operation is optional but not necessary. In order to implement AllGather( (Global collection) effect, each working node can directly distribute it to each working node in turn without any operation after receiving the local data of other working nodes, and complete the broadcast. The working node only plays the role of a central summary.

Step S82: Send the first notice message and the global data message to all working nodes. The global data message carries the global data of the data fragments.

The service thread of the working node determines whether all the data fragments responsible for this node have been processed. If all processing is completed, the service thread ends; if not all processing is completed, continue to follow steps S81 to S82 to process the data responsible for this working node. Processed in slices.

In order to clearly illustrate the technical solution of the embodiment of the present disclosure, the data processing process of the working node will be described in detail below with reference to Figure 9 and a specific example. As shown in Figure 9, the working node performs data processing process steps 1 to 10.

Step 1: After the working node is ready, fragment the data and add priority labels, and send a policy update request message to the scheduling node to request policy update.

Step 2: The working node waits to receive the policy response message from the scheduling node. When receiving a policy response message that is not empty, it performs the policy update and then goes to step 3; if the policy response message is empty, it directly executes step 3.

Step 3: The working nodes create sending threads, receiving threads and service threads respectively. The sending thread is responsible for sending local data to other nodes for collection respectively. The receiving thread is responsible for receiving global data for the next round of calculation. The service thread is responsible for receiving and merging other working nodes. The local data is sent to obtain global data, and then the global data is distributed to other worker nodes. The sending thread performs steps 4-5 (i.e., the steps in the left column of Figure 9), the receiving thread performs steps 6-7 (i.e., the steps in the middle column of Figure 9), and the service thread performs steps 8-10 (i.e., the steps in the right column of Figure 9). (Steps in the next column), the sending thread, receiving thread and service thread can execute simultaneously.

Step 4: The sending thread selects a data fragment and determines its destination receiving node, and then sends the second notice message and the partial data message (carrying the selected data fragment) to the destination receiving node.

Step 5: The sending thread determines whether all data fragments have been sent. If so, the sending thread ends, otherwise go to step 4.

Step 6: The receiving thread waits to receive the first notice message and global data message from other working nodes. If received in sequence, calculate the throughput. Among them, the throughput between the same nodes needs to be accumulated and averaged, and then go to step 7.

Step 7: The receiving thread determines whether all data fragments have been received. If all data fragments are received, subsequent operations on local data are performed, and then the receiving thread ends. Otherwise, go to step 6.

Step 8: The service thread waits to receive the second notice message and the local data message from the worker node. If received in sequence, calculate the throughput (the specific process is the same as step 6), and then go to step 9.

Step 9: The service thread determines whether it has received the current data fragments sent by all working nodes. If so, it completes the data merging of the data fragments, and then sends the first notice message and the data fragments containing the data to all other working nodes in sequence. The global data message of the merged data (that is, global data), and go to step 10; otherwise, go to step 8.

Step 10: The service thread determines whether all the data fragments responsible for this working node have been processed. If so, the service thread ends, otherwise go to step 8.

This disclosed embodiment is based on the premise that the network will not change in a short period of time, and uses the network status of the previous iteration to plan the next iteration strategy. Therefore, by default, the working node must perform the next iteration at the beginning of each round of data transmission. Inquiries for policy updates. Considering the specific usage scenario, if the network status of the environment is relatively stable, you can also perform a policy update query after a certain number of rounds.

This disclosed embodiment is based on multi-center communication task load balancing, and treats each working node as a central node by default. In fact, if the actual scenario does not require it, only some working nodes can be selected as central nodes to participate in data sharding aggregation, without Just open a service thread for the working node of the non-central node and calibrate the node at the scheduling node.

The embodiments of this disclosure solve the problem of different delays caused by sending large and small granularity data separately by simultaneously segmenting large-granularity data and merging small-granularity data, and use a transmission priority strategy to ensure that more important data is transmitted as early as possible to achieve effective calculations. Overlap with communications. In addition, the amount of data sent to each convergence node (i.e., the destination receiving node) is dynamically adjusted according to the network throughput, thereby achieving convergence tasks for real-time network throughput in more common dynamic network and multi-convergence node scenarios. Load balancing avoids the creation of bottleneck links, improves network utilization, and makes dynamic network aggregation and distribution more efficient.

Key features of the disclosed embodiments include network throughput monitoring, network information-based policy formulation optimization algorithms, optimized granular priority data transmission mechanisms, and shard allocation methods that maximize bandwidth under minimal competition. The network throughput monitoring mechanism determines the uplink and downlink throughput between nodes in the network during the data transmission process, providing a basis for policy formulation. The strategy formulation optimization algorithm evaluates the communication performance of working nodes based on network uplink and downlink throughput, and dynamically allocates the data it is responsible for to each node to achieve load balancing of aggregation and distribution tasks. Optimize the granularity priority data transmission mechanism to reduce the long blocking time in data merging at larger granularities through granularity refinement, and to avoid the delay caused by the transmission of a single too small granularity data by merging multiple data with too small granularity, and based on the transmission priority This method sends important data earlier, enabling later work to start earlier and accelerating the overall process. The fragmentation allocation method that maximizes bandwidth under minimal competition can further effectively improve the utilization of network bandwidth resources and accelerate the completion of transmission.

The embodiment of the present disclosure optimizes transmission scheduling according to the characteristics of the wide area network. Due to the scarcity of WAN bandwidth, it is necessary to optimize transmission efficiency through transmission scheduling; due to the heterogeneity of WAN bandwidth, it is feasible to avoid bottleneck links through appropriate transmission scheduling optimization; due to the bandwidth of WAN Due to the dynamic nature of the network, there is a need to perceive the network status in real time through network awareness and adjust the transmission strategy in a timely manner. The solution of the disclosed embodiment completely covers the three characteristics of the wide area network and realizes highly targeted transmission scheduling optimization. Embodiments of the present disclosure can perceive the current network status by measuring the current network, and on the premise that the network status remains basically stable in a short period of time, the current network status is used to determine the data allocation ratio of each central node, and Use the results of this decision for a period of time in the future. Therefore, using the solutions of the embodiments of the present disclosure to realize data aggregation and distribution of distributed services has significant iterative characteristics, allowing enough time for network sensing and decision-making results to be applied, so that dynamic decision-making can achieve ideal results.

Based on the same technical concept, embodiments of the present disclosure also provide a data processing device. As shown in Figure 10, the data processing device is applied to a scheduling node in a computing power network distributed cluster system and includes a throughput information sensing module 101 , policy formulation module 102 and policy release module 103.

The throughput information sensing module 101 is configured to, in response to receiving a policy update request message carrying the network throughput information of the working node sent by the working node, update the global network throughput information according to the network throughput information, the Global network throughput information records the network throughput between any two worker nodes.

The policy formulation module 102 is configured to, upon receiving policy update request messages sent by all working nodes, determine data allocation information according to the global network throughput information, and the data allocation information is used to record the data distribution of each working node. Data allocation ratio.

The policy issuing module 103 is configured to, in response to determining that the preset policy update conditions are met based on the preset first threshold, the data allocation information and historical data allocation information, send the data allocation information carrying the data to each of the working nodes. policy response message.

In some embodiments, the global network throughput information is an n*n matrix, and the element a _ij of the matrix is the network throughput from node i to node j, 1≤i≤n, 1≤j≤n, n is the total number of working nodes in the distributed cluster system; the policy formulation module 102 is used to calculate the sum of the row elements R _i in the matrix and the sum C _j of the column elements in the matrix respectively; according to the The sum of the row elements R _i and the sum of the column elements C _j in the matrix are used to determine the effective throughput of each working node; according to the effective throughput of each working node and the effective throughput of each working node Minimum value to determine data allocation information.

In some embodiments, satisfying the preset policy update condition includes: the change rate of the data allocation ratio of any one of the working nodes in the data allocation information is greater than a preset first threshold; wherein, the working node The change rate of the data allocation ratio is calculated based on the data allocation ratio of the working node in the data allocation information and the data allocation ratio of the working node in the historical data allocation information.

In some embodiments, the throughput information sensing module 101 is further configured to, after the policy formulation module 102 determines the data allocation information according to the global network throughput information, respond to the data allocation information according to the first threshold, the data allocation information and The historical data allocation information determines that the preset policy update condition is met, and for the working node whose change rate is greater than the first threshold, the historical data allocation information is updated according to the data allocation information.

Based on the same concept, embodiments of the present disclosure also provide a data processing device. As shown in Figure 11, the data processing device is applied to working nodes in a computing power network distributed cluster system and includes a data segmentation module 201, a purpose Receiving node determination module 202 and data distribution module 203.

The data segmentation module 201 is configured to segment the data to be processed into multiple data fragments, where the data amount of each data fragment is less than or equal to the second threshold.

The destination receiving node determination module 202 is configured to determine the destination receiving node of the data fragments to be distributed according to the locally stored data allocation information. The locally stored data allocation information is the data allocation information carried in the policy response message sent by the scheduling node. .

The data distribution module 203 is configured to send a second notice message and a partial data message carrying the data fragments to be allocated to each of the destination receiving nodes.

In some embodiments, the data segmentation module 201 is configured to, when the data to be processed is data with an internal structure, segment the internal structure in the data to be processed whose data amount is greater than 1/4 of the second threshold into Multiple data blocks, where the data amount of each data block is less than or equal to 1/4 of the second threshold; according to the generation order of each of the data blocks, each of the data blocks is sequentially merged to generate data fragments; or, in all If the data to be processed is data without an internal structure, the data to be processed is divided into multiple data fragments according to the second threshold.

In some embodiments, the data segmentation module 201 is configured to determine the second threshold based on the total amount of data to be processed by the working node and the total number of working nodes.

In some embodiments, as shown in Figure 12, the data processing apparatus further includes a priority module 204. The priority module 204 is used to assign priorities to each of the data fragments and generate priority pairs. The priority list includes each of the data fragments arranged from high to low.

In some embodiments, the destination receiving node determination module 202 is configured to calculate the sum of data allocation proportions of the working nodes according to the locally stored data allocation information; in response to the sum of the data allocation proportions of the working nodes being greater than or equal to the The number of data fragments, determine the working node with a data allocation ratio greater than or equal to 1 in the data allocation information as the destination receiving node; in response to the sum of the data allocation ratios of the working nodes being less than the number of data fragments, determine All the working nodes are destination receiving nodes.

In some embodiments, the data distribution module 203 is configured to, when the sum of the data allocation proportions of the working nodes is greater than or equal to the number of the data fragments, distribute the data fragments to be received by each destination receiving node. The number is a first number, and the first number is the difference between the data allocation ratio of each destination receiving node and the data fragmentation adjustment amount. The data fragmentation adjustment amount is based on the data allocation ratio of the working node. The sum, the number of the data fragments and the data distribution ratio of each destination receiving node are determined; when the sum of the data distribution ratios of the working nodes is less than the number of the data fragments, each destination receiving node waits for The number of received data fragments is a second number, and the second number is determined according to the data allocation ratio of each destination receiving node.

In some embodiments, the destination receiving node determination module 202 is configured to calculate the load of each working node according to the total amount of data to be processed by each working node and the locally stored data allocation information; The load of the node and the data volume of the data fragments to be allocated determine the destination receiving node of the data fragments to be allocated.

In some embodiments, as shown in Figure 13, the data processing apparatus further includes a network throughput calculation module 205. The network throughput calculation module 205 is configured to respond to receiving the first preview message and the global data message sent by other working nodes. , obtain the global data carried in the global data message, the global data is the data obtained after data processing of the data fragments distributed to the other working nodes; according to the data volume of the global data, the third The network throughput between the working node and the other working nodes is calculated based on the reception time of a preview message and the reception time of the global data message.

In some embodiments, as shown in Figure 14, the data processing device also includes a data processing module 206. The data processing module 206 is configured to, in response to receiving all data fragments distributed to this working node, process all the data fragments. Data is fragmented for processing.

In some embodiments, the network throughput calculation module 205 is further configured to, in response to receiving the second notice message and the partial data message, obtain the data fragments carried in the partial data messages; according to the data of the data fragments The network throughput between corresponding working nodes is calculated based on the quantity, the reception time of the second preview message and the reception time of the local data message.

In some embodiments, the data processing module 206 is also configured to, for the same data fragment, perform data merging on the data fragments to obtain the data after receiving the data fragments sent by all working nodes. Global data of the fragments; sending a first notice message and a global data message to all working nodes, where the global data message carries the global data of the data fragments.

The disclosed embodiments are suitable for data aggregation and distribution scenarios in various large-scale or large-volume distributed services carried by WAN-based computing power networks. From the perspective of comprehensive utilization of the solution, the solution of the embodiments of the present disclosure is particularly suitable for global model aggregation of distributed machine learning and federated learning. In terms of technical application scenarios, the model aggregation process and data collection process of the control module based on machine learning algorithms in smart factories, Internet of Vehicles and Cloud VR scenarios can all use the solutions of the embodiments of the present disclosure, such as the collection of large amounts of sensor data in smart factories. In summary, edge servers in the Internet of Vehicles need to obtain information from different devices on the roadside (such as traffic lights, cameras, intelligent signs, etc.) for unified scheduling and cloud nodes in Cloud VR scenarios to obtain intermediate data from each working node.

Embodiments of the present disclosure also provide a computer device. The computer device includes: one or more processors and a storage device; wherein one or more programs are stored on the storage device. When the one or more programs are used by the above-mentioned one When executed by or multiple processors, the above one or more processors implement the data processing method as provided in the foregoing embodiments.

Embodiments of the present disclosure also provide a computer-readable medium on which a computer program is stored, wherein when the computer program is executed, the data processing method as provided in the foregoing embodiments is implemented.

Those of ordinary skill in the art can understand that all or some steps in the methods disclosed above and functional modules/units in the devices can be implemented as software, firmware, hardware, and appropriate combinations thereof. In hardware implementations, the division between functional modules/units mentioned in the above description does not necessarily correspond to the division of physical components; for example, one physical component may have multiple functions, or one function or step may consist of several physical components. Components execute cooperatively. Some or all of the physical components may be implemented as software executed by a processor, such as a central processing unit, a digital signal processor, or a microprocessor, or as hardware, or as an integrated circuit, such as an application specific integrated circuit . Such software may be distributed on computer-readable media, which may include computer storage media (or non-transitory media) and communication media (or transitory media). As is known to those of ordinary skill in the art, the term computer storage media includes volatile and nonvolatile media implemented in any method or technology for storage of information such as computer readable instructions, data structures, program modules or other data. removable, removable and non-removable media. Computer storage media includes, but is not limited to, RAM, ROM, EEPROM, flash memory or other memory technology, CD-ROM, Digital Versatile Disk (DVD) or other optical disk storage, magnetic cassettes, tapes, disk storage or other magnetic storage devices, or may Any other medium used to store the desired information and that can be accessed by a computer. Additionally, it is known to those of ordinary skill in the art that communication media typically embodies computer readable instructions, data structures, program modules or other data in a modulated data signal such as a carrier wave or other transport mechanism, and may include any information delivery media .

The disclosed embodiments can significantly accelerate the data aggregation and distribution process of distributed applications carried by WAN-based computing power networks, thereby effectively reducing the communication time for data aggregation and distribution and improving communication efficiency; based on network throughput sensing technology, through nodes The performance evaluation algorithm dynamically allocates data volumes of different sizes to different aggregation nodes, which can effectively avoid bottleneck links and more effectively utilize high-bandwidth resource links, thereby achieving efficient use of current WAN bandwidth resources.

Example embodiments have been disclosed herein, and although specific terms are employed, they are used and should be interpreted in a general illustrative sense only and not for purpose of limitation. In some instances, it will be apparent to those skilled in the art that features, characteristics and/or elements described in connection with a particular embodiment may be used alone, or may be used in conjunction with other embodiments, unless expressly stated otherwise. Features and/or components used in combination. Accordingly, it will be understood by those skilled in the art that various changes in form and details may be made without departing from the scope of the present disclosure as set forth in the appended claims.

Claims (19)

A data processing method, applied to scheduling nodes in a computing power network distributed cluster system, the method includes: In response to receiving the policy update request message carrying the network throughput information of the working node sent by the working node, the global network throughput information is updated according to the network throughput information, and the global network throughput information is used to record any Network throughput between two worker nodes; When policy update request messages sent by all working nodes are received, data allocation information is determined based on the global network throughput information, and the data allocation information is used to record the data allocation ratio of each working node; and In response to determining that the preset policy update condition is met based on the preset first threshold, the data allocation information and historical data allocation information, a policy response message carrying the data allocation information is sent to each of the working nodes. The method according to claim 1, wherein the global network throughput information is an n*n matrix, and the element a _ij of the matrix is the network throughput from node i to node j, 1≤i≤n,1 ≤j≤n, n is the total number of working nodes in the distributed cluster system; Determining data allocation information based on the global network throughput information includes: Calculate the sum of the row elements R _i in the matrix and the sum C _j of the column elements in the matrix respectively; Determine the effective throughput of each working node according to the sum of the row elements R _i and the sum of the column elements C _j in the matrix; Data allocation information is determined according to the effective throughput of each working node and the minimum value of the effective throughput of each working node. The method of claim 1, wherein satisfying the preset policy update conditions includes: The rate of change of the data allocation ratio of any one of the working nodes in the data allocation information is greater than the preset first threshold; wherein, the rate of change of the data allocation ratio of the working node is based on the work node described in the data allocation information. The data distribution ratio of the node is calculated from the data distribution ratio of the working node in the historical data distribution information. The method of claim 3, wherein after determining the data allocation information according to the global network throughput information, the method further includes: In response to determining that the preset policy update condition is met based on the first threshold, the data allocation information and historical data allocation information, for the working node whose change rate is greater than the first threshold, update according to the data allocation information The historical data distribution information. A data processing method, applied to working nodes in a computing power network distributed cluster system, the method includes: Divide the data to be processed into multiple data fragments, wherein the data amount of each data fragment is less than or equal to the second threshold; Determine the destination receiving node for the data fragments to be allocated based on locally stored data allocation information, which is the data allocation information carried in the policy response message sent by the scheduling node; and Send a second notice message and a partial data message carrying the data fragments to be allocated to each of the destination receiving nodes. The method of claim 5, wherein dividing the data to be processed into multiple data fragments includes: In the case where the data to be processed is data with an internal structure, the internal structure in the data to be processed with a data amount greater than 1/4 of the second threshold is divided into multiple data blocks, wherein the data amount of each data block is less than Or equal to 1/4 of the second threshold; according to the generation order of each data block, each of the data blocks is sequentially merged to generate data fragments; or, in the case where the data to be processed is data without an internal structure , divide the data to be processed into multiple data fragments according to the second threshold. The method of claim 5, wherein the second threshold is determined based on a total amount of data to be processed by a working node and a total number of working nodes. The method of claim 5, wherein after dividing the data to be processed into multiple data fragments, it further includes: A priority is assigned to each of the data fragments, and a priority list is generated, where the priority list includes each of the data fragments arranged from high to low. The method of claim 5, wherein determining the destination receiving node of the data fragments to be allocated based on locally stored data allocation information includes: Calculate the sum of data allocation proportions of working nodes based on locally stored data allocation information; In response to the sum of the data allocation proportions of the working nodes being greater than or equal to the number of data shards, determining the working node with a data allocation proportion greater than or equal to 1 in the data allocation information as the destination receiving node; and In response to the sum of the data allocation proportions of the working nodes being less than the number of data fragments, all the working nodes are determined to be destination receiving nodes. The method of claim 9, wherein when the sum of the data allocation ratios of the working nodes is greater than or equal to the number of data fragments, the number of data fragments to be received by each destination receiving node is the first quantity, and the first quantity is the difference between the data allocation proportion of each destination receiving node and the data fragmentation adjustment amount. The data fragmentation adjustment amount is based on the sum of the data allocation proportions of the working nodes, The number of data fragments and the data distribution ratio of each destination receiving node are determined; When the sum of the data allocation proportions of the working nodes is less than the number of data fragments, the number of data fragments to be received by each destination receiving node is a second number, and the second number is based on the number of data fragments. The data distribution ratio of the destination receiving node is determined. The method of claim 5, wherein determining the destination receiving node of the data fragments to be allocated based on locally stored data allocation information includes: Calculate the load of each working node according to the total amount of data to be processed by each working node and the locally stored data allocation information; According to the load of each working node and the data volume of the data fragments to be allocated, the destination receiving node of the data fragments to be allocated is determined. The method of claim 5, further comprising: In response to receiving the first notice message and the global data message sent by other working nodes, obtain the global data carried in the global data message, where the global data is data processing of the data fragments distributed to the other working nodes. The data obtained later; Calculate the network throughput between the working node and the other working nodes according to the data amount of the global data, the receiving time of the first preview message and the receiving time of the global data message. The method of claim 12, wherein the method further includes: In response to receiving all data fragments distributed to this working node, all data fragments are processed. The method of claim 5, further comprising: In response to receiving the second preview message and the partial data message, obtain the data fragments carried in the partial data message; The network throughput between corresponding working nodes is calculated according to the data amount of the data fragment, the reception time of the second preview message and the partial data message. The method of claim 14, wherein the method further includes: For the same data fragment, after receiving the data fragments sent by all working nodes, perform data merging on the data fragments to obtain the global data of the data fragments; Send a first notice message and a global data message to all working nodes, where the global data message carries the global data of the data fragments. A data processing device applied to a scheduling node in a computing power network distributed cluster system, including a throughput information sensing module, a policy formulation module and a policy release module; The throughput information sensing module is configured to, in response to receiving a policy update request message carrying the network throughput information of the working node sent by the working node, update the global network throughput information according to the network throughput information, The global network throughput information is used to record the network throughput between any two working nodes; The policy formulation module is configured to, upon receiving policy update request messages sent by all working nodes, determine data allocation information based on the global network throughput information, and the data allocation information is used to record each of the work nodes. Data allocation ratio of nodes; The policy publishing module is configured to, in response to determining that the preset policy update condition is met based on the preset first threshold, the data allocation information and historical data allocation information, send the data carrying the data to each of the working nodes. Policy response message for allocation information. A data processing device applied to working nodes in a computing power network distributed cluster system, including a data segmentation module, a destination receiving node determination module and a data distribution module, The data segmentation module is configured to segment the data to be processed into multiple data fragments, wherein the data amount of each data fragment is less than or equal to the second threshold; The destination receiving node determination module is configured to determine the destination receiving node of the data fragments to be distributed based on locally stored data allocation information. The locally stored data allocation information is the data carried in the policy response message sent by the scheduling node. distribution information; The data distribution module is configured to send a second notice message and a partial data message carrying the data fragments to be allocated to each of the destination receiving nodes. A computer device consisting of: one or more processors; A storage device on which one or more programs are stored; When the one or more programs are executed by the one or more processors, the one or more processors are caused to implement the data processing method according to any one of claims 1-15. A computer-readable medium having a computer program stored thereon, wherein when the program is executed, the data processing method according to any one of claims 1-15 is implemented.

Images