Patent Publication Number: US-9411659-B2

Title: Data processing method used in distributed system

Description:
This application claims priority from Korean Patent Application No. 10-2013-0096118 filed on Aug. 13, 2013 in the Korean Intellectual Property Office, the disclosure of which is incorporated herein by reference in its entirety. 
     BACKGROUND OF THE INVENTION 
     1. Field of the Invention 
     The present invention relates to a data processing method used in a distributed system. 
     2. Description of the Related Art 
     Hadoop and MapReduce are technologies that can be used to implement distributed computing. In these technologies, an increase in the size of a cluster leads to an increase in data processing speed. However, increase in the the size of the cluster increases other complexities like increases power and space requirement. Therefore, a technology for increasing data processing speed without adding a new node to the cluster is required. 
     SUMMARY OF THE INVENTION 
     Aspects of the present invention provide a data processing method which can increase data processing speed without adding a new node to a distributed system. 
     However, aspects of the present invention are not restricted to the one set forth herein. The above and other aspects of the present invention will become more apparent to one of ordinary skill in the art to which the present invention pertains by referencing the detailed description of the present invention given below. 
     According to an aspect of the present invention, there is provided a data processing method including: calculating a conversion number of cores corresponding to a number of processing blocks included in a graphics processing unit (GPU) of a node of a distributed system; calculating a adding up number of cores by adding up a number of cores included in a central processing unit (CPU) of the node of the distributed system and the conversion number of cores; splitting job data allocated to the node of the distributed system into a number of job units data equal to the adding up number of cores; and allocating a number of job units data equal to the number of cores included in the CPU to the CPU of the node of the distributed system and a number of job units data equal to the conversion number of cores to the GPU of the node of the distributed system. 
     According to another aspect of the present invention, there is provided a data processing method including: calculating a conversion number of first-type processors corresponding to a number of second-type processors included in a node of a distributed system; calculating a adding up number of first-type processors by adding up a number of first-type processors included in the node of the distributed system and the conversion number of first-type processors; splitting job data allocated to the node of the distributed system into a number of job data unit equal to the adding up number of first-type processors; allocating a number of job units data equal to the number of first-type processors included in the node of the distributed system to the first-type processors of the node of the distributed system; and allocating a number of job units data equal to the conversion number of first-type processors to the second-type processors of the node of the distributed system, wherein the job units data equal to the conversion number of the first-type processors are split into a plurality of data blocks whose sizes are respectively proportional to throughputs of the second-type processors per unit of time, and the data blocks are allocated to the second-type processors, respectively. 
    
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS 
       The above and other aspects and features of the present invention will become more apparent by describing in detail exemplary embodiments thereof with reference to the attached drawings, in which: 
         FIG. 1  is a schematic diagram of a distributed system according to an embodiment of the present invention; 
         FIG. 2  illustrates a master node of  FIG. 1 ; 
         FIG. 3  illustrates a slave node of  FIG. 1 ; 
         FIG. 4  illustrates the physical structure of the slave node of  FIG. 1 ; 
         FIG. 5  is a schematic diagram illustrating a data processing method according to an embodiment of the present invention; 
         FIG. 6  illustrates a data processing process of a slave node according to an embodiment of the present invention; 
         FIG. 7  illustrates a central processing unit (CPU) and a graphics processing unit (GPU) included in a slave node of the distributed system according to an embodiment of the present invention; 
         FIG. 8  illustrates a process of calculating the conversion number of cores corresponding to the number of processing blocks of a GPU included in a slave node of the distributed system according to an embodiment of the present invention; 
         FIG. 9  illustrates a process of distributing input data to a CPU and a GPU according to an embodiment of the present invention; 
         FIG. 10  illustrates a process of distributing input data to a CPU and a GPU according to anther embodiment of the present invention; 
         FIG. 11  is a flowchart illustrating a data processing method according to an embodiment of the present invention; and 
         FIG. 12  is a flowchart illustrating a data processing method according to another embodiment of the present invention. 
     
    
    
     DETAILED DESCRIPTION OF THE INVENTION 
     Advantages and features of the present invention and methods of accomplishing the same may be understood more readily by reference to the following detailed description of exemplary embodiments and the accompanying drawings. The present invention may, however, be embodied in many different forms and should not be construed as being limited to the embodiments set forth herein. Rather, these embodiments are provided so that this disclosure will be thorough and complete and will fully convey the concept of the invention to those skilled in the art, and the present invention will only be defined by the appended claims. Like reference numerals refer to like elements throughout the specification. 
     It will be understood that, when an element is referred to as being “connected to” or “coupled” to another element, it can be directly connected or coupled to the other element or intervening elements may be present. In contrast, when an element is referred to as being “directly connected to” or “directly coupled” to another element, there are no intervening elements present. Like numbers refer to like elements throughout. As used herein, the term “and/or” includes any and all combinations of one or more of the associated listed items. 
     It will be understood that, although the terms first, second, third, etc., may be used herein to describe various elements, components and/or sections, these elements, components and/or sections should not be limited by these terms. These terms are only used to distinguish one element, component or section from another element, component or section. Thus, a first element, component or section discussed below could be termed a second element, component or section without departing from the teachings of the present invention. 
     The terminology used herein is for the purpose of describing particular embodiments only and is not intended to be limiting of the invention. 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 be further understood that the terms “comprises” and/or “comprising,” when used in this specification, specify the presence of stated components, steps, operations, and/or elements, but do not preclude the presence or addition of one or more other components, steps, operations, and/or elements. 
     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 to which this invention belongs. It will be further understood that terms, such as those defined in commonly used dictionaries, should be interpreted as having a meaning that is consistent with their meaning in the context of the relevant art and will not be interpreted in an idealized or overly formal sense unless expressly so defined herein. 
       FIG. 1  is a schematic diagram of a distributed system according to an embodiment of the present invention. 
     Referring to  FIG. 1 , the distributed system according to the current embodiment may include a client  100 , a master node  200 , and slave nodes  300 . A cluster that forms the distributed system may include one or more master nodes  200  and one or more slave nodes  300 . In  FIG. 1 , one master node  200  and n slave nodes  300  are illustrated. In some embodiments of the present invention, the cluster may be, e.g., a Hadoop cluster employing Hadoop architecture. 
     The client  100  may transmit data to be processed to the cluster and retrieve or read the result of processing the data. When transmitting the data to be processed to the cluster, the client  100  according to embodiments of the present invention may specify in which way the data should be processed. 
     The client  100  may be connected to the master node  200  or the slave nodes  300  and exchange various information with the master node  200  or the slave nodes  300 . In an example, the client  100  may request the master node  200  to store data to be processed, and the master node  200  may designate a slave node  300 , which will store the data, for the client  100 . Then, the client  200  may store the data to be processed in the slave node  300  designated by the master node  200 . In another example, the client  100  may request the master node  200  to calculate data stored in the cluster, and the master node  200  may generate a job for calculating the data stored in the cluster and transmit the job to a slave node  300 . Then, the client  100  may receive a value of the result of processing or calculating the data from the slave node  300 . 
     The client  100 , the master node  200 , and the slave nodes  300  may be connected to one another through a network. According to embodiments of the present invention, the network may be, but is not limited to, a wireless network such as WiFi or a wired network such as a local area network (LAN). In the current embodiment of the present invention, each of the client  100 , the master node  200 , and the slave nodes  300  may be a single server. In another embodiment of the present invention, at least one of the client  100 , the master node  200 , and the slave nodes  300  may be included in one server that plays multiple roles. Here, the server may be a physical server itself or a server run on a personal computer (PC) (e.g., a desktop computer, a notebook computer, etc.), a tablet, and a smartphone. In particular, in the current embodiment of the present invention, the server may be a rack server. 
       FIG. 2  illustrates the master node  200  of  FIG. 1 .  FIG. 3  illustrates a slave node  300  of  FIG. 1 . One or more master nodes  200  and one or more slave nodes  300  that form one cluster can be described from a perspective of processing distributed data and a perspective of storing distributed data. The perspective of processing distributed data includes generating a job for processing data stored in the cluster and processing the data stored in the cluster using the generated job. On the other hand, the perspective of storing distributed data includes storing data to be processed in the cluster and reading the result of processing the data from a slave node  300  using the client  100 . In a Hadoop cluster according to an embodiment of the present invention, MapReduce may be taken into consideration in the perspective of processing distributed data, and a Hadoop distributed file system (HDFS) may be taken into consideration in the perspective of storing distributed data. The master node  200  and the slave node  300  will now be described from these perspectives with reference to  FIGS. 2 and 3 . 
     Referring to  FIG. 2 , the master node  200  may include a JobTracker  202  and a NameNode  204 . In the current embodiment, the JobTracker  202  and the NameNode  204  may be server instances generated in the master node  200 . The JobTracker  202  is a server instance corresponding to the perspective of processing distributed data. For example, when the client  100  makes a request to process data stored in the cluster, the JobTracker  202  may generate a job for calculating the data and transmit the generated job to a slave node  300 . The NameNode  204  is an instance corresponding to the perspective of storing distributed data. For example, when the client  100  makes a request to store data, the NameNode  204  may designate a slave node  300  which will store the data. 
     Referring to  FIG. 3 , the slave node  300  may include a TaskTracker  302  and a DataNode  304 . In the current embodiment of the present invention, the TaskTracker  302  and the DataNode  304  may be server instances generated in the slave node  300 . The TaskTracker  302  is an instance corresponding to the perspective of processing distributed data. For example, the TaskTracker  302  may process data stored in the slave node  300  at the request of the client  100 . The DataNode  304  is an instance corresponding to the perspective of storing distributed data. For example, the DataNode  304  may store data, which is to be processed, at the request of the client  100 . 
       FIG. 4  illustrates the physical structure of the slave node  300  of  FIG. 1 . The slave node  300  may include a processor  400  which processes data stored in the cluster and a storage  500  which stores data to be processed or the result of processing data. The processor  400  may include at least one of a central processing unit (CPU)  402  and a graphics processing unit (GPU)  404 . In some embodiments of the present invention, the processor  400  may include a plurality of CPUs  402  and a plurality of GPUs  404 . In some embodiments of the present invention, the processor  400  may be a semiconductor device such as a field programmable gate array (FPGA). The storage  500  may include a hard disk drive (HDD), a solid state drive (SSD), and an optical drive such as a compact disc (CD) or a digital video disc (DVD). Although not illustrated in  FIG. 4 , the slave node  300  may include a memory such as a random access memory (RAM). 
       FIG. 5  is a schematic diagram illustrating a data processing method according to an embodiment of the present invention. 
     Referring to  FIG. 5 , a client  100  may receive, e.g., from a user, input data  600  to be processed. The input data  600  may be split into one or more job data blocks  600   a  through  600   c  that can be processed in a distributed manner, in other words, processed by a plurality of slave nodes. In the embodiment of  FIG. 5 , the input data  600  is split into three job data blocks  600   a  through  600   c.    
     The job data blocks  600   a  through  600   c  are stored in a plurality of slave nodes  300   a  through  300   c  and then processed by processors  400  of the slave nodes  300   a  through  300   c , respectively. The slave nodes  300   a  through  300   c  produce the results of processing (e.g., calculating) the job data blocks  600   a  through  600   c  as result data  610   a  through  610   c  and store the result data  610   a  through  610   c  in their storages  500 , respectively. In the current embodiment, any one of the slave nodes  300   a  through  300   c  that form a cluster may collect the result data  610   a  through  610   c  stored in the storages  500  of the slave nodes  300   a  through  300   c  and produce one final result  620 , and the client  100  may retrieve or read the final result  620 . 
       FIG. 6  illustrates a data processing process of a slave node according to an embodiment of the present invention. 
     Referring to  FIG. 6 , a slave node  300   a  may receive a job data block  600   a . In the current embodiment of the present invention, the slave node  300   a  may store the job data block  600   a  in its storage  500  or in a memory. The slave node  300   a  illustrated in  FIG. 6  may include a CPU  402  and a GPU  404  as a processor  400 . In some embodiments of the present invention, the CPU  402  may include a plurality of cores, and the GPU  404  may be provided in a plurality. 
     The job data block  600   a  into which input data  600  has been split for distributed processing may additionally be split into a first data block  602  to be processed by the CPU  402  and a second data block  604  to be processed by the GPU  404 . Since the job data block  600   a  is additionally split into the first data block  602  and the second data block  604  within one slave node  300   a  and then processed by the CPU  402  and the GPU  404 , it can be processed (calculated) with increased speed. The process of splitting the input data  600  into the first data block  602  to be processed by the CPU  402  and the second data block  604  to be processed by the GPU  404  will be described later with reference to  FIGS. 8 through 10 . 
     Referring continuously to  FIG. 6 , the slave node  300   a  may further include a result data collector  700 . In the current embodiment of the present invention, the result data collector  700  may be provided separate from the CPU  402 . In another embodiment of the present invention, a core of the CPU  402  may operate as the result data collector  700 . The result data collector  700  may collect the processing result of the CPU  402  and the processing result of the GPU  404  and produce result data  610   a.    
       FIG. 7  illustrates a CPU  402  and a GPU  404  included in a slave node  300   a  of the distributed system according to an embodiment of the present invention.  FIG. 8  illustrates a process of calculating the number of cores (hereinafter, referred to as a conversion number of cores) corresponding to the number of processing blocks of the GPU  404  included in the slave node  300   a  of the distributed system according to an embodiment of the present invention. 
     Referring to  FIG. 7 , the CPU  402  according to the current embodiment of the present invention may include a plurality of cores  412  through  417 , and the GPU  404  may include a plurality of processing blocks included in a GPU grid  420 . To further improve the data processing speed of the distributed system, the slave node  300   a  according to the current embodiment of the present invention uses both the CPU  402  and the GPU  404 . Here, it is required to appropriately adjust a ratio of data processed by the CPU  402  and data processed by the GPU  404  in order to satisfactorily improve the data processing speed. 
     In the current embodiment of the present invention, to split a job data block  600   a  into a first data block  602  to be processed by the CPU  402  and a second data block  604  to be processed by the GPU  404 , the conversion number of cores corresponding to the number of processing blocks included in the GPU  404  of the slave node  300   a  is calculated. In other words, the job data block  600   a  is split into the first data block  602  and the second data block  604  by converting the throughput of the GPU  404  for a predetermined period of time in a processor  400  of the slave node  300   a  into the number of CPU cores of the processor  400  of the slave node  300   a . Here, the size of the first data block  602  corresponds to the number of cores included in the CPU  402  of the slave node  300   a , and the size of the second data block  604  corresponds to the conversion number of cores. 
     In the current embodiment of the present invention, for example, referring to  FIG. 8 , the conversion number of cores corresponding to the number of processing blocks included in the GPU  404  of the slave node  300   a  may be two. In some embodiments, this may indicate that the processing power or performance of the GPU  404  corresponds to the processing power or performance of two cores of the CPU  402 . 
     A method of calculating the conversion number of cores corresponding to the number of processing blocks included in the GPU  404  of the slave node  300   a  will now be described in detail. To calculate the conversion number of cores, an average number of processing blocks included in one GPU  404  of the slave node  300   a  is calculated. To this end, a total number of processing blocks of all GPUs  404  is divided by the number of GPUs  404  included in the slave node  300   a . Assuming that N G  is the number of GPUs  404  included in the slave node  300   a  and that G C  is the sum of the numbers of processing blocks included in the GPUs  404  of the slave node  300   a , an average number G CAVG  of processing blocks included in one GPU  404  can be calculated by Equation (1):
 
 G   CAVG   =G   C   /N   G   (1).
 
     Then, a conversion number GCI CAVG  of cores corresponding to the number of processing blocks included in the GPU  404  of the slave node  300   a  is calculated. The conversion number GCI CAVG  of cores may be calculated by a core table such as [Table 1] created based on the average number G CAVG  of processing blocks included in one GPU  404  and may be represented by an integer. 
     
       
         
           
               
               
               
             
               
                   
                 TABLE 1 
               
               
                   
                   
               
               
                   
                   
                 Conversion number 
               
               
                   
                 Number of Processing Blocks 
                 of Cores (GCI CAVG ) 
               
               
                   
                   
               
             
            
               
                   
                 Greater than 0 to 96 
                 1 
               
               
                   
                 Greater than 96 to 192 
                 2 
               
               
                   
                 Greater than 192 to 384 
                 3 
               
               
                   
                 Greater than 384 to 512 
                 4 
               
               
                   
                 Greater than 512 to 1024 
                 5 
               
               
                   
                 Greater than 1024 to 2048 
                 6 
               
               
                   
                   
               
            
           
         
       
     
     In some embodiments, the conversion number GCI CAVG  of cores calculated as described above can be corrected by reflecting a weight allocated to the GPU  404  according to attributes of job data. For example, the processing speed of the CPU  402  and the processing speed of the GPU  404  for the same job data may be compared. Then, a speed improvement index PI indicating the degree of improvement in the processing speed of the GPU  404  may be reflected in the conversion number GCI CAVG  of cores. If the GPU  404  requires a time of 10 to process a job when the CPU  402  requires a time of 1000 to process the job, the degree of improvement in the processing speed of the GPU  404  may be 100 times (100×). The speed improvement index PI representing this concept may be calculated by [Table 2] and represented by an integer. 
     
       
         
           
               
               
               
             
               
                   
                 TABLE 2 
               
               
                   
                   
               
               
                   
                   
                 Speed Improvement 
               
               
                   
                 Speed Improvement 
                 Index (PI) 
               
               
                   
                   
               
             
            
               
                   
                  0-50x 
                 1 
               
               
                   
                  50x-100x 
                 2 
               
               
                   
                 100x-150x 
                 3 
               
               
                   
                 150x-200x 
                 4 
               
               
                   
                 200x-300x 
                 5 
               
               
                   
                 &gt;300x 
                 6 
               
               
                   
                   
               
            
           
         
       
     
     Based on [Table 2], the corrected conversion number N Peq  of cores can be calculated by Equation (2):
 
 N   Peq   =GCI   CAVG   ×PI   (2).
 
     Assuming that the number of cores included in the CPU  402  of the slave node  300   a  is N p , a total number N′ p  of cores (hereinafter, referred to as a adding up number N′ p  of cores) required to process a job can be calculated by Equation (3):
 
 N′   p   =N   p   +N   Peq   (3).
 
     In this way, the job data block  600   a  is split into the first data block  602  and the second data block  604 . Here, the size of the first data block  602  corresponds to the number of cores included in the CPU  402  of the slave node  300   a , and the size of the second data block  604  corresponds to the conversion number of cores. 
       FIG. 9  illustrates a process of distributing input data to a CPU and a GPU according to an embodiment of the present invention. 
     Referring to  FIG. 9 , according to the current embodiment of the present invention, a job data block  600   a  may be split into a number of job units data equal to the adding up number of cores. Then, a number of job units data equal to the number of cores included in a CPU  402  may be allocated to the CPU  402  of a slave node  300   a , and a number of job units data equal to the conversion number of cores may be allocated to a GPU  404  of the slave node  300   a . In some embodiments of the present invention, the slave node  300   a  may include a plurality of GPUs  404 . In this case, the job units data equal to the conversion number of cores may be split into a plurality of data blocks whose sizes are respectively proportional to the numbers of processing blocks included in the GPUs  404 , and the data blocks may be allocated to the GPUs  404 , respectively. 
       FIG. 10  illustrates a process of distributing input data to a CPU and a GPU according to anther embodiment of the present invention. 
     Referring to  FIG. 10 , in the current embodiment of the present invention, one of cores included in a CPU  402  of a slave node  300   a  may be a GPU control core  412 ′ which performs a control operation of allocating job units data to a GPU  404 . The GPU control core  412 ′ may be excluded from the number of cores included in the CPU  402  and may not be allocated with a job data unit. In the current embodiment, one of the cores included in the CPU  402  of the slave node  300   a  may be a job units data distribution core  417 ′ which performs a control operation of allocating job units data to other CPU cores. 
       FIG. 11  is a flowchart illustrating a data processing method according to an embodiment of the present invention. 
     Referring to  FIG. 11 , the data processing method according to the current embodiment may include calculating an average number G CAVG  of processing blocks included in one GPU  404  (operation S 800 ), calculating a conversion number GCI CAVG  of cores corresponding to the number of processing blocks included in the GPU  404  of a slave node  300   a  based on the average number G CAVG  of processing blocks included in one GPU  404  (operation S 802 ), calculating a speed improvement index PI indicating the degree of improvement in the processing speed of the GPU  404  by comparing the processing speed of a CPU  402  and the processing speed of the GPU  404  for the same job data (operation S 804 ), calculating the corrected conversion number N Peq  of cores by reflecting the conversion number GCI CAVG  of cores and the speed improvement index PI (operation S 806 ), and distributing a job data block between the CPU  402  and the GPU  404  (operation S 808 ). 
     In some embodiments of the present invention, the data processing method may include distributing the job data block, which has been distributed between the CPU  402  and the GPU  404 , to each core of the CPU  402  and the GPU  404  (operation S 810 ). Specifically, a job data block  600   a  may be split into a number of job units data equal to the adding up number of cores. Then, a number of job units data equal to the number of cores included in the CPU  402  may be allocated to the CPU  402  of the slave node  300   a , and a number of job units data equal to the conversion number of cores may be allocated to the GPU  404  of the slave node  300   a . In some embodiments of the present invention, the slave node  300   a  may include a plurality of GPUs  404 . In this case, the job units data equal to the conversion number of cores may be split into a plurality of data blocks whose sizes are respectively proportional to the numbers of processing blocks included in the GPUs  404 , and the data blocks may be allocated to the GPUs  404 , respectively. 
       FIG. 12  is a flowchart illustrating a data processing method according to another embodiment of the present invention. 
     Referring to  FIG. 12 , the data processing method according to the current embodiment may include calculating a conversion number of first-type processors corresponding to the number of second-type processors included in a node of a distributed system (operation S 900 ), calculating a adding up number of first-type processors by adding up the number of first-type processors included in the node of the distributed system and the conversion number of first-type processors (operation S 902 ), splitting job data allocated to the node of the distributed system into a number of job units data equal to the adding up number of first-type processors (operation S 904 ), allocating a number of job units data equal to the number of first-type processors included in the node of the distributed system to the first-type processors of the node of the distributed system (operation S 906 ), and allocating a number of job units data equal to the conversion number of first-type processors to the second-type processors of the node of the distributed system (operation S 908 ). 
     Here, the job units data equal to the conversion number of the first-type processors are split into a plurality of data blocks whose sizes are respectively proportional to throughputs of the second-type processors per unit of time, and the data blocks are allocated to the second-type processors, respectively. 
     The method may further include obtaining first result data by processing some of the job units data using the first-type processors and obtaining second result data by processing the other ones of the job units data using the second-type processors (operation S 910 ) and producing final result data by collecting the first result data and the second result data (operation S 912 ). 
     According to various embodiments of the present invention, there is no need to add a new node to a cluster in order to implement distributed computing. Instead, the processing speed of a large amount of data (e.g., big data) can be increased by using a GPU included in an existing node as a general purpose GPU (GPGPU). Furthermore, a policy for distributing a job between a CPU and a GPU according to various embodiments of the present invention can be effectively used without an additional modification even when a new GPU node is added to the cluster. 
     While the present invention has been particularly shown and described with reference to exemplary embodiments thereof, it will be understood by those of ordinary skill in the art that various changes in form and detail may be made therein without departing from the spirit and scope of the present invention as defined by the following claims. The exemplary embodiments should be considered in a descriptive sense only and not for purposes of limitation.