WO2025161762A1 - Data processing method and apparatus, and computing cluster - Google Patents

Abstract

The present application relates to the technical field of computers, and provides a data processing method and apparatus, and a computing cluster, capable of reducing data transmission time delay and improving data processing efficiency. In the computing cluster, a plurality of first computing nodes are used for executing N first tasks for processing data to be processed, wherein M first results are generated upon execution of each first task and are respectively identified as A0 to AM-1; each data processing unit among a plurality of first data processing units is used for executing M second tasks, wherein each second task among the M second tasks is used for processing a first result having the same identifier to generate a second result, and M second results are generated upon execution of the M second tasks and are respectively identified as B0 to BM-1; a plurality of second computing nodes are used for executing M third tasks, wherein each third task is used for processing a plurality of second results having the same identifier to generate a third result, and M third results are generated upon execution of the M third tasks.

Description

Data processing method, device and computing cluster

This application claims priority to the Chinese patent application filed with the State Intellectual Property Office on February 1, 2024, with application number 202410147754.3 and application name “A Data Processing Method, Device and Computing Cluster”, the entire contents of which are incorporated by reference into this application.

Technical Field

The present application relates to the field of computer technology, and in particular to a data processing method, device, and computing cluster.

Background Art

With the rapid development and widespread adoption of data analysis technology, "big data" has become a household buzzword, and how to process big data has attracted considerable attention. Typically, large-scale data can be processed in parallel using programming models.

For example, a commonly used big data programming model is the Map-Reduce model. The data processing process using this model consists of two phases: the Map and Reduce phases, each of which executes a number of operators. In the Map phase, complex tasks are broken down into several simple subtasks, reducing the scale of data processing, localizing data processing, and enabling parallel processing. In the Reduce phase, the results generated in the Map phase are aggregated.

The subtasks divided in the Map phase are executed by multiple compute nodes, and a single compute node typically performs multiple subtasks. When the Reduce phase further processes the data generated in the Map phase, it needs to obtain (or pull) the data processing results of each subtask. Because there are many subtasks in the Map phase, multiple data pulls are required during the Reduce phase, resulting in high data transmission latency during the data processing process.

Summary of the Invention

The present application provides a data processing method, device, and computing cluster, which can reduce data transmission delay during data processing and improve data processing efficiency.

This application adopts the following technical solutions:

In a first aspect, the present application provides a computing cluster, comprising: a plurality of first computing nodes, a plurality of first data processing units, each first computing node being connected to a first data processing unit, and a plurality of second computing nodes. The plurality of first computing nodes are configured to execute N first tasks, the N first tasks being configured to process N data shards of to-be-processed data, and generating M first results after executing each first task, each of which is identified as A0 to AM-1; each first data processing unit being configured to execute M second tasks, each of the M second tasks being configured to process a first result having the same identifier in the first computing node to which the first data processing unit is connected, generating a second result after processing, and generating M second results after executing the M second tasks, the M second results being identified as B0 to BM-1; and a plurality of second computing nodes being configured to execute M third tasks, each of the third tasks being configured to process a second result having the same identifier in the plurality of first data processing units, generating a third result, and generating M third results after executing the M third tasks.

In the computing cluster of the present application, since the first data processing unit connected to each first computing node can first execute the second task on the first result generated after the first computing node executes multiple first tasks to generate the second result, and then the second computing node executes the third task to process the second results generated by multiple computing nodes, it can avoid the data generated after the same computing node executes multiple first tasks being directly pulled by the second computing node multiple times and in large quantities, thereby solving the problem of large data transmission delay caused by multiple and massive data pulling, and can reduce the data transmission delay in the data processing process and improve the efficiency of data processing.

In a possible implementation, the first task is a map task, and the second and third tasks are reduce tasks.

In a possible implementation, a third computing node among the multiple first computing nodes is further configured to send a first request to the connected first data processing unit to instruct the first data processing unit to execute M second tasks.

In a possible implementation, the first request carries the operator identifier of the operator to be called during the execution of the second task and the storage location information of the M first results.

In one possible implementation, the computing cluster provided in this application further includes: a plurality of second data processing units, each second computing node being connected to a second data processing unit. A fourth computing node among the plurality of second computing nodes is further configured to send a second request to the connected second data processing unit to instruct the second data processing unit to execute M third tasks. In this application, offloading the M third tasks executed by the second computing node to the second data processing unit for execution can reduce the load on the second computing node, thereby saving resources of the second computing node.

In a possible implementation, the second request carries the operator identifier of the operator to be called during the execution of the third task and the storage location information of the M second results.

In one possible implementation, the third data processing unit among the multiple first data processing units is also used to send a notification to the fifth computing node among the multiple second computing nodes to indicate the completion of execution of M second tasks, thereby triggering the second computing node to execute M third tasks.

In one possible implementation, the fourth data processing unit among the multiple first data processing units is also used to receive a third request to instruct the update of the operator to be called by the fourth data processing unit, and load the operator logic microcode of the operator to be updated to the fourth data processing unit. Due to the limited hardware physical resources of the data processing unit (such as DPU), it is often only possible to solidify a small number of high-frequency used hardware operators. In an embodiment of the present application, the logic circuit of the data processing unit supports hardware operator updates. Therefore, the operators in the data processing unit can be flexibly updated, expanding the range of operators that can be executed by the data processing unit.

In a possible implementation, the third request carries the operator logic microcode of the operator to be updated.

In one possible implementation, the operator to be updated is a custom operator. Based on actual data processing requirements, the user designs and generates the operator logic microcode, then updates the custom operator to the data processing unit, thereby persisting the operator logic microcode within the data processing unit. In this application, supporting user-defined operators can meet the needs of diverse users and expand the application scope of computing clusters.

In a second aspect, the present application provides a data processing method, which is applied to a computing cluster, and is applied to a computing cluster, wherein the computing cluster includes multiple first computing nodes, multiple first data processing units, and multiple second computing nodes, wherein each first computing node is connected to a first data processing unit; the method includes: multiple first computing nodes execute N first tasks, and generate M first results after executing each first task, and are respectively identified as A0~AM-1; wherein the N first tasks are used to process N data shards of the data to be processed; and each first data processing unit executes M second tasks, and generates M second results after executing the M second tasks, and the M second results are identified as B0~BM-1; wherein each second task in the M second tasks is used to process the first result with the same identifier in the first computing node connected to the first data processing unit, and generate the second result after processing; then the multiple second computing nodes execute M third tasks, and generate M third results after executing the M third tasks; each third task is used to process the second results with the same identifier in the multiple first data processing units to generate the third result.

