Abstract:
Disclosed are a data processing system, etc., whereby total processing time of a data process including a parallel process and a serial process is optimized. This data processing system comprises: a task segmenting means for receiving a plurality of tasks which execute, upon a record set which is a set of a plurality of records, a data process which includes parallel processes which a plurality of computers are capable of executing in parallel and serial processes which are post-processes of the parallel processes and is executed by one computer, and, for a parallel process which is the pre-process of a serial process, among the serial processes which are included in the task, which is envisioned as requiring more time to process than another serial process, by subdividing the record set which is to be processed in the parallel process, segmenting the parallel process into a plurality of sub-tasks; and an assignment means for assigning the plurality of sub-tasks to the plurality of computers, such that the plurality of sub-tasks are executed prior to a task other than the task which includes the parallel processes being executed.

Description:
TECHNICAL FIELD 
       [0001]    The present invention relates to a data processing system in a decentralized processing, a data processing method, and a program. 
       BACKGROUND ART 
       [0002]    In a data processing using a computer, a parallel processing technology is generally well known. Patent literature 1 discloses a technology for performing a parallel processing in which a new idea about a hash join is created. In the technology disclosed in patent literature 1, the hash join is divided into a plurality of tasks on the basis of a frequency of occurrence of a join key of a table and a plurality of the tasks are assigned to a plurality of computers. In the technology disclosed in patent literature 1, in a case in which the frequency of occurrence of the join key is biased and whereby, the processing amounts of the tasks are not uniform, the task with a large processing amount is divided into a plurality of small sub tasks. In the technology disclosed in patent literature 1, the task and the sub task are assigned to a plurality of computers so that the processing amounts of the computers are equal to each other. 
       CITATION LIST 
     Patent Literature 
       [0003]    [PTL 1] Japanese Patent Publication No. 1995-111718 
       SUMMARY OF INVENTION 
     Technical Problem 
       [0004]    However, the technology disclosed in patent literature 1 is only a technology for optimizing the processing time of the parallel processing. The technology disclosed in patent literature 1 has a problem in which when the technology is applied to a data processing including both a parallel processing and a serial processing, the processing time of the whole data processing is not necessarily optimized. 
         [0005]    An object of the present invention is to provide a data processing system which can optimize the processing time of the whole data processing including the parallel processing and the serial processing. 
       Solution to Problem 
       [0006]    A first invention that solves the above-mentioned problem is a data processing system including task dividing means for receiving a plurality of tasks in which a data processing including a parallel processing that can be performed by a plurality of computers in parallel and a serial processing that is a post-processing of the parallel processing and performed by one computer is performed to a record set that is a set of a plurality of records and dividing the parallel processing that is a preprocessing of the serial processing to which it is expected that the processing time thereof is greater than that of the serial processing included in another task into a plurality of sub tasks by subdividing the record set that is a processing target of the parallel processing and assignment means for assigning a plurality of the sub tasks to a plurality of the computers so that a plurality of the sub tasks are performed before the task other than the task including the parallel processing is performed. 
         [0007]    A second invention that solves the above-mentioned problem is a data processing method including a division step in which a plurality of tasks in which a data processing including a parallel processing that can be performed by a plurality of computers in parallel and a serial processing that is a post-processing of the parallel processing and performed by one computer is performed to a record set that is a set of a plurality of records are received and the parallel processing that is a preprocessing of the serial processing to which it is expected that the processing time thereof is greater than that of the serial processing included in another task is divided into a plurality of sub tasks by subdividing the record set that is a processing target of the parallel processing and an assignment step in which a plurality of the sub tasks are assigned to a plurality of the computers so that a plurality of the sub tasks are performed before the task other than the task including the parallel processing is performed. 
         [0008]    A third invention that solves the above-mentioned problem is a program which causes a computer to perform a division process in which a plurality of tasks in which a data processing including a parallel processing that can be performed by a plurality of computers in parallel and a serial processing that is a post-processing of the parallel processing and performed by one computer is performed to a record set that is a set of a plurality of records are received and the parallel processing that is a preprocessing of the serial processing to which it is expected that the processing time thereof is greater than that of the serial processing included in another task is divided into a plurality of sub tasks by subdividing the record set that is a processing target of the parallel processing and an assignment process in which a plurality of the sub tasks are assigned to a plurality of the computers so that a plurality of the sub tasks are performed before the task other than the task including the parallel processing is performed. 
         [0009]    The object of the present invention can be achieved by a computer-readable storage medium storing the above-mentioned program. 
       Advantageous Effects of Invention 
       [0010]    By using the present invention, the processing time of the whole data processing including the parallel processing and the serial processing can be optimized. 
     
    
     
       BRIEF DESCRIPTION OF DRAWINGS 
         [0011]      FIG. 1  is a hardware configuration diagram of a data processing system  1000  of the present invention. 
           [0012]      FIG. 2  is a block diagram of a data processing system  1000  according to a first exemplary embodiment of the present invention. 
           [0013]      FIG. 3  is a figure for explaining a concept of a task in the present invention. 
           [0014]      FIG. 4  is a figure for explaining a concept of a sub task in the present invention. 
           [0015]      FIG. 5  is a figure for explaining a concept of a task assignment in the present invention. 
           [0016]      FIG. 6  is a flowchart for explaining operation of a data processing system  1000  of the present invention. 
           [0017]      FIG. 7  is a block diagram of a data processing system  1000  according to a second exemplary embodiment of the present invention. 
           [0018]      FIG. 8  is a flowchart for explaining operation of a data processing system  1000  according to a second exemplary embodiment of the present invention. 
           [0019]      FIG. 9  is a figure for explaining operation of a task division unit  10  according to a third exemplary embodiment of the present invention. 
           [0020]      FIG. 10  is a figure for explaining operation of a task division unit  10  according to a third exemplary embodiment of the present invention. 
           [0021]      FIG. 11  is a figure for explaining operation of a task division unit  10  according to a third exemplary embodiment of the present invention. 
           [0022]      FIG. 12  is a figure for explaining operation of a task division unit  10  and an assignment unit  20  according to a fourth exemplary embodiment of the present invention. 
           [0023]      FIG. 13  is a figure for explaining operation of a task division unit  10  and an assignment unit  20  according to a fourth exemplary embodiment of the present invention. 
           [0024]      FIG. 14  is a figure for explaining an effect of a fourth exemplary embodiment of the present invention. 
           [0025]      FIG. 15  is a figure for explaining operation of an assignment unit  20  according to a fifth exemplary embodiment of the present invention. 
           [0026]      FIG. 16  is a figure for explaining an effect of a fifth exemplary embodiment of the present invention. 
           [0027]      FIG. 17  is a figure for explaining an effect of a fifth exemplary embodiment of the present invention. 
           [0028]      FIG. 18  is a figure for explaining an effect of a fifth exemplary embodiment of the present invention. 
           [0029]      FIG. 19  is a figure for explaining a specific example of a table, a record set, and a record of the present invention. 
           [0030]      FIG. 20  is a figure for explaining a specific example of statistical information of the present invention. 
           [0031]      FIG. 21  is a block diagram of a data processing system  1000  according to a sixth exemplary embodiment of the present invention. 
           [0032]      FIG. 22  is a block diagram of a data processing system  1000  according to a sixth exemplary embodiment of the present invention. 
           [0033]      FIG. 23  is a figure for explaining a deadline in a seventh exemplary embodiment of the present invention. 
           [0034]      FIG. 24  is a figure for explaining a deadline in a seventh exemplary embodiment of the present invention. 
           [0035]      FIG. 25  is a flowchart for explaining operation of a data processing system  1000  according to a seventh exemplary embodiment of the present invention. 
           [0036]      FIG. 26  is a block diagram of a data processing system  1000  according to an eighth exemplary embodiment of the present invention. 
           [0037]      FIG. 27  is a block diagram of a task execution unit  200  according to an eighth exemplary embodiment of the present invention. 
           [0038]      FIG. 28  is a figure for explaining a specific example of schedule data in an eighth exemplary embodiment of the present invention. 
           [0039]      FIG. 29  is a figure for explaining a specific example of schedule data in an eighth exemplary embodiment of the present invention. 
           [0040]      FIG. 30  is a flowchart for explaining operation of a task execution unit  200  according to an eighth exemplary embodiment of the present invention. 
           [0041]      FIG. 31  is a figure for explaining a specific example of schedule data in an eighth exemplary embodiment of the present invention. 
           [0042]      FIG. 32  is a figure for explaining operation of a task execution unit  200  in a modification example of an eighth exemplary embodiment of the present invention. 
           [0043]      FIG. 33  is a block diagram of a data processing system  1000  according to a ninth exemplary embodiment of the present invention. 
           [0044]      FIG. 34  is a block diagram of a task execution unit  400  according to a ninth exemplary embodiment of the present invention. 
           [0045]      FIG. 35  is a figure for explaining a specific example of a shuffle key in a ninth exemplary embodiment of the present invention. 
       
    
    
     DESCRIPTION OF EMBODIMENTS 
       [0046]    The data processing that is a target of the present invention includes a parallel processing and a serial processing. The parallel processing is a process which can be performed by a plurality of computers in parallel. The serial processing is performed as a post-processing of the parallel processing. 
         [0047]    In the parallel processing, when the number of the computers is increased, the processing time is reduced. On the other hand, the serial processing is a processing performed by one computer. Even if the number of the computers is increased, the processing time of the serial processing is not reduced. For this reason, when a plurality of computers perform the data processing including both the parallel processing and the serial processing, the processing time of the serial processing tends to become a bottleneck of the processing time of the whole data processing. 
         [0048]    Next, a definition of terms used for explaining an exemplary embodiment of the present invention and their relations will be given below. A “table” is a data that is a target to which the data processing is performed. The table includes a plurality of record sets. A “record set” is a set of a plurality of records and a unit in which the serial processing is performed by one computer. A “job” is a data processing to the table. A “task” is a data processing to the record set. The job is divided into a plurality of tasks when the table of the processing target is subdivided into a plurality of record sets. 
         [0049]    The task can be divided into two tasks: a parallel processable task and a serial processing task. The “parallel processable task” is a processing in which the parallel processing is performed to the record set. The “serial processing task” is a processing in which the serial processing that is the post-processing of the parallel processing is performed to the record set. 
         [0050]    The parallel processable task can be divided into a plurality of “sub tasks” when the record set that is the processing target thereof is subdivided into a set of small records. Namely, one task can be divided into a plurality of sub tasks and one serial processing task. 
         [0051]    “The sub task corresponds to the serial processing task” means that the sub task and the serial processing task are obtained when the same task is divided. A “large task” means a task of which the number of records included in the record set that is an execution target is large. The processing time of the task is basically determined by the number of records included in the record set that is the processing target. The large task is the task whose processing time is long. 
         [0052]    Next, an exemplary embodiment of the present invention will be described in detail with reference to the drawing. 
       First Exemplary Embodiment 
       [0053]      FIG. 1  is a hardware configuration diagram of a data processing system  1000  according to a first exemplary embodiment. As shown in  FIG. 1 , the data processing system  1000  includes for example, a CPU (Central Processing Unit)  1 , a RAM (Random Access Memory)  2 , a hard disk drive  3 , a communication interface  4 , an input device  5 , an output device  6 , and the like. 
         [0054]    For example, the CPU  1  reads a program, stores it in the RAM  2 , and executes it and whereby, the function of the data processing system  1000  can be realized. An application program controls the communication interface  4  by using for example, a function provided by an OS (Operating System) and whereby, transmission and reception of information by the data processing system  1000  can be realized. The input device  5  is for example a keyboard or a mouse. The output device  6  is for example, a display. Further, the input device  5  and the output device  6  may be integrated as for example, a touch panel. 
         [0055]    Further, the present invention that is explained by using this exemplary embodiment or each exemplary embodiment described later as an example may be configured as a non-volatile storage medium  8  such as a compact disc or the like storing the program. The program stored in the storage medium  8  is read by for example, a drive device  7 . 
         [0056]    The data processing system  1000  may be configured by using two or more devices that are physically separated from each other and communicably connected by wire or wireless. 
         [0057]      FIG. 2  is a block diagram of the data processing system  1000  according to the first exemplary embodiment. As shown in  FIG. 2 , the data processing system  1000  includes a task division unit  10  and an assignment unit  20 . 
         [0058]    In the data processing system  1000 , the task division unit  10  receives a request for execution of a plurality of tasks. The task division unit  10  calculates the processing time of the serial processing task included in the task for each of a plurality of the received tasks. For example, the task division unit  10  calculates the processing time of the task on the basis of the number of the records included in the record set that is the processing target of the task. The task division unit  10  divides the parallel processable task that is the preprocessing of the serial processing task larger than another serial processing task into a plurality of sub tasks. 
         [0059]    The assignment unit  20  assigns a plurality of the sub tasks to a plurality of computers so that a plurality of the sub tasks are performed before the task other than the task including the divided parallel processable task is performed. 
         [0060]      FIG. 3  is a figure for explaining a concept of a task performed by the data processing system  1000 . Three rectangles are shown in  FIG. 3 . Each of three rectangles represents a task A, a task B, or a task C. Namely,  FIG. 3  shows three tasks of the task A, the task B, and the task C. Each task includes the parallel processable task (the part with a horizontal-stripe pattern) and the serial processing task (the part filled in white). The length of the rectangle (the task) shown in  FIG. 3  represents the processing time of the task when the task is performed by a single computer. It is expected that the processing time is basically proportional to the size of the record set processed by the task. As shown in  FIG. 3 , the serial processing task of the task A is larger than the serial processing task of the task B or the serial processing task of the task C. 
         [0061]      FIG. 4  is a figure for explaining a state in which the parallel processable task of the task A shown in  FIG. 3  is divided into the sub-tasks. As shown in  FIG. 4 , the task division unit  10  divides the parallel processable task of the task A into a plurality of sub tasks (the part with a grid pattern). Specifically, the task division unit  10  divides the parallel processable task into the sub tasks whose number is equal to the number of the computers which perform the job. For example, when the job is performed by three computers, the task division unit  10  divides the parallel processable task of the task A into three sub tasks. 
         [0062]      FIG. 5  is a figure for explaining a state in which the task shown in  FIG. 4  is assigned to three computers. As shown in  FIG. 5 , the assignment unit  20  assigns the sub task of the task A, the serial processing task of the task A, the task B, and the task C to three computers. The assignment unit  20  performs the assignment so that the sub task of the task A is performed before the task B and the task C are performed. 
         [0063]    As shown in  FIG. 5 , the computer  1  performs the sub task of the task A and then performs the serial processing task of the task A. The computer  2  performs the sub task of the task A, transmits a result of the execution to the computer  1 , and then, performs the task B. The computer  3  performs the sub task of the task A, transmits a result of the execution to the computer  1 , and then, performs the task C. 
         [0064]      FIG. 6  is a flowchart for explaining the operation of the data processing system  1000 . As shown in  FIG. 6 , the task division unit  10  receives a request for execution of a plurality of tasks (A 100 ). The task division unit  10  calculates the processing time of the serial processing task included in each task on the basis of the number of the records included in the record set that is the processing target of each task for which the execution request is received (A 110 ). 
         [0065]    The task division unit  10  divides the task including the large serial processing task into the parallel processable task and the serial processing task and further divides the parallel processable task into a plurality of sub tasks (A 120 ). The assignment unit  20  assigns the task, the sub task, and the serial processing task to a plurality of computers so that a plurality of sub tasks are performed before the task other than the task including the large serial processing task is performed. 
         [0066]    As described above, the data processing system  1000  according to the exemplary embodiment can provide a data processing system which optimizes the processing time of the whole data processing including the parallel processing which can be performed by a plurality of computers in parallel and the serial processing which is the post-processing of the parallel processing and performed by one computer, a data processing method, and a program. The reason is because the data processing system  1000  schedules the execution of a job so that the large serial processing task that tends to become a bottleneck of the whole data processing is performed as soon as possible. As a result, the large serial processing task and another task are performed concurrently. Therefore, the processing time of the whole data processing can be optimized. 
       Second Exemplary Embodiment 
       [0067]      FIG. 7  is a block diagram of the data processing system  1000  according to a second exemplary embodiment. As shown in  FIG. 7 , the data processing system  1000  according to the second exemplary embodiment further includes a job reception unit  30  and a job division unit  40  in addition to the data processing system  1000  according to the first exemplary embodiment. 
         [0068]    The job reception unit  30  receives a request for execution of the job. The job division unit  40  divides the job into a plurality of tasks by subdivides a table that is a processing target of the job into a plurality of record sets. The task division unit  10  receives a plurality of tasks from the job division unit  40  and performs the operation like the first exemplary embodiment. 
         [0069]      FIG. 8  is a flowchart explaining the operation of the data processing system  1000 . Further, the same reference number is used for the processing shown in  FIG. 8  in which the substantially same operation is performed as the processing shown in  FIG. 6 . As shown in  FIG. 8 , the job reception unit  30  receives the request for execution of the job (B 100 ). The job division unit  40  divides the job for which the execution request is received into a plurality of tasks (B 110 ). After this process, the processing of steps A 110  to A 130  is performed like the first exemplary embodiment. 
         [0070]    As described above, the data processing system  1000  according to the exemplary embodiment can provide a data processing system which optimizes the processing time of the whole data processing including the parallel processing which can be performed by a plurality of computers in parallel and the serial processing which is the post-processing of the parallel processing and performed by one computer, a data processing method, and a program. The reason is because the data processing system  1000  schedules the execution of the job so that the large serial processing task that tends to become the bottleneck of the whole data processing is performed as soon as possible. As a result, the large serial processing task and another task are performed concurrently. Therefore, the processing time of the whole data processing can be optimized. 
       Third Exemplary Embodiment 
       [0071]    The operation of the task division unit  10  in the data processing system  1000  according to a third exemplary embodiment is different from the operation of the task division unit  10  according to the first or second exemplary embodiment. The task division unit  10  divides the parallel processable task included in the task with respect to the task larger than the largest serial processing task. The task division unit  10  does not divide the parallel processable task included in the task with respect to the task smaller than the largest serial processing task. 
         [0072]    The operation of the task division unit  10  according to the third exemplary embodiment will be described by using a case in which the job that can be divided into three tasks is performed by N computers as an example.  FIG. 9 ,  FIG. 10 , and  FIG. 11  are figures for explaining the operation of the task division unit  10  according to the third exemplary embodiment. 
         [0073]      FIG. 9  shows three tasks (the task A, the task B, and the task C). As shown in  FIG. 9 , the largest serial processing task is the serial processing task of the task A. The task division unit  10  compares the processing times of the tasks A to C with the processing time of the serial processing task of the task A. The task A and the task B are larger than the serial processing task of the task A. Accordingly, the task division unit  10  divides the parallel processable task of the task A into N sub tasks and divides the parallel processable task of the task B into N sub tasks. On the other hand, the processing time of the task C is shorter than the processing time of the serial processing task of the task A. Therefore, the task division unit  10  does not divide the parallel processable task of the task C. 
         [0074]      FIG. 10  is a figure for explaining a state in which the parallel processable task of the task A and the parallel processable task of the task B shown in  FIG. 9  are divided into the sub tasks. As shown in  FIG. 10 , the parallel processable task of the task A is divided into N sub tasks and the parallel processable task of the task B is divided into N sub tasks. Hereinafter, the sub task obtained by dividing the parallel processable task of the task A is referred to as “the sub task of the task A”. The sub task obtained by dividing the parallel processable task of the task B is referred to as “the sub task of the task B”. 
         [0075]      FIG. 11  is a figure showing a state in which the tasks shown in  FIG. 10  are assigned to N computers. As shown in  FIG. 11 , an assignment unit  21  assigns the sub task of the task A, the serial processing task of the task A, the sub task of the task B, the serial processing task of the task B, and task C to a plurality of computers  1  to N. 
         [0076]    The reason that the task division unit  10  does not divide the task C is as follows. As shown in  FIG. 11 , the bottleneck of the processing time of the whole job is the processing time of the serial processing task of the task A. Even when the number of computers is increased, the processing time of the serial processing task of the task A cannot be reduced. Therefore, even if the task division unit  10  performs parallelization of the task C whose processing time is shorter than that of the serial processing task of the task A, the processing time of the whole job cannot be reduced. 
         [0077]    With respect to the task smaller than the largest serial processing task, the task division unit  10  according to the third exemplary embodiment does not divide the parallel processable task included in the task. The data processing system  1000  according to the third exemplary embodiment has such configuration. Accordingly, the unnecessary parallelization is prevented and the overhead due to unnecessary parallelization can be prevented from being generated. 
         [0078]    Further, when the task A and the task B are larger than the serial processing task of the task A and the task A is sufficiently larger than the task B, the task division unit  10  may perform the division so that the number of the divisions of the parallel processable task of the task B is smaller than the number of the divisions of the parallel processable task of the task A. At this time, the assignment unit  20  makes a schedule so as not to assign the sub task of the task B to the computer which performs the serial processing task of the task A. 
         [0079]    Specifically, when the total processing time of one sub task of the task A and the serial processing task of the task A is longer than the total processing time of one sub task of the task A, one sub task of the task B, and the serial processing task of the task B, the task division unit  10  sets the number of the divisions so that the number of the divisions of the parallel processable task of the task B of the task B is smaller than the number of the divisions of the parallel processable task of the task A. 
         [0080]    For example, when the parallel processable task of the task A is divided into N sub tasks and the parallel processable task of the task B is divided into (N−1) sub tasks, the assignment unit  20  does not assign the sub task of the task B to the computer (the computer  1 ) which performs the serial processing task of the task A. When such schedule is made, the processing time of the serial processing task of the task A that is the bottleneck of the processing time of the whole job can be further reduced and the execution time of the whole job can be further reduced. 
       Fourth Exemplary Embodiment 
       [0081]    The operations of the task division unit  10  and the assignment unit  20  in the data processing system  1000  according to a fourth exemplary embodiment are different from those of the task division unit  10  and the assignment unit  20  in the data processing system  1000  according to the first, second, or third exemplary embodiment. When the task division unit  10  divides the parallel processable task of the task A into a plurality of sub tasks, the task division unit  10  divides it so that the number of the records processed in each sub task is different from each other. The assignment unit  20  assigns the sub task of which the number of the records processed therein is greater than the number of the records processed in another sub task to the computer to which the serial processing task that is the post-processing of the parallel processable task is assigned. The assignment unit  20  may not assign the sub task to the computer other than this computer. 
         [0082]      FIG. 12 ,  FIG. 13 , and  FIG. 14  are figures for explaining the operations of the task division unit  10  and the assignment unit  20  according to the fourth exemplary embodiment. As shown in  FIGS. 12 to 14 , in this example, a case in which a job which can be divided into five tasks is performed by four computers is considered. 
         [0083]      FIG. 12  shows five tasks (the tasks A to E). As shown in  FIG. 12 , the serial processing task of the task A is larger than the serial processing tasks of the task B to the task E. The task division unit  10  divides the parallel processable task of the task A into a plurality of sub tasks. 
         [0084]      FIG. 13  is a figure showing a state in which the task division unit  10  divides the parallel processable task of the task A shown in  FIG. 12  into the sub tasks wherein the division is performed so that the numbers of the records processed in all the sub tasks are different from each other and the assignment unit  20  assigns the sub task of which the number of the records processed therein is greater than the number of the records processed in another sub task to the computer (the computer  1 ) to which the serial processing task of the task A is assigned. In an example shown in  FIG. 13 , the parallel processable task of the task A is divided into three sub tasks and three sub tasks are assigned to the computer  1 , the computer  2 , and the computer  3 , respectively. In an example shown in  FIG. 13 , the size of the sub task assigned to the computer  1  is two times larger than the size of the sub task assigned to the computer  2  and the computer  3 . 
         [0085]    The assignment unit  20  selects the computer to which the sub task is not assigned from among the computers other than the computer (the computer  1 ) to which the serial processing task that is the post-processing of the parallel processable task is assigned. In an example shown in  FIG. 13 , the sub task is not assigned to the computer  4 . Therefore, the computer  4  does not need to transfer a result of the execution of the sub task to the computer  1 . It is desirable that the assignment unit  20  select the computer whose data transfer cost to transfer data to the computer (the computer  1 ) which performs the serial processing is high as the computer to which the sub task is not assigned. 
         [0086]      FIG. 14  is a figure for explaining an effect of this exemplary embodiment. Here, it is assumed that the parallel processable task of the task A shown in  FIG. 12  is divided into sub tasks wherein the division is performed so that the numbers of the records processed in all the sub tasks are equal to each other. As shown in  FIG. 14 , the parallel processable task of the task A is divided into four sub tasks and four sub tasks are assigned to the computers  1  to  4 , respectively. As shown in  FIG. 14 , the sub task of the task A, the serial processing task of the task A, and the tasks B to E are assigned to four computers (the computers  1  to  4 ). When the computers  2 ,  3  and  4  perform the sub task of the task A, the computers  2 ,  3  and  4  have to transfer a result of the execution of the sub task to the computer  1 . In this case, the data communication cost of the computer  4  increases in comparison with a case in which the tasks are scheduled as shown in  FIG. 13 . 
         [0087]    Thus, the task division unit  10  according to the fourth exemplary embodiment divides the parallel processable task wherein the division is performed so that the numbers of the records processed in all the sub tasks are different from each other. The assignment unit  20  according to the fourth exemplary embodiment assigns the sub task of which the number of the records processed therein is greater than the number of the records processed in another sub task to the computer to which the serial processing task that is the post-processing of the parallel processable task is assigned. The assignment unit  20  selects the computer to which the sub task is not assigned from among the computers other than the computer to which the serial processing task that is the post-processing of the parallel processable task is assigned. The data processing system  1000  according to the fourth exemplary embodiment has such configuration. Therefore, the data processing system  1000  according to the fourth exemplary embodiment can suppress the data transfer cost of the computer to which the sub task is not assigned. 
       Fifth Exemplary Embodiment 
       [0088]    The operations of the task division unit  10  and the assignment unit  20  in the data processing system  1000  according to a fifth exemplary embodiment are different from those of the task division unit  10  and the assignment unit  20  in the data processing system  1000  according to the first, second, third, or fourth exemplary embodiment. The data processing system  1000  according to this exemplary embodiment performs the serial processing task in which the records included in the record set are processed in a predetermined order. 
         [0089]    In the data processing system  1000 , when the task division unit  10  receives a task execution instruction, the task division unit  10  receives information in which a processing order of the records is determined. The task division unit  10  calculates a processing execution time of the serial processing task included in the task on the basis of the number of the records included in the record set that is the processing target of the task. The task division unit  10  divides the parallel processable task that is the preprocessing of the serial processing task larger than another serial processing task into a plurality of sub tasks. The task division unit  10  subdivides the record set into a set of small records according to the processing order of the records. 
         [0090]    For example, when a task in which two hundred records that are the processing targets are processed is performed by four computers, the task division unit  10  divides the record set into four sets, each of which includes fifty records. In this case, the task division unit  10  sets the records from a first processed record to a fiftieth processed record to a set of the records that is processed in a first sub task. The task division unit  10  sets the records from a fifty-first sequentially processed record to a one hundredth sequentially processed record to a set of the records that is processed in a second sub task. The task division unit  10  sets the records from a one hundred-first sequentially processed record to a one hundredth fiftieth sequentially processed record to a set of the records that is processed in a third sub task. The task division unit  10  sets the records from a one hundred fifty-first sequentially processed record to a two hundredth sequentially processed record to a set of the records that is processed in a fourth sub task. 
         [0091]    The assignment unit  20  assigns the sub task that includes the first record processed in the serial processing task as the processing target to the computer to which the serial processing task that is the post-processing of the parallel processable task is assigned. 
         [0092]      FIG. 15  is a figure showing a state in which the assignment unit  20  assigns the sub task that includes the first record processed in the serial processing task as the processing target to the computer (the computer  1 ) to which the serial processing task is assigned.  FIG. 16  is a figure for explaining an effect of the fifth exemplary embodiment. In  FIG. 15  and  FIG. 16 , the sub task assigned to the computer  1  is “the sub task (represented by a in  FIGS. 15 to 18 ) that includes the first processed record as the processing target”. In  FIG. 15 , the sub task assigned to the computers  2 ,  3 , and  4  is “the sub task (represented by ( 3  in  FIGS. 15 to 18 ) that does not include the first processed record as the processing target”. 
         [0093]      FIG. 16  shows a state in which when the schedule is made as shown in  FIG. 15 , a delay of the execution of the sub task performed by the computer  2  occurs. As shown in  FIG. 16 , the delay of the computer  2  does not influence the processing time of the whole job. That is because when the computer  1  completes the execution of the sub task that includes the first processed record as the processing target, the computer  1  can start the execution of the serial processing task. During a period in which the computer  1  performs the serial processing task, the computer  2  completes the execution of the sub task assigned to the computer  2  and transmits a result of the execution to the computer  1 . In this case, the delay of the execution of the sub task performed by the computer  2  does not influence the processing time of the whole job. 
         [0094]      FIG. 17  and  FIG. 18  are figures for explaining an effect of the fifth exemplary embodiment. In  FIG. 17  and  FIG. 18 , “the sub task that includes the first processed record as the processing target” is represented by a and “the sub task that does not include the first processed record as the processing target” is represented by β.  FIG. 17  is a figure showing a state in which the assignment unit  20  does not assign the sub task that includes the first processed record as the processing target to the computer (the computer  1 ) to which the serial processing task is assigned. As shown in  FIG. 17 , the sub task that includes the first processed record as the processing target is assigned to the computer  2 . 
         [0095]      FIG. 18  shows a state in which when the schedule is made as shown in  FIG. 17 , the delay of the execution of the sub task performed by the computer  2  occurs. As shown in  FIG. 18 , the processing time of the whole job is delayed by the delay of the computer  2 . That is because the computer  1  cannot start to perform the serial processing task before the computer  2  completes the execution of the sub task. 
         [0096]    Thus, the assignment unit  20  according to the fifth exemplary embodiment assigns the sub task that includes the first record processed in the serial processing task as the processing target to the computer to which the serial processing task that is the post-processing of the parallel processable task is assigned. The data processing system  1000  according to the fifth exemplary embodiment has such configuration. Therefore, even when the delay occurs by influence of the re-execution of the task or the like due to for example, an error, a system failure, or the like, the start time of the serial processing task can be prevented from being influenced. 
         [0097]    Further, in the data processing system  1000  according to the fifth exemplary embodiment, the job reception unit  30  may receive an input of information in which the processing order of the records is determined instead of the task division unit  10 . 
       Sixth Exemplary Embodiment 
       [0098]    The data processing device  1000  according to a sixth exemplary embodiment performs a data processing including a JOIN processing and a scan processing that is a post-processing of the JOIN processing to a table. 
         [0099]      FIG. 19  is a figure showing an example of the table, the record set, and the record in the data processing device  1000  according to the sixth exemplary embodiment.  FIG. 19  is shown as an example for ease of understanding the explanation. Therefore, the table, the record set, and the record are not interpreted in a limited sense. 
         [0100]    As shown in an example of  FIG. 19 , the record includes a record identifier (ID), a branch office name, a customer ID, and transaction time. Here, the record ID, the branch office name, the customer ID, and the transaction time are attributes of the record. The attribute of the record includes an “attribute which determines a unit in which the serial processing is performed”. 
         [0101]    A value of the “attribute which determines a unit in which the serial processing is performed” determines the record set to which the record belongs. For example, when the attribute of the “branch office name” is the “attribute which determines a unit in which the serial processing is performed”, a set of the records of which the values of the branch office name are the same as each other is the record set. 
         [0102]    For example, in the parallel processing according to the exemplary embodiment, a transaction table shown in  FIG. 19  and a master table (not shown) are joined by using the branch office name as a key. For example, in the serial processing according to the exemplary embodiment, with respect to the joined record, the record is scanned in the record set of which the values of the branch office name are the same as each other in transaction time order. 
         [0103]      FIG. 20  shows an example of statistical information about the table shown in  FIG. 19  (hereinafter, described as statistical information). As shown in  FIG. 20 , the example of the statistical information is a histogram showing the number of records for each value of the attribute (in an example shown in  FIG. 20 , the branch office name) which determines a unit in which the serial processing is performed. The statistical information shown in  FIG. 20  is shown as an example for ease of understanding the explanation. Therefore, the statistical information is not interpreted in a limited sense. The statistical information may include for example, information about the total number of records, a bias of the number of records for each value of the attribute which determines a unit in which the serial processing is performed, or the like. 
         [0104]    Further, the record may include the attribute which determines an order of execution of the serial processing. For example, when the transaction time is the attribute which determines an order of execution of the serial processing, the serial processing may be performed in an increasing order of transaction time in the record set. The statistical information described above may include information on a distribution or the like for each value of the attribute which determines an order of execution of the serial processing. 
         [0105]      FIG. 21  is a block diagram of the data processing system  1000  according to the sixth exemplary embodiment. Further, the same reference numbers are used for the elements shown in  FIG. 21  which have the substantially same function as the elements shown in  FIG. 2  and  FIG. 7 . As shown in  FIG. 21 , the data processing system  1000  according to the sixth exemplary embodiment includes the task division unit  10 , a record information collection unit  110 , the assignment unit  20 , the job reception unit  30 , the job division unit  40 , and a data storage unit  300 . 
         [0106]    The job reception unit  30  receives a request for execution of the job. The execution request for the job includes information for specifying a content of the data processing and information for specifying the table that is the processing target. The information for specifying a content of the data processing includes for example, a parameter indicating a percentage of a parallel processable time of the job processing time. The job reception unit  30  outputs the information for specifying the table that is the processing target included in the execution request for the job to the record information collection unit  110 . 
         [0107]    The record information collection unit  110  acquires the statistical information from the data storage unit  300  on the basis of the information for specifying the table that is the processing target. The record information collection unit  110  may read the table stored in the data storage unit  300  and calculate the statistical information. Alternatively, the record information collection unit  110  may acquire the statistical information that is stored together with the table in the data storage unit  300  in advance. The record information collection unit  110  outputs the statistical information to the job division unit  40 . 
         [0108]    The job division unit  40  divides the job received by the job reception unit  30  into a plurality of tasks on the basis of the statistical information. For example, a case in which the job reception unit  30  receives the job in which the table shown in  FIG. 19  is used as the processing target will be explained. The table shown in  FIG. 19  includes 600 records. 
         [0109]    When a set of the records in which the values of the branch office name are the same as each other is the record set, the job division unit  40  divides the table including 600 records into the record sets: one record set including 200 records (Tokyo branch office), two record sets each of which includes 100 records (Sapporo branch office and Sendai branch office), two record sets each of which includes 50 records (Yokohama branch office and Osaka branch office), and four record sets each of which includes 25 records (Kyoto branch office, Kobe branch office, Hiroshima branch office, and Fukuoka branch office). 
         [0110]    The task division unit  10  receives the execution request for a plurality of the divided tasks. The task division unit  10  calculates the processing time of the task and the serial processing task included in the task on the basis of the information for specifying a content of the data processing and the statistical information. For example, the task division unit  10  inputs the number of the records included in the record set that is the processing target of the task and the information for specifying a content of the data processing received by the job reception unit  30  and calculates the processing time of the task. The task division unit  10  divides the task into a plurality of sub tasks and one serial processing task. 
         [0111]    The assignment unit  20  assigns a plurality of the sub tasks to a plurality of computers so that the sub tasks are performed before the corresponding serial processing task and another task are performed. 
         [0112]    The data processing system  1000  according to the sixth exemplary embodiment has such configuration. Therefore, a data processing system which optimizes the processing time of the job in which the data processing including the JOIN processing and the scan processing that is the post-processing of the JOIN processing is performed to the table, a data processing method, and a program can be provided. 
       Seventh Exemplary Embodiment 
       [0113]    The operations of a job division unit  41 , a task division unit  11  and the assignment unit  21  in the data processing device  1000  according to a seventh exemplary embodiment are different from the operations of these units according to the sixth exemplary embodiment. The data processing device  1000  according to the seventh exemplary embodiment divides the task and assigns the divided task by considering a deadline. 
         [0114]      FIG. 22  is a block diagram of the data processing system  1000  according to the seventh exemplary embodiment. Further, the same reference numbers are used for the elements shown in  FIG. 22  which have the substantially same function as the elements shown in  FIG. 21 . 
         [0115]    The job division unit  41  calculates the deadline. For example, the deadline is an average time obtained by dividing the processing time of the job by the number of the computers. When the processing time of the task exceeds the deadline, the task division unit  11  divides the task. The assignment unit  21  adjusts a position of the deadline as described later. 
         [0116]    The deadline will be explained below. In the explanation, a case in which the job which can be divided into four tasks is processed by four computers is taken as an example.  FIG. 23  is a figure showing a state in which four tasks (the tasks A to D) are assigned to four computers. The dotted line shown in  FIG. 23  indicates an average time (an average processing time) obtained by dividing the total processing time of all the tasks (A to D) by the number (four) of the computers. When the job is ideally scheduled, the execution of the job is completed within this average processing time. Accordingly, a job division unit  40 A sets this average processing time as the deadline. 
         [0117]    A task division unit  10 A calculates the processing time of the task A that is the largest task. The processing time of the task A exceeds the deadline. As shown in  FIG. 23 , if this goes on, the processing time of the task A will be the bottleneck of the processing time of the whole job. Accordingly, the task division unit  10 A divides the task A. The assignment unit  20  assigns the sub tasks of the task A to the computers  1  to  4 . 
         [0118]      FIG. 24  is a figure showing a state in which the tasks are assigned to the computers in such manner. However, as shown in  FIG. 24 , even when the task division unit  10 A divides the task A, the processing time of the task A exceeds the deadline. In an example shown in  FIG. 24 , because the serial processing task of the task A exceeds the deadline regardless of any division of the task, the scheduling on which the execution of the job is completed by the deadline cannot be made. Accordingly, the assignment unit  20  changes the position of the deadline from the average processing time (old deadline) of all the tasks to the completion time (new deadline) of a serial processing sub task of the task A. 
         [0119]    The processing time of the task B that is the second largest task exceeds the old deadline but it does not exceed the new deadline. Therefore, the task division unit  10  does not divide the task B. That is because even if the task division unit  10  divides the task B, the processing time of the whole job cannot be reduced so as not to exceed the new deadline. 
         [0120]    [Explanation of Operation] 
         [0121]      FIG. 25  is a flowchart for explaining the operation of the data processing system  1000 . As shown in  FIG. 25 , the job reception unit  30  receives the execution request for the job (C 100 ). The record information collection unit  110  acquires the statistical information that is the processing target of the job from the data storage unit  300  on the basis of the information for specifying the table that is the processing target included in the execution request for the job received by the job reception unit  30  (C 110 ). 
         [0122]    Next, the job division unit  10  calculates the processing time of the job on the basis of the information which specifies a content of the data processing included in the execution request for the job, the acquired statistical information, and the like. The job division unit  10  calculates the average value by dividing the processing time of the job by the number of the computers as the deadline (C 120 ). 
         [0123]    The job division unit  40  divides the job into a plurality of tasks on the basis of the statistical information. The task division unit  10  stores the divided tasks in an unscheduled task queue in a decreasing order of the size of the task (C 130 ). The task division unit  10  repeats the following processing until the unscheduled task queue is empty (C 140 ). 
         [0124]    The task division unit  10  takes out the task from the unscheduled task queue and calculates the processing time thereof (C 150 ). The task division unit  10  determines whether or not the calculated processing time exceeds the deadline (C 160 ). When the processing time exceeds the deadline, the task division unit  10  divides the task (C 170 ). When the processing time does not exceed the deadline, the data processing system  1000  proceeds to the processing of step C 180 . 
         [0125]    The assignment unit  20  assigns the task, the sub task, and the serial processing task to the computer so that the sub task is performed before another task and the serial processing task are performed (C 180 ). 
         [0126]    After the schedule is made, the assignment unit  20  determines whether or not the processing time of the serial processing task exceeds the deadline (C 190 ). When the processing time of the serial processing task exceeds the deadline, the assignment unit  20  sets the time at which the serial processing task is completed to the new deadline (C 200 ). 
         [0127]    The data processing device  1000  according to the seventh exemplary embodiment divides the task and assigns the divided task by considering the deadline. The data processing system  1000  according to the seventh exemplary embodiment has such configuration. Accordingly, the unnecessary parallelization is prevented and the overhead due to unnecessary parallelization can be prevented from being generated. 
       Eighth Exemplary Embodiment 
       [0128]    In the data processing device  1000  according to an eighth exemplary embodiment, a task execution unit  200  performs the task on the basis of a result outputted by the assignment unit  20 . 
         [0129]      FIG. 26  is a block diagram of the data processing system  1000  according to the eighth exemplary embodiment. Further, the same reference numbers are used for the elements shown in  FIG. 26  which have the substantially same function as the elements shown in  FIG. 21 . As shown in  FIG. 26 , the data processing system  1000  according to the eighth exemplary embodiment includes the task division unit  10 , the record information collection unit  110 , the assignment unit  20 , the job reception unit  30 , the job division unit  40 , task execution units  200 A to  200 N, and the data storage unit  300 . Further, the task execution units  200 A to  200 N may be abbreviated as the task execution unit  200 . 
         [0130]    The assignment unit  20  assigns the task, the sub task, and the serial processing task to the task execution unit  200  and outputs a result of assignment as schedule data. The task execution unit  200  performs the assigned task. 
         [0131]      FIG. 27  is a block diagram showing a configuration of the task execution unit  200  shown in  FIG. 26 . As shown in  FIG. 27 , the task execution unit  200  includes a schedule data acquisition unit  210 , a task processing unit  220 , a progress sharing unit  230 , and an output unit  240 . 
         [0132]    A schedule data acquisition unit  210  acquires the schedule data outputted by the assignment unit  20 . The schedule data acquisition unit  210  may input the schedule data from the assignment unit  20  or read the schedule data that is outputted and stored at a referable place by the assignment unit  20 . The task processing unit  220  performs the assigned task. 
         [0133]    The progress sharing unit  230  shares the progress (for example, Not Executed, Under-execution, Executed) of the assigned task with another task execution unit  200 . The progress sharing unit  230  may share the progress by communicating with another task execution unit  200  or may share the progress by writing the progress of the task in the schedule data and reading the progress of the task from the schedule data. The output unit  240  outputs the data that is a processing result of the assigned task. 
         [0134]      FIG. 28  is a figure showing an example of the schedule data outputted by the assignment unit  20 . The example shown in  FIG. 28  is shown as an example for ease of understanding the explanation. Therefore, the schedule data is not interpreted in a limited sense. 
         [0135]    Hereinafter, the task, the sub task, and the serial processing task assigned to the task execution unit  200  may be totally abbreviated to the “task”. 
         [0136]    The schedule data shown in  FIG. 28  corresponds to a state of the task assignment shown in  FIG. 29 . As shown in  FIG. 28 , for example, the schedule data includes a task ID, an assignment destination ID, a processing target record, a priority, a task type, and an aggregation destination task ID. 
         [0137]    The task ID is an identifier for distinguishing a certain task from another task. The assignment destination ID is information to identify the task execution unit  200  which performs the task. The processing target record is information to show the record that is the processing target of the task. The priority is information to show an order in which the tasks are processed when a plurality of tasks are assigned to the task execution unit  200 . In an example shown in  FIG. 28 , the smaller priority value means higher priority. The task type is information to show a type of the task: the task that is not divided (Normal), the sub task (Parallelized), or the serial processing task (Sequential). The aggregation destination task ID is a task ID of the serial processing task corresponding to the sub task. 
         [0138]    The processing target record may be designated by for example, a value or a value range of the attribute (the branch office name) which determines the unit in which the serial processing is performed, a value or a value range of the attribute (the transaction time) which determines the order in which the serial processing is performed, or a combination of these values or value ranges of the attributes. 
         [0139]    The processing target record of the sub task (Parallelized) may be determined by making hash dispersion of the record that is the processing target of the task corresponding to the sub task by the number of the task execution units  200 . For example, the record that is the processing target of the sub task may be assigned to each of the task execution units  200  in a dictionary order of the hash value. 
         [0140]      FIG. 30  is a flowchart for explaining the operation of the task execution unit  200 . As shown in  FIG. 30 , the task execution unit  200  repeats the following processing until all the tasks assigned by the assignment unit  20  are completed (D 100 ). 
         [0141]    The task execution unit  200  takes out one of the tasks with high priority from among the assigned tasks. The task execution unit  200  determines whether or not the taken out task is the serial processing task (Sequential) (D 110 ). When the taken out task is the serial processing task (Sequential), the process proceeds to the processing of step D 150 . When the taken out task is not the serial processing task (Normal or Parallelized), the task execution unit  200  acquires the processing target record of the taken out task from the data storage unit  300  and performs the task (D 120 ). 
         [0142]    The task execution unit  200  determines whether or not the taken out task is the sub task (Parallelized) (D 130 ). When the taken out task is the sub task (Parallelized), the task execution unit  200  performs the sub task and notifies the serial processing task corresponding to the aggregation destination task ID of the completion of the processing (D 140 ). When the taken out task is not the sub task (Parallelized) (Normal), the processing returns to the processing of step D 100  and the task execution unit  200  takes out the task with next higher priority. 
         [0143]    In the processing of step D 110 , when the task execution unit  200  takes out the serial processing task (Sequential), the task execution unit  200  waits for the completion of the execution of the sub task corresponding to the serial processing task (D 150 ) and performs the serial processing task as much as it can (D 160 ). The task execution unit  200  determines whether or not the execution of the serial processing task is completed (D 170 ). When the task execution unit  200  completes the execution of the serial processing task, the processing proceeds to the processing of step D 100  and when the task execution unit  200  does not complete the execution of the serial processing task, the task execution unit  200  waits for the completion of the execution of the corresponding sub task (D 150 ). 
         [0144]    As described above, in the data processing device  1000  according to the eighth exemplary embodiment, the task execution unit  200  performs the task on the basis of a result outputted by the assignment unit  20 . 
       Modification Example of Eighth Exemplary Embodiment 
       [0145]    Further, when the task processing unit  220  performs the assigned task, the task processing unit  220  may share a progress of the task with another task processing unit  200 .  FIG. 31  is a figure showing an example of the schedule data in a case in which the progress of the task is shared. As shown in  FIG. 31 , the schedule data includes an item of the progress which holds three states: for example, “Not Executed”, “Under-execution”, and “Executed” for each task ID. 
         [0146]    When the assigned task has already been performed and the task assigned to the another task execution unit  200  is not yet performed, the task processing unit  220  according to the modification example of the eighth exemplary embodiment performs the task that is not yet performed instead of the another task execution unit  200 . For example, in an example shown in  FIG. 32 , when the task execution unit  200 A completes the execution of the task corresponding to “F” in the task ID column, there is no more assigned task to be performed by the task execution unit  200 A. Accordingly, the task execution unit  200 A may perform the task corresponding to “G” in the task ID column instead of the task execution unit  200 B. 
         [0147]    The data processing system  1000  according to the modification example of the eighth exemplary embodiment has such configuration. Therefore, the data processing system  1000  according to the modification example of the eighth exemplary embodiment can dynamically perform the leveling of task execution even when a system failure occurs during the task execution or a task estimation error exists and whereby, the processing time of the task is increased or decreased. Further, when the assignment unit  20  makes a schedule so that the task whose processing time is large is preferentially performed, the task is subdivided into small tasks and the small tasks are assigned to the task processing units  220 . Therefore, the leveling of the execution of the task can be further easily performed. 
       Ninth Exemplary Embodiment 
       [0148]    The data processing system  1000  according to a ninth exemplary embodiment is performed on a distributed parallel execution platform.  FIG. 33  is a block diagram of the data processing system  1000  according to the ninth exemplary embodiment. Further, the same reference numbers are used for the elements shown in  FIG. 33  which have the substantially same function as the elements shown in  FIG. 21 . 
         [0149]    As shown in  FIG. 33 , the data processing system  1000  according to the ninth exemplary embodiment includes a scheduling unit  100 , task execution units  400 A to  400 N, and data storage units  500 A to  500 N. Further, the task execution units  400 A to  400 N may be abbreviated as the task execution unit  400 . The data storage units  500 A to  500 N may be abbreviated as the data storage unit  500 . 
         [0150]    As shown in  FIG. 33 , in the data processing system  1000 , the schedule unit  100  includes the task division unit  10 , the assignment unit  20 , the job reception unit  30 , the job division unit  40 , and the record information collection unit  110 . 
         [0151]      FIG. 34  is a block diagram showing a configuration of the task execution unit  400  shown in  FIG. 33 . As shown in  FIG. 34 , the task execution unit  400  according to the ninth exemplary embodiment includes a schedule data acquisition unit  410 , a record acquisition unit  420 , a shuffle key assignment unit  430 , a record transmission unit  440 , a record reception unit  450 , a task processing unit  460 , and an output unit  470 . 
         [0152]    The schedule data acquisition unit  410  acquires the schedule data outputted by the schedule unit  100 . The record acquisition unit  420  acquires the record stored in the data storage unit  500  located in the neighborhood of the task execution unit  400  irrespective of the schedule data. The shuffle key assignment unit  430  assigns a shuffle key based on the schedule data to the record acquired by the record acquisition unit  420 . The record transmission unit  440  transmits the record to the record reception unit  450  of another task processing unit  400  on the basis of the shuffle key assigned to the record. 
         [0153]    The record reception unit  450  receives the record to which the shuffle key is assigned from the another task processing unit  400 . The task processing unit  460  performs the task to the record on the basis of the shuffle key assigned to the record. The output unit  470  outputs a result of the processing. 
         [0154]      FIG. 35  is a figure for explaining an example of the shuffle key. The example shown in  FIG. 35  is shown as an example for ease of understanding the explanation. Therefore, the shuffle key is not interpreted in a limited sense. The shuffle key shown in  FIG. 35  corresponds to the schedule data shown in  FIG. 29 . 
         [0155]    In the data processing system  1000  according to the ninth exemplary embodiment, each task execution unit  400  performs the task in a dictionary order of the shuffle key assigned to the distributed record. Accordingly, the shuffle key assignment unit  430  configures the shuffle key so that each task execution unit  400  performs each task in the scheduled order as follows. 
         [0156]    The shuffle key assignment unit  430  gives a tag indicating the task type to the head of the record. At this time, the tag is determined so that the record to the sub task is positioned in front of the record to the task other than the sub task in a dictionary order. In an example shown in  FIG. 35 , a value of “0” is given in the task type column when the record is the record to the sub task and a value of “1” is given in the task type column when the task is the task that is not divided. The tag is determined in this manner. Therefore, the sub task is performed before the task that is not divided is performed. 
         [0157]    The shuffle key assignment unit  430  gives the tag representing the priority of the execution to the record. In an example shown in  FIG. 35 , smaller priority value means higher priority. This information may be given to only the task whose processing time is large. This is because the number of the tasks whose processing time is small tends to be large and the order of the execution of them has little influence. 
         [0158]    The shuffle key assignment unit  430  gives the tag indicating the task identifier to the record. At this time, the task identifier of the sub task is numbered as “A- 1 ”, “A- 2 ”, “A- 3 ”, . . . up to “A-N”. For example, a rule in which the task execution unit  400  to which the sub task whose task ID is “A- 1 ” is assigned takes charge of the corresponding serial processing task may be set in advance. 
         [0159]    In an example shown in  FIG. 35 , the tag for identifying the task execution unit  400  that is the assignment destination is given to the shuffle key itself. However, the shuffle key may not necessarily have such configuration. For example, the record transmission unit  440  may determine the task execution unit  400  to which the record is transmitted by referring to the value of the attribute (branch office name) which determines the unit in which the serial processing is performed. The record transmission unit  440  may determine the task execution unit  400  to which the record is transmitted by using information such as the statistical information of the record or the like. 
         [0160]    The task processing unit  450  may perform the record for each task identifier unit in a dictionary order of the shuffle key. 
         [0161]    Next, the task execution unit  400  which takes charge of the serial processing task determines whether or not the corresponding sub task is completed. A method of determination will be explained. For example, a method in which the task execution unit  400  confirms whether or not an intermediate file whose name includes the task identifier is generated may be used. As another method, for example, a method in which a master server for managing a progress state of the task is introduced in the data processing system  1000  may be used. The task execution unit  400  which takes charge of the serial processing task may recognize the progress of the sub task performed by another task execution unit  400 , appropriately acquire the intermediate data required for the calculation, and perform the serial processing task by using for example, the above-mentioned method. 
         [0162]    As described above, by using the data processing system  1000  according to the ninth exemplary embodiment, the data processing system which operates on the distributed parallel execution platform and optimizes the processing time of the whole data processing including the parallel processing and the serial processing, the data processing method, and the program can be provided. 
         [0163]    The table described in the exemplary embodiment of the invention corresponds to a data set described in the claims. The task execution unit  200  or the task execution unit  400  described in the exemplary embodiment of the invention corresponds to a computer described in the claims. 
         [0164]    This application claims priority based on Japanese Patent Application No. 2012-239782 filed on Oct. 31, 2012, the disclosure of which is hereby incorporated by reference in its entirety. 
       INDUSTRIAL APPLICABILITY 
       [0165]    As an example of the utilization of the present invention, the present invention can be applied to increase the speed of a batch processing of a mission critical system of a company. 
       REFERENCE SIGNS LIST 
       [0000]    
       
         
           
               1  CPU 
               2  RAM 
               3  HDD 
               4  communication interface 
               5  input device 
               6  output device 
               10  task division unit 
               20  assignment unit 
               21  assignment unit 
               30  job reception unit 
               31  job reception unit 
               40  job division unit 
               41  job division unit 
               100  schedule unit 
               110  record information collection unit 
               200 A to  200 N task execution unit 
               210  schedule data acquisition unit 
               220  task processing unit 
               230  progress sharing unit 
               240  output unit 
               300  data storage unit 
               400 A to  400 N task execution unit 
               410  schedule data acquisition unit 
               420  record acquisition unit 
               430  shuffle key assignment unit 
               440  record transmission unit 
               450  record reception unit 
               460  task processing unit 
               470  output unit 
               500 A to  500 N data storage unit 
               1000  data processing system