In a possible implementation, the first task is a map task, and the second and third tasks are reduce tasks.

In one possible implementation, the data processing method provided in the present application further includes: a third computing node among the multiple first computing nodes sends a first request to the connected first data processing unit to instruct the first data processing unit to execute M second tasks.

In a possible implementation, the first request carries the operator identifier of the operator to be called during the execution of the second task and the storage location information of the M first results.

In one possible implementation, the computing cluster also includes multiple second data processing units, and each second computing node is connected to a second data processing unit; based on this, the data processing method provided by the present application also includes: a fourth computing node among the multiple second computing nodes sends a second request to the connected second data processing unit to instruct the second data processing unit to execute M third tasks; wherein the second request carries the operator identifier of the operator to be called during the execution of the third task and the storage location information of the M second results.

In one possible implementation, the data processing method provided in the present application further includes: a third data processing unit among the multiple first data processing units sends a notification to a fifth computing node among the multiple second computing nodes to indicate that the execution of the M second tasks is completed.

In one possible implementation, the data processing method provided by the present application also includes: a fourth data processing unit among multiple first data processing units receives a third request to instruct the update of the operator to be called by the fourth data processing unit, and the third request carries the operator logic microcode of the operator to be updated; and the fourth data processing unit loads the operator logic microcode of the operator to be updated.

In a possible implementation, the operator to be updated is a custom operator.

In a third aspect, the present application provides a computing node comprising a memory and at least one processor connected to the memory, the memory being used to store computer program code, the computer program code comprising computer instructions, which, when executed by at least one processor, causes the computing node to perform the actions performed by the first computing node in the first aspect or the second aspect; or, to perform the actions performed by the second computing node in the first aspect or the second aspect.

In a fourth aspect, the present application provides a data processing unit comprising a memory and at least one processor connected to the memory, the memory being used to store computer program code, the computer program code comprising computer instructions, which, when executed by at least one processor, causes the data processing unit to perform the actions performed by the first data processing unit in the first aspect or the second aspect; or, to perform the actions performed by the second data processing unit in the first aspect or the second aspect.

In a fifth aspect, the present application provides a computer-readable storage medium storing computer instructions. When the computer instructions are run on a computer, the method of the second aspect and any one of its possible implementations is executed.

In a sixth aspect, the present application provides a computer program product, which includes computer instructions. When the computer instructions are run on a computer, the method of the second aspect and any one of its possible implementations is executed.

In the seventh aspect, the present application provides a chip system, comprising: a processor for calling and running a computer program from a memory, so that a device equipped with the chip system executes the method of the second aspect and any one of its possible implementation methods.

It should be understood that the beneficial effects achieved by the technical solutions of the second to seventh aspects of this application and the corresponding possible implementation methods can be referred to the technical effects of the first aspect and its corresponding possible implementation methods mentioned above, and will not be repeated here.

BRIEF DESCRIPTION OF THE DRAWINGS

FIG1 is a diagram showing a data processing framework based on a Map-Reduce model according to an embodiment of the present application;

FIG2 is a second framework diagram of data processing based on the Map-Reduce model provided in an embodiment of the present application;

FIG3 is one of the schematic diagrams of the architecture of a computing cluster provided in an embodiment of the present application;

FIG4 is a second schematic diagram of the architecture of a computing cluster provided in an embodiment of the present application;

FIG5 is a third schematic diagram of the architecture of a computing cluster provided in an embodiment of the present application;

FIG6 is a fourth schematic diagram of the architecture of a computing cluster provided in an embodiment of the present application;

FIG7 is a flowchart of a data processing method according to an embodiment of the present application;

FIG8 is a second flow chart of a data processing method provided in an embodiment of the present application;

FIG9 is a schematic diagram of an operator updating method provided in an embodiment of the present application.

DETAILED DESCRIPTION

The term "and/or" in this article is merely a description of the association relationship between associated objects, indicating that three relationships may exist. For example, A and/or B can mean: A exists alone, A and B exist at the same time, and B exists alone.

In the description and claims of the embodiments of this application, the terms "first" and "second" are used to distinguish different objects, rather than to describe a specific order of objects. For example, "first task" and "second task" are used to distinguish different tasks, rather than to describe a specific order of tasks; "first request" and "second request" are used to distinguish different requests, rather than to describe a specific order of requests.

In the embodiments of this application, words such as "exemplary" or "for example" are used to indicate examples, illustrations, or descriptions. Any embodiment or design described as "exemplary" or "for example" in the embodiments of this application should not be interpreted as being preferred or advantageous over other embodiments or designs. Rather, the use of words such as "exemplary" or "for example" is intended to present the relevant concepts in a concrete manner.

In the description of the embodiments of the present application, unless otherwise specified, “multiple” means two or more, and “multiple” can also be described as “at least two”.

The data processing method, apparatus, and computing cluster provided in the embodiments of the present application are primarily used in distributed big data processing scenarios, and the programming model for big data processing should meet at least one of the following conditions:

1) The programming model includes different processing stages. Different processing stages have data dependencies (for example, the subsequent processing stage depends on the processing results of the previous processing stage). Different processing stages include multiple tasks distributed on different computing nodes (that is, tasks executed by different computers), and multiple tasks are executed on one computing node.

2) The subsequent processing stage needs to obtain massive data fragments from the processing results generated by the previous processing stage.

It should be noted that the programming model applicable to the embodiments of the present application can be any programming model that meets the above conditions, including but not limited to the Map-Reduce model for data processing. In the embodiments of the present application, the relevant content of the data processing method is mainly described using the Map-Reduce model as an example.

First, a brief introduction to the Map-Reduce model is given.

The Map-Reduce model is a big data programming paradigm, which divides a data processing task (hereinafter referred to as a task) into multiple Map-Reduces. A Map-Reduce includes a Map phase and a Reduce phase. The Map phase and the Reduce phase in each Map-Reduce execute the operators corresponding to each phase. The types of operators executed in the Map phase and the Reduce phase are related to actual business needs and are not limited in the embodiments of this application.

The Map-Reduce model is used for big data processing in a computing cluster. The computing cluster includes computing nodes that execute tasks, and may also include management nodes for task scheduling and resource management. Taking a distributed computing cluster as an example, the resource manager pools the computing resources (processing cores) and memory resources of all computing nodes in the computing cluster. For a data processing task to be executed, the task scheduler divides the input data (also called raw data) into multiple data shards according to a certain granularity (such as 64MB). Then, it requests sufficient resources from the resource pool for the data processing task according to a certain container specification (4 cores + 12GB) to execute the task.

Specifically, referring to the framework diagram for data processing based on the Map-Reduce model shown in Figure 1, in the Map phase, based on the number of data shards after the input data is divided (for example, N data shards, where N is an integer greater than 1), N map tasks (also called map tasks) are generated for the data processing task. The N map tasks can be homogeneous (that is, the N map tasks execute the same operator or function). Each of the N map tasks performs operator calculations, and each map task generates M data shards. In the Reduce phase, M reduce tasks (also called reduce tasks) are generated for the data processing task. Among them, for each reduce task, it is necessary to pull the data shard with a certain index corresponding to the reduce task from the M data shards generated by all map tasks, and then execute the operator of the Reduce phase. For example, as shown in Figure 1, reduce task 1 needs to pull data shard 1_1 from the M data shards generated by map task 1, and data shard 2_1 from the M data shards generated by map task 2, and so on, until data shard M_1 from the M data shards generated by map task N is pulled. It can be seen that for a reduce task, N data shards need to be pulled.

Optionally, the data generated by each reduce task can be further divided into K data shards, for example, and then these K fine-grained data shards are used as input data to execute the next Map-Reduce.

It should be understood that, generally, the N map tasks generated in the Map phase are executed by different first computing nodes, and multiple map tasks are executed on one first computing node. The M reduce tasks generated in the Reduce phase can also be executed by different second computing nodes, and multiple reduce tasks can also be executed on one second computing node.

Optionally, the map task and the reduce task can be executed by the same computing node (i.e., the first computing node and the second computing node are the same computing node), or can be executed by different computing nodes (i.e., the first computing node and the second computing node are different computing nodes), which is not limited in the embodiments of the present application.

Referring to Figure 2, a framework diagram for data processing based on the Map-Reduce model in a computing cluster, the data to be processed is divided into N data shards. The Map phase requires executing N map tasks, each of which is executed by multiple compute nodes in the computing cluster. For example, as shown in Figure 2, three compute nodes process N data shards. Compute node 1 executes N1 map tasks, compute node 2 executes N2 map tasks, and compute node 3 executes N3 map tasks. N1 + N2 + N3 = N. Each map task generates M data shards.

Continuing with Figure 2, in the Reduce phase, M reduce tasks need to be executed. These M reduce tasks can be executed by one or more compute nodes. Figure 2 illustrates that the M reduce tasks are executed by M compute nodes respectively. For example, compute node 4 executes reduce task 1, compute node 5 executes reduce task 2, and compute node 6 executes reduce task M. As shown in Figure 2, reduce task 1 needs to pull the first data shard from all the data shards generated by map tasks. Specifically, each of the N1 map tasks on compute node 1 generates M data shards. Therefore, reduce task 1 needs to pull N1 data shards from compute node 1. Each of the N2 map tasks on compute node 2 generates M data shards. Therefore, reduce task 1 needs to pull N2 data shards from compute node 2. The N3 map tasks on compute node 3 generate M data shards. Each map task in ask generates M data shards, then reduce task 1 needs to pull N3 data shards from computing node 3, and reduce task 1 needs to pull N (N1+N2+N3=N) data shards a total of N times; similarly, reduce task 2 needs to pull the second data shard among the data shards generated by all map tasks. For example, reduce task 2 pulls N1 data shards from computing node 1, N2 data shards from computing node 2, and N3 data shards from computing node 3, and needs to pull N times in total.

With reference to Figure 2, currently, in one implementation, when the map task and the reduce task are executed by different computing nodes, the data pulling process mentioned above needs to be performed across computing nodes. For example, the reduce task pulls the corresponding data shards from the data shards generated by multiple map tasks through shuffle (data pulling operation). Shuffle is a data distribution mechanism across nodes and processes within a computing cluster, and can also be understood as data exchange between the Reduce stage and the Map stage.

For more information about shuffle and the detailed process of pulling data through shuffle, please refer to the existing technical information, which will not be described in detail in this application.

The shuffle process requires cross-node data fetching over the network. Due to limited network bandwidth in the compute cluster, shuffling large numbers of data shards generated by map tasks consumes significant bandwidth. Furthermore, when there are a large number of map tasks, fetching these massive data shards repeatedly results in a high number of input/output operations per second (IOPS) on the network, which can easily cause network congestion and significant end-to-end data transmission latency (i.e., from the compute nodes executing the map tasks to the compute nodes executing the reduce tasks). Furthermore, when network congestion is caused by excessive network load, network performance deteriorates dramatically, impacting the normal operation of other jobs.

It should be understood that in the Reduce stage, after the reduce task pulls a data partition generated by all map tasks through shuffle, the types of operators executed by the reduce task may include but are not limited to Join (data splicing), Aggregate (aggregation), Filter (filtering), Sort (sorting), distinct (deduplication), subtract (set difference), sample (sampling) and other operators. Among them, each type of operator can also include different types of operators. For example, the Join operator can include cross join, inner join, outer join, self join and other types of operators.

Some operators executed during the Reduce phase feature a significant reduction in the amount of data before and after the operation. This means that the input data for these operators is large, but the output data is very small. This means that for these operators, shuffling across the network actually pulls in a massive amount of raw data (i.e., input data), resulting in more significant data transmission delays.

In view of the problem that in the above data processing process, the reduce task pulls all the data shards generated by the map task through shuffle, resulting in a large data transmission delay, the embodiment of the present application provides a data processing method, device and computing cluster, the computing cluster includes multiple first computing nodes, multiple first data processing units and multiple second computing nodes, wherein each first computing node is connected to a first data processing unit, the multiple first computing nodes can execute N first tasks (such as map tasks), the N first tasks are used to process N data shards of the data to be processed, and after executing each first task, M first results (M second results) are generated. A result is M data fragments), and they are respectively identified as A0~AM-1; each first data processing unit executes M second tasks, each of the M second tasks is used to process the first result with the same identification in the first computing node connected to the first data processing unit, and generate a second result after processing. After executing M second tasks, M second results are generated, and the M second results are identified as B0~BM-1; then multiple second computing nodes execute M third tasks, each third task is used to process the second results with the same identification in multiple first data processing units, and generate a third result. After executing M third tasks, M third results are generated.

The method has the following steps: during the data processing process, the first data processing unit connected to each first computing node executes a second task on the first result generated after the first computing node executes multiple first tasks to generate a second result, and then the second computing node executes a third task to process the second results generated by the multiple computing nodes. This can avoid the data generated after the same computing node executes multiple first tasks being directly pulled multiple times and in large quantities by the second computing node, thereby solving the problem of large data transmission delay caused by multiple and massive data pulling, that is, it can reduce the data transmission delay in the data processing process and improve the efficiency of data processing.

The data processing method, device and computing cluster provided in the embodiments of the present application can be applied to scenarios of big data distributed computing based on the Map-Reduce model. The computing cluster and data processing method provided in the embodiments of the present application are described in detail below.

An embodiment of the present application provides a computing cluster. FIG3 is a schematic diagram of the architecture of a computing cluster provided in an embodiment of the present application. The computing cluster includes multiple first computing nodes 301 and multiple second computing nodes 302 . Each first computing node 301 is connected to a first data processing unit 303 .

The computing nodes (including the first computing node and the second computing node) in the computing cluster can be servers, such as central servers, edge servers, or local servers in a local data center. In some embodiments, the computing nodes can also be terminal devices such as desktop computers, laptop computers, or smart phones.

Optionally, the data processing unit provided in the embodiments of the present application may be a DPU (data processing unit). It is understood that a DPU is a dedicated data processing unit (or data processor) with strong data processing capabilities. As a smart network card, the DPU connects to a host (such as a computing node) via a network card interface and performs data processing functions under the control of the computing node. Of course, the data processing unit connected to each computing node may also be other data processing devices, and the embodiments of the present application do not limit this.

For a data processing task, the data processing task is used to process the data to be processed, and the data to be processed is divided into N data slices for processing. In the embodiment of the present application, executing the data processing task includes executing N first tasks, M second tasks, and M third tasks, where M and N are both integers greater than or equal to 2. Taking the Map-Reduce model as an example, the first task mentioned above is a map task (hereinafter referred to as map task), and the second and third tasks mentioned above are reduce tasks (hereinafter referred to as reduce task). The N first tasks (map tasks) are executed by multiple first computing nodes 301, the M second tasks are executed by the first data processing unit 303, and the M third tasks are executed by multiple second computing nodes 302.

In the embodiment of the present application, the Map-Reduce model is used to process the N data shards into which the data to be processed is divided.

As shown in FIG4 , in conjunction with FIG3 , multiple first computing nodes 301 are configured to execute N first tasks (map tasks), each of which is configured to process N data shards of the to-be-processed data. Specifically, each first computing node 301 executes multiple first tasks to process multiple data shards of the N data shards. After executing each first task, M first results are generated, each of which is labeled A0 through AM-1.

Each of the multiple first data processing units 303 is configured to execute M second tasks (reduce tasks). Each of the M second tasks is configured to process first results with the same identifier in the first computing node to which the first data processing unit is connected, generating a second result after processing. After executing the M second tasks, M second results are generated, and the M second results are identified as B0 to BM-1. Referring to FIG4 , taking a second task executed by a first data processing unit as an example, when executing the second task, multiple first results with the identifier A0 are pulled from the multiple first results obtained after the connected first computing node executes multiple first tasks, and then the second task is executed to generate a second result B0.

Multiple second computing nodes 302 are configured to execute M third tasks (reduce tasks), each of which is configured to process second results with the same identifier from multiple first data processing units to generate a third result. After executing M third tasks, M third results are generated. Each second computing node 302 executes one or more third tasks to process second results generated by multiple first data processing units that are communicatively connected to the second computing node 302. For example, in FIG4 , the first second computing node 302 executes n1+1 reduce tasks, and the generated third results are identified as C0 to Cn1. Referring to FIG4 , taking the third task executed by a second computing node 302 as an example, when executing the third task, multiple second results with the identifier B0 are pulled from the multiple second results generated by the multiple first data processing units, and then the third task is executed to generate a single third result C0.

It should be noted that the input data of each of the multiple first tasks (i.e., one data shard among the N data shards) is different, and after executing the multiple first tasks, the first results generated by each first task are different. Similarly, the input data of each of the multiple second tasks is different, and after the multiple first data processing units execute the multiple second tasks, the second results generated are different. Therefore, the identification of the first results (A0~AM-1) and the second results (B0~BM-1) in Figure 4 is only used to identify the data shards in each processing result, and does not limit the content of the data shards in the processing results.

Referring to Figure 4 , for example, the first result (i.e., the first data split) among M first results generated by different map tasks is labeled A0, but this does not mean that the contents of the multiple first results labeled A0 are identical. For another example, the first result among M second results generated by different first data processing units after executing the first reduce task is labeled B0, but this does not mean that the contents of the multiple second results labeled B0 are identical.

In conjunction with FIG4 , as shown in FIG5 , the computing cluster provided in the embodiment of the present application may further include multiple second data processing units 304, with each second computing node 302 connected to a second data processing unit 304. When the computing cluster includes multiple second data processing units 304, the above-mentioned process of executing M third tasks (reduce tasks) by the multiple second computing nodes 302 may be replaced by execution by the second data processing units 304 connected to the second computing nodes 302. For example, each second computing node may instruct the second data processing unit 304 connected thereto to process one or more third tasks to process the second results with the same identifier in the multiple first data processing units. After the multiple second computing nodes execute the M third tasks, M third results are generated.

In the embodiment of the present application, the M third tasks executed by the second computing node are offloaded to the second data processing unit for execution, which can reduce the load of the second computing node and thus save resources of the second computing node.

Optionally, the computing cluster may also include other nodes, such as a management node, which is used to manage the computing nodes in the computing cluster. The management node may have task scheduling functions and resource management functions. For example, a task scheduler and/or a resource manager may be deployed on the management node to schedule data processing tasks and allocate computing resources and memory resources to data processing tasks.

In summary, the computing cluster provided by the embodiment of the present application can avoid the data generated after the same computing node executes multiple first tasks directly by the second computing node and pulls massive data multiple times, thereby solving the problem of large data transmission delay caused by multiple and massive data pulling, reducing the data transmission delay in the data processing process and improving the efficiency of data processing.

It can be understood that the multiple first computing nodes 301, the multiple second computing nodes 302, the multiple first data processing units 303, and the multiple second data processing units 304 in the computing cluster interact with each other to complete the processing of the data to be processed.

In one implementation, a third computing node among the plurality of first computing nodes 301 is further configured to send a first request to the connected first data processing unit 303 to instruct the first data processing unit 303 to execute M second tasks, wherein the third computing node may be any computing node among the plurality of first computing nodes 301. It is understandable that after each first computing node 301 executes the plurality of first tasks (map tasks), it sends the first request to the connected first data processing unit 303, so that the first data processing unit 303 begins to execute the second task on the plurality of first results generated in the first computing node.

Optionally, the first request includes the operator identifier of the operator to be called during the execution of the second task and the storage location information of the M first results. The operator identifier indicates the operator to be called included in the second task, and the storage location information indicates the storage location of the M first results, such as the physical address of the M first results in memory. In this manner, the first data processing unit 303 reads the M first results based on the storage location information and calls the operator indicated by the operator identifier to execute the second task.

In one implementation, a third data processing unit among the plurality of first data processing units 303 is further configured to send a notification to a fifth computing node among the plurality of second computing nodes to indicate the completion of execution of the M second tasks, wherein the third data processing unit is any first data processing unit among the plurality of first data processing units 303, and the fifth computing node is the second computing node corresponding to the third data processing unit (i.e., the second computing node for processing the results generated by the third data processing unit). Optionally, after the third data processing unit completes executing the M second tasks, it sends a notification indicating the completion of execution of the M second tasks to the first computing node 301 connected thereto, and the first computing node 301 then forwards the notification to the second computing node 302.

Optionally, the notification indicating the completion of the execution of the M second tasks carries the storage location information of the M second results. Thus, after receiving the notification, the second computing node reads the M second results according to the storage location information to execute the third task.

In one implementation, when the computing cluster includes multiple second data processing units 304, a fourth computing node among the multiple second computing nodes 302 is further configured to send a second request to the connected second data processing unit 304 to instruct the second data processing unit 304 to execute M third tasks, wherein the fourth computing node can be any computing node among the multiple second computing nodes 301. It is understandable that after each first data processing unit 303 executes multiple second tasks (reduce tasks), it sends a first request to the first data processing unit 303 connected thereto, so that the first data processing unit 303 begins to execute the second task on the multiple first results generated in the first computing node.

Optionally, the second request carries the operator identifier of the operator to be called during the execution of the third task and the storage location of the M second results. In this way, the second data processing unit 304 reads the M second results according to the storage location information and calls the operator indicated by the operator identifier to execute the third task.

It can be understood that each first data processing unit 303 executes M second tasks, and the process of executing the second task is actually calling the to-be-called operator included in the second task.

Based on this, in one implementation, a fourth data processing unit among the plurality of first data processing units 303 is further configured to receive a third request instructing the fourth data processing unit to update an operator to be called, and to load the operator logic microcode of the operator to be updated into the fourth data processing unit, thereby persisting the operator to be updated in the fourth data processing unit. The fourth data processing unit may be any one of the plurality of first data processing units 303.

Optionally, the third request carries the operator logic microcode of the operator to be updated and may further include an operator identifier, so that the operator logic microcode is stored according to the correspondence between the operator identifier and the operator logic microcode.

It should be understood that, generally, due to the limited hardware physical resources of a data processing unit (e.g., a DPU), only a small number of frequently used hardware operators can be solidified. In the embodiments of the present application, the logic circuit of the data processing unit supports hardware operator updates. Therefore, operators in the data processing unit can be flexibly updated, expanding the range of operators that can be executed by the data processing unit.

Optionally, in this embodiment of the present application, the operator to be updated can be a user-defined operator. Based on actual data processing requirements, the user can design and generate the operator logic microcode and update the customized operator to the data processing unit, thereby persisting the operator logic microcode within the data processing unit. Supporting user-defined operators can meet the needs of different users and expand the application range of computing clusters.

The following embodiments describe the data processing process from the perspective of the internal module interactions between the various computing nodes and data processing units in a computing cluster. Taking the computing cluster shown in Figure 5 as an example, for ease of description, the data processing process of the computing cluster is explained using two first computing nodes and one second computing node in the computing cluster as examples.

As shown in Figure 6, for the first computing node 301a in the computing cluster, the first data processing unit 303a connected to the first computing node 301a, the first computing node 301b, the first data processing unit 303b connected to the first computing node 301b, the second computing node 302 and the second data processing unit 304 connected to the second computing node 302.

Among them, the first computing node 301a, the first computing node 301b and the second computing node 302 all include an agent module, which is responsible for converting the received request from business semantics into a hardware computing request. The first data processing unit 303a, the first data processing unit 303b and the second data processing unit 304 all include a request handler module and a data offloading engine (DOE), which is a hardware module for operator calculation.

Among them, after the first computing node 301a executes multiple map tasks (first tasks), each map task generates M first results (identified as A0~AM-1), and the agent module in the first computing node 301a sends a first request to the request processing module in the first data processing unit 303a to instruct the first data processing unit 303a to execute the reduce task (second task) on the multiple first results generated by the first computing node 301b.

The request processing module in the first data processing unit 303a forwards the first request to the DOE of the first data processing unit 303a, causing the DOE to execute the reduce task. For example, referring to Figure 6, for each first computing node 301a, the first data processing unit 303a pulls the first result A0 generated by each map task and executes the reduce task (the reduce task execution process may require the invocation of operators such as Join and Aggregate), resulting in a second result B0.

It should be understood that the process of data processing performed by the first computing node 301b and the first data processing unit 303b can refer to the process of data processing performed by the first computing node 301a and the first data processing unit 303a, which will not be repeated here.

Furthermore, after the first data processing unit 303a and the first data processing unit 303b have executed multiple reduce tasks, they notify the proxy module of the second computing node 302 through the proxy module, and then the proxy module of the second computing node 302 sends a second request to the request processing module of the second data processing unit 304 to instruct the second data processing unit 304 to execute M reduce tasks (the third task).

The request processing module of the second data processing unit 304 forwards the second request to the DOE of the second data processing unit 304, causing the DOE to execute the reduce task. For example, referring to Figure 6, the second data processing unit 304 pulls the second result B0 generated by the first data processing unit 303a and the second result B0 generated by the first data processing unit 303b, and executes the reduce task, obtaining a third result C0 after the execution is completed.

Based on the above content, it can be seen that in the embodiment of the present application, all operator operations in the Reduce stage originally completed by the second computing node 302 are first executed as node-level reduce tasks within the first computing node according to the granularity of the computing node to obtain M second results. In this way, the number of times and the amount of data pulled by the second computing node are reduced.

Based on the architecture of the computing cluster described in the above embodiments, the embodiments of the present application also provide a data processing method. The data processing method provided in the embodiments of the present application is described in more detail below from the perspective of interaction between various modules within a first computing node, a first data processing unit (first DPU), a second computing node, and a second data processing unit (second DPU).

As shown in FIG7 , in one implementation, the data processing method provided in an embodiment of the present application includes the following steps.

S701: After executing a plurality of first tasks, the processing module of the first computing node sends a notification message to the agent module of the first computing node to indicate completion of execution of the plurality of first tasks.

After each first task is executed, M first results are generated (respectively labeled A0 to AM-1). Optionally, after the first computing node executes each first task, the first results generated are cached in the memory of the first computing node.

S702. The agent module of the first computing node sends a first request to the request processing module of the first DPU.

The first request is used to instruct the first DPU to execute M second tasks. The first request carries the operator identifier of the operator to be called during the execution of the second task and the storage location information of the M first results. For the description of the first request, please refer to the description of the above embodiment and will not be repeated here.

S703: After receiving the first request, the request processing module of the first DPU applies for a cache that meets the required size.

The requested cache is used to store the result generated after the first DPU executes the task.

S704: The request processing module of the first DPU sends a first request to the DOE of the first DPU to trigger the DOE of the first DPU to execute M second tasks.

Optionally, the request processing module of the first DPU fills the operator identifier of the operator to be called and the storage location information of the M first results into the register of the DOE of the first DPU to trigger the DOE to execute the M second tasks.

S705 : The DOE of the first DPU executes M second tasks and generates M second results (respectively labeled as B0 to BM-1).

After the DOE of the first DPU obtains the operator identifier of the operator to be called and the storage location information of the M first results, it obtains the M first results and executes the M second tasks to obtain the M second results.

The operators called during the execution of the second task may include operators with the following characteristics: operators that have a large input data volume and a significantly reduced output data volume after execution. Examples include the Join operator and Aggregate operator mentioned in the above embodiments. The operators called during the second task may also include other operators, such as the Sort operator.

S706 : The DOE of the first DPU writes the M second results into the requested cache.

S707: The first DPU sends a notification message to the agent module of the first computing node through the request processing module of the first DPU. The notification message is used to notify the completion of execution of the M second tasks.

The notification message may carry the storage location information of the M second results and the data length of the M second results.

S708. The proxy module of the first computing node forwards the notification message in S707 to the proxy module of the second computing node.

S709. The second computing node sends a second request to the request processing module of the second DPU through its processing module and proxy module.

The second request is used to instruct the second data processing unit to execute M third tasks. The second request carries the operator identifier of the operator to be called during the execution of the third task and the storage location information of the M second results.

Optionally, the proxy module of the second computing node fills the storage location information of the M second results generated by the first DPU and the operator identifier of the operator to be called during the execution of the third task into the second request according to the interface format and sends it to the request processing module of the second DPU.

S710: The request processing module of the second DPU obtains M second results generated by the first DPU from the cache.

Specifically, the request processing module of the second DPU obtains M second results from the cache corresponding to each first DPU, that is, the request processing module of the second DPU obtains M second results generated by connecting multiple first computing nodes to multiple first DPUs.

S711: The request processing module of the second DPU applies for cache.

S712: The request processing module of the second DPU sends a second request to the DOE of the second DPU.

The second request also carries M second results obtained from the cache (generated by each first DPU executing the M second tasks).

S713: The DOE of the second DPU executes M third tasks and generates M third results.

S714 : The DOE of the second DPU writes the M third results into the cache.

S715. The DOE of the second DPU sends a notification message indicating completion of execution of the M third tasks to the second computing node through the request processing module of the second DPU.

S716. The second computing node obtains M third results from the cache.

As shown in FIG8 , in another implementation, the data processing method provided in an embodiment of the present application includes the following steps.

S801: After executing a plurality of first tasks, the processing module of the first computing node sends a notification message to the agent module of the second computing node to indicate completion of execution of the plurality of first tasks.

After each first task is executed, M first results are generated.

S802: The proxy module of the second computing node sends a fourth request to the proxy module of the first computing node.

The fourth request is used to instruct the first DPU connected to the first computing node to execute M second tasks.

In one implementation, the second computing node can obtain a list of computing nodes where the data required to perform the second task (M first results) are located from the metadata server node (which can be implemented by the management node), and send a fourth request to the agent module of the first computing node.

The fourth request carries the index information of the data and the operator identifier of the operator to be called during the execution of the second task, wherein the index information of the data is used to indicate the characteristics of the data required for the second task, and the index information of the data may include the identification information of the application (such as APP id), shuffle id, and stage id, wherein APP id is used to indicate which application the input data comes from, shuffle id is used to indicate which shuffle operation of the application the input data corresponds to, and stage id is used to indicate the task (indicating the second task).

S803. The agent module of the first computing node sends the first request to the request processing module of the first DPU according to the fourth request.

For the description of the first request, reference may be made to the relevant description in the above embodiment.

The first computing node determines the storage locations (such as physical addresses) in the memory of M first results generated after the first computing node executes multiple first tasks through the index information in a mapping table (a mapping table of index information and addresses).

S804: After receiving the first request, the request processing module of the first DPU applies for a cache that meets the required size.

S805: The request processing module of the first DPU sends a first request to the DOE of the first DPU to trigger the DOE of the first DPU to execute M second tasks.

S806: The DOE of the first DPU executes M second tasks and generates M second results.

S807 : The DOE of the first DPU writes the M second results into the requested cache.

S808. The first DPU sends a notification message to the agent module of the first computing node through the request processing module of the first DPU. The notification message is used to notify the completion of execution of the M second tasks.

S809. The proxy module of the first computing node forwards the notification message in S808 to the proxy module of the second computing node.

S810. The second computing node sends a second request to the request processing module of the second DPU through its processing module and proxy module.

For the description of the second request, reference may be made to the relevant description in the above embodiment.

S811. The request processing module of the second DPU obtains M second results generated by the first DPU from the cache.

S812: The request processing module of the second DPU applies for cache.

S813: The request processing module of the second DPU sends a second request to the DOE of the second DPU.

The second request also carries M second results obtained from the cache (generated by each first DPU executing the M second tasks).

S814. The DOE of the second DPU executes M third tasks and generates M third results.

S815. The DOE of the second DPU writes the M third results into the cache.

S816. The DOE of the second DPU sends a notification message indicating the completion of execution of the M third tasks to the second computing node through the request processing module of the second DPU.

S817. The second computing node obtains M third results from the cache.

In summary, the data processing method described in S701-S716 and S801-S817, since the first data processing unit connected to each first computing node can first execute the second task on the first result generated after the first computing node executes multiple first tasks to generate the second result, and then the second computing node executes the third task to process the second results generated by multiple computing nodes, it can avoid the data generated after the same computing node executes multiple first tasks being directly pulled by the second computing node multiple times and massive data, thereby solving the problem of large data transmission delay caused by multiple and massive data pulling, and can reduce the data transmission delay in the data processing process and improve the efficiency of data processing.

In combination with the contents of the above embodiments, in an embodiment of the present application, the DPU may include a programmable logic circuit (such as an FPGA), which can support users to update custom operators to the DPU, that is, persist them to the DPU, so that operators can be updated flexibly during the DPU execution of tasks.

In an embodiment of the present application, operator update can be achieved by interacting with the computing node and the DPU connected thereto. Referring to FIG9 , the operator update process includes the following steps.

S901. The proxy module of the computing node receives a third request to instruct the DPU connected to the computing node to update an operator to be called.

The third request carries the operator identifier (such as operator ID) and the operator logic microcode of the operator to be updated.

Optionally, in an embodiment of the present application, the third request may be triggered by a user, and the user may send a request for online update of the hardware operator to the proxy module of the computing node through the configuration interface.

S902: The proxy module of the computing node sends a third request to the request processing module of the DPU.

S903: The request processing module of the DPU forwards the third request to the DOE of the DPU.

S904. The DOE of the DPU stores the operator logic microcode corresponding to the operator identifier.

After the DPU's DOE receives the third request, it first determines whether the operator indicated by the operator identifier in the third request is a new operator, that is, determines whether the operator logic microcode is a new operator logic microcode. If it is a new operator logic microcode, the DOE will persist the operator logic microcode in the DPU's flash memory.

S905. The DPU's DOE loads the operator logic microcode from the flash into the DOE's hardware engine pool.

It can be understood that when DOE receives the first request or the second request in the above embodiment, it loads the operator logic microcode of the operator from the flash into the hardware engine pool of DOE according to the operator identifier carried in the request, thereby executing the corresponding task.

In summary, as described above for S901-S905, in this embodiment of the present application, the logic circuitry of the data processing unit supports hardware operator updates. Therefore, operators within the data processing unit can be flexibly updated, expanding the range of operators that can be executed by the data processing unit. Furthermore, these operators to be updated can be user-defined operators. Supporting user-defined operators can meet the needs of diverse users, broadening the application scope of the computing cluster.

An embodiment of the present application also provides a computing node, including a memory and at least one processor connected to the memory, the memory being used to store computer program code, the computer program code including computer instructions, which, when executed by at least one processor, causes the computing node to perform an action performed by a first computing node in a computing cluster; or, to perform an action performed by a second computing node in the computing cluster.

An embodiment of the present application provides a data processing unit, including a memory and at least one processor connected to the memory, wherein the memory is used to store computer program code, and the computer program code includes computer instructions. When the computer instructions are executed by the at least one processor, the data processing unit performs the action performed by the first data processing unit in the computing cluster; or, performs the action performed by the second data processing unit in the computing cluster.

An embodiment of the present application further provides a computer-readable storage medium storing computer instructions. When the computer instructions are run on a computer, the data processing method described in the above embodiment is executed.

An embodiment of the present application further provides a computer program product, which includes computer instructions. When the computer instructions are run on a computer, the data processing method described in the above embodiment is executed.

An embodiment of the present application also provides a chip system, including: a processor, used to call and run a computer program from a memory, so that a device equipped with the chip system executes the data processing method described in the above embodiment.

In the above embodiments, all or part of the embodiments may be implemented by software, hardware, firmware or any combination thereof. When implemented using a software program, all or part of the embodiments may be implemented in the form of a computer program product. The computer program product includes one or more computer instructions. When the computer instructions are loaded and executed on a computer, all or part of the processes or functions in accordance with the embodiments of the present application are generated. The computer may be a general-purpose computer, a special-purpose computer, a computer network or other programmable device. The computer instructions may be stored in a computer-readable storage medium or transmitted from one computer-readable storage medium to another computer-readable storage medium. For example, the computer instructions may be transmitted from one website, computer, server or data center to another website, computer, server or data center via a wired (e.g., coaxial cable, optical fiber, digital subscriber line (DSL)) or wireless (e.g., infrared, wireless, microwave, etc.) method. The computer-readable storage medium may be any available medium that a computer can access or a data storage device such as a server or data center that includes one or more available media integrated therein. The available medium can be a magnetic medium (e.g., floppy disk, magnetic disk, tape), an optical medium (e.g., a digital video disc (DVD)), or a semiconductor medium (e.g., a solid state drive (SSD)), etc.

Through the description of the above embodiments, those skilled in the art will clearly understand that for the sake of convenience and brevity, only the division of the above functional modules is used as an example. In actual applications, the above functions can be assigned to different functional modules as needed, that is, the internal structure of the device can be divided into different functional modules to complete all or part of the functions described above. The specific working processes of the above-described systems, devices, and units can refer to the corresponding processes in the aforementioned method embodiments and will not be repeated here.

In the several embodiments provided in this application, it should be understood that the disclosed systems, devices and methods can be implemented in other ways. For example, the device embodiments described above are merely schematic. For example, the division of the modules or units is only a logical function division. In actual implementation, there may be other division methods, such as multiple units or components can be combined or integrated into another system, or some features can be ignored or not executed. Another point is that the mutual coupling or direct coupling or communication connection shown or discussed can be an indirect coupling or communication connection through some interfaces, devices or units, which can be electrical, mechanical or other forms.

The units described as separate components may or may not be physically separate, and the components shown as units may or may not be physical units, that is, they may be located in one place or distributed across multiple network units. Some or all of these units may be selected to achieve the purpose of this embodiment according to actual needs.

In addition, the functional units in the various embodiments of the present application may be integrated into a single processing unit, or each unit may exist physically separately, or two or more units may be integrated into a single unit. The aforementioned integrated units may be implemented in the form of hardware or software functional units.

If the integrated unit is implemented in the form of a software functional unit and sold or used as an independent product, it can be stored in a computer-readable storage medium. Based on this understanding, the technical solution of the present application is essentially or the part that contributes to the prior art or all or part of the technical solution can be embodied in the form of a software product, and the computer software product is stored in a storage medium, including a number of instructions for enabling a computer device (which can be a personal computer, server, or network device, etc.) or a processor to execute all or part of the steps of the method described in each embodiment of the present application. The aforementioned storage medium includes: various media that can store program codes, such as flash memory, mobile hard disk, read-only memory, random access memory, magnetic disk or optical disk.

The above is only a specific embodiment of the present application, but the scope of protection of this application is not limited to this. Any changes or substitutions within the technical scope disclosed in this application should be included in the scope of protection of this application. Therefore, the scope of protection of this application should be based on the scope of protection of the claims.

Claims (19)

A computing cluster, comprising: A plurality of first computing nodes, configured to execute N first tasks, wherein the N first tasks are configured to process N data shards of to-be-processed data, and generate M first results after executing each first task, and the results are identified as A0 to AM-1, respectively; a plurality of first data processing units, each first computing node being connected to a first data processing unit; each first data processing unit being configured to execute M second tasks, each of the M second tasks being configured to process a first result having the same identifier in the first computing node to which the first data processing unit is connected, and generate a second result after the processing; after executing the M second tasks, M second results are generated, and the M second results are identified as B0 to BM-1; Multiple second computing nodes are used to execute M third tasks, each third task is used to process the second results with the same identifier in the multiple first data processing units to generate a third result, and M third results are generated after executing M third tasks. The computing cluster according to claim 1, characterized in that The first task is a map task, and the second and third tasks are reduce tasks. The computing cluster according to claim 1 or 2, characterized in that: The third computing node among the multiple first computing nodes is further configured to send a first request to the connected first data processing unit to instruct the first data processing unit to execute the M second tasks. The computing cluster according to claim 3, characterized in that The first request carries the operator identifier of the operator to be called during the execution of the second task and the storage location information of the M first results. The computing cluster according to any one of claims 1 to 4, further comprising: a plurality of second data processing units, each second computing node being connected to a second data processing unit; The fourth computing node among the multiple second computing nodes is also used to send a second request to the connected second data processing unit to instruct the second data processing unit to execute the M third tasks; the second request carries the operator identifier of the operator to be called during the execution of the third task and the storage location information of the M second results. The computing cluster according to any one of claims 1 to 5, characterized in that: The third data processing unit among the multiple first data processing units is further configured to send a notification to a fifth computing node among the multiple second computing nodes to indicate that the execution of the M second tasks has ended. The computing cluster according to any one of claims 1 to 6, characterized in that: The fourth data processing unit among the multiple first data processing units is also used to receive a third request to instruct the update of the operator to be called by the fourth data processing unit; the third request carries the operator logic microcode of the operator to be updated; and loads the operator logic microcode of the operator to be updated. The computing cluster according to claim 7, characterized in that The operator to be updated is a user-defined operator. A data processing method, characterized by being applied to a computing cluster, wherein the computing cluster includes a plurality of first computing nodes, a plurality of first data processing units, and a plurality of second computing nodes, wherein each first computing node is connected to a first data processing unit; the method comprises: The plurality of first computing nodes execute N first tasks, and generate M first results after executing each first task, and the results are respectively identified as A0 to AM-1; the N first tasks are used to process N data slices of the data to be processed; Each first data processing unit executes M second tasks, and generates M second results after executing the M second tasks, wherein the M second results are identified as B0 to BM-1; each of the M second tasks is used to process the first result with the same identification in the first computing node connected to the first data processing unit, and generate the second result after processing; Multiple second computing nodes execute M third tasks, and generate M third results after executing the M third tasks; each third task is used to process the second results with the same identifier in the multiple first data processing units to generate the third result. The method according to claim 9, characterized in that The first task is a map task, and the second and third tasks are reduce tasks. The method according to claim 9 or 10, characterized in that the method further comprises: A third computing node among the plurality of first computing nodes sends a first request to the connected first data processing unit to instruct the first data processing unit to execute the M second tasks. The method according to claim 11, characterized in that The first request carries the operator identifier of the operator to be called during the execution of the second task and the storage location information of the M first results. The method according to any one of claims 9 to 12, wherein the computing cluster further comprises: a plurality of second data processing units, each second computing node being connected to a second data processing unit; the method further comprises: The fourth computing node among the multiple second computing nodes sends a second request to the connected second data processing unit to instruct the second data processing unit to execute the M third tasks; the second request carries the operator identifier of the operator to be called during the execution of the third task and the storage location information of the M second results. The method according to any one of claims 9 to 13, further comprising: The third data processing unit among the plurality of first data processing units sends a notification to a fifth computing node among the plurality of second computing nodes to indicate that the execution of the M second tasks is completed. The method according to any one of claims 9 to 14, further comprising: A fourth data processing unit among the plurality of first data processing units receives a third request to instruct an update of an operator to be called by the fourth data processing unit, wherein the third request carries an operator logic microcode of the operator to be updated; The fourth data processing unit loads the operator logic microcode of the operator to be updated. The method according to claim 15, characterized in that The operator to be updated is a user-defined operator. A data processing device, characterized in that it includes a memory and at least one processor connected to the memory, the memory is used to store computer program code, and the computer program code includes computer instructions. When the computer instructions are executed by at least one processor, the data processing device executes the method according to any one of claims 9 to 16. A computer-readable storage medium, characterized in that computer instructions are stored therein, and when the computer instructions are executed on a computer, the method according to any one of claims 9 to 16 is executed. A computer program product, characterized in that it contains instructions, which, when executed by a computing device, cause the computing device to perform the method according to any one of claims 9 to 16.