Abstract:
A series data distributed processing system including a parallel processing infrastructure and a distributed series data management unit managing distributed series data, wherein: the parallel processing infrastructure includes, on each computing device, a data block, a data block processing server for processing the data block, and a block processing integration server for processing a result from the data block processing server, the data block being formed from a plurality of values each associated with one of a plurality of sequential labels in the series data; and the distributed series data management unit includes a distributed information management database for managing data blocks, which retains management data, including sequential label ranges, which refer to ranges of sequential labels in the data blocks, series IDs corresponding to value names in the data blocks, and meta-information identifying computing devices retaining the data blocks.

Description:
BACKGROUND 
       [0001]    The present invention pertains to a parallel distributed processing method and a computer system for processing large amounts of series data in parallel using a plurality of distributed computers. 
         [0002]    In recent years, the big data processing is receiving attention, which analyzes and processes large amounts of data to find and make use of findings having never been obtained so far. In the big data, sensor data obtained from a device, for example, has a form of data called series data. Series data is a data set of a plurality of data pieces arranged in accordance with respective sequential labels, each of the plurality of data pieces consisting of values for a plurality of data items. 
         [0003]    While there are needs for analysis of large amounts of data, it is necessary to design a system for each processing in existing distributed analysis systems, resulting in the high cost of the system configuration. 
         [0004]    Concerning such an issue, the MapReduce framework is proposed as core technology for implementing analysis processing with ease as described in Patent Literature 1 and Non-Patent Literature 1. The MapReduce framework is a programming model for writing an analysis procedure of data in two parts: an extraction procedure (MAP procedure) that extracts desired data from data store, and an aggregation procedure (Reduce procedure) that transforms the extracted data into a readily usable form or statistic information. It allows the execution engine of the MapReduce framework to determine the division unit of an analysis application and to control parallel processing. 
         [0005]    However, the MapReduce framework is originally aimed at writing processing of unstructured and non-sequential data such as a search system in the Web. Thus, it is impossible to expect an increase in the processing performance for series data from the MapReduce framework. For example, the extraction procedure is executed simultaneously at a large number of infrastructures as a plurality of tasks. Thus, it greatly contributes to enhancing the processing speed; however, it is difficult to apply an analysis method usually applied to series data, such as moving average calculation and furrier transformation. 
         [0006]    The aggregation procedure is used for writing such processing in the MapReduce framework; however, it is difficult in the aggregation processing to increase the number of infrastructures for the processing to enhance the processing speed. 
         [0007]    Concerning such an issue, a technique to utilize a stream processing infrastructure in the aggregation processing for speedup is known, as described in Non-Patent Literature 2. However, even if using the stream processing infrastructure, there is a problem that the waiting time for all data to be extracted in the extraction processing occurs, and the transmission of the extracted data directly to another server via a network results in an increase in the communication loads. Further, the aggregation processing of series data does not always reduce the amount of data sufficiently in the writing process of the result, and the relocation of large amounts of data leads to increases in the communication and processing loads or a reduction in the speed. 
       PRIOR ART 
       [0008]    Patent Literature 1: US2008/0086442 A1 
         [0009]    Non-Patent Literature 1: “MapReduce: Simplified Data Processing on Large Clusters” Jeffrey Dean, Sanjay Ghemawat, Google, Inc. OSDI&#39;04: Sixth Symposium on Operating System Design and Implementation, San Francisco, Calif., Dec. 6, 2004. 
         [0010]    Non-Patent Literature 2: “SCOPE: parallel databases meet MapReduce” Jingren Zhou, Nicolas Bruno, Ming-Chuan Wu, Per-Ake Larson, Ronnie Chaiken, Darren Shakib, Microsoft Corp., The VLDB Journal 
       SUMMARY 
       [0011]    It is impossible for conventional distributed analysis systems without the MapReduce framework to write data processing flexibly. 
         [0012]    Conventional system configurations utilizing the MapReduce framework cannot achieve the processing improvement in accordance with the number of computing devices (scalability). The conventional system configurations utilizing the MapReduce framework cannot save the primary processed data at high speed when subjecting the primary processed data to the secondary processing after the primary processing of the original data. 
         [0013]    A representative example of invention disclosed in the present application is as follows. 
         [0014]    Provided is a series data parallel analysis infrastructure performing parallel distributed processing on series data including sequential labels each associated with at least one value in a one-to-one or one-to-many relation and at least one value name to distinguish values in the series data, the parallel analysis infrastructure including: 
         [0015]    a parallel processing infrastructure configured to perform data processing in parallel, and including at least one computing device and a network for the at least one computing device to communicate data; and a distributed series data management unit configured to manage series data distributed and placed in the at least one computing device, 
         [0016]    wherein the parallel processing infrastructure includes: at least one data block on each of the at least one computing device, the at least one data block being formed from a plurality of values each associated with one of a plurality of sequential labels in the series data; at least one data block processing server configured to process the at least one data block on each of the at least one computing device; and at least one block processing aggregation server configured to process a result of the at least one data block processing server on each of the at least one computing device, and 
         [0017]    wherein the distributed series data management unit includes: a data input unit for holding the series data in the series data parallel analysis infrastructure; a data selection unit for obtaining the series data from the series data parallel analysis infrastructure; and a distribution information management database for managing the at least one data block, containing a sequential label range specifying an range of sequential labels of the at least one data block, a series ID corresponding to a value name of the at least one data block and meta-information specifying a computing device on which the at least one data block is placed. 
         [0018]    A series data parallel analysis infrastructure or a distributed processing system with the above descried configuration allows flexible writing of distributed processing for series data. Further, a data store and a processing infrastructure with scalability for series data are accomplished. In addition, new data can be saved at high speed when processing original data retained in a data store for creating the new data. 
     
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS 
         [0019]      FIG. 1  is an explanatory diagram depicting the outline of a series data parallel analysis infrastructure according to Embodiment 1 of the present invention. 
           [0020]      FIG. 2  is an explanatory diagram depicting the outline of series data according to Embodiment 1 of the present invention. 
           [0021]      FIG. 3  is an explanatory diagram depicting a configuration of a series data column store according to Embodiment 1 of the present invention. 
           [0022]      FIG. 4  is an explanatory diagram depicting a configuration of a parallel processing infrastructure according to Embodiment 1 of the present invention. 
           [0023]      FIG. 5  is a block diagram depicting a configuration of the series data parallel analysis infrastructure according to Embodiment 1 of the present invention. 
           [0024]      FIG. 6  is a flowchart depicting steps of series data parallel analysis processing according to Embodiment 1 of the present invention. 
           [0025]      FIG. 7  is a flowchart depicting steps of data selection for a series data parallel store according to Embodiment 1 of the present invention. 
           [0026]      FIG. 8  is a flowchart depicting steps of data recording for the series data parallel store according to Embodiment 1 of the present invention. 
           [0027]      FIG. 9  is a flowchart depicting steps of determining distribution targets in the data recording for the series data parallel store according to an embodiment of the present invention. 
           [0028]      FIG. 10  is a flowchart depicting steps of deter mining distribution targets in the data recording for the series data parallel store according to an embodiment of the present invention. 
           [0029]      FIG. 11  is a flowchart depicting steps of determining distribution targets in the data recording for the series data parallel store according to an embodiment of the present invention. 
           [0030]      FIG. 12  is a flowchart depicting steps of determining distribution targets in the data recording for the series data parallel store according to an embodiment of the present invention. 
           [0031]      FIG. 13  is a flowchart depicting steps of data recording in Map processing for the series data parallel store according to Embodiment 1 of the present invention. 
           [0032]      FIG. 14  is an explanatory diagram depicting an advantage of Embodiment 1 of the present invention. 
           [0033]      FIG. 15  is a block diagram depicting Embodiment 2 of the present invention. 
           [0034]      FIG. 16  is a flowchart depicting steps of series data parallel analysis processing according to Embodiment 2 of the present invention. 
           [0035]      FIG. 17  is a flowchart depicting steps of data selection for the series data parallel store according to Embodiment 2 of the present invention. 
       
    
    
     DETAILED DESCRIPTION OF THE EMBODIMENTS 
       [0036]    Hereinafter, embodiments of the present invention are described with reference to drawings. 
       Embodiment 1 
       [0037]      FIG. 1  depicts the outline of a series data parallel analysis infrastructure  102  according to the present invention. The series data parallel analysis infrastructure  102  is configured to include a series data parallel store  103  that receives and stores series data  101 , and a parallel processing infrastructure  104  that performs distributed analysis using data stored in the series data parallel store  103 . The parallel processing infrastructure  104  receives a processing instruction from a user  106  and outputs the processing result as the series data parallel store  103  or analysis result data  105 . The user  106  may be a person, another system or a program. 
         [0038]    The above configuration allows series data  101  to be stored sequentially and the stored series data  101  to be analyzed using the parallel analysis infrastructure at any time requested by the user  106 . 
         [0039]      FIG. 2  depicts a structure of series data. Series data is configured to include one or more data blocks  203 . One data block  203  is configured to include sequential labels  201  indicating a sequence and one or more types of values  202 . Typically, numerical values or dates and times are used for the sequential labels. Alternatively, symbols that can define a sequence, such as a, b and c, may be used for the sequential labels. 
         [0040]      FIG. 3  depicts a typical configuration of a series data column store  301 . The series data column store  301  is configured to include a data input unit  303  that reads in the series data  101 , a data base  302  that stores data, and a data selection unit  306  that obtains data. The data base  302  includes a column of sequential label range  303 , a column of series ID  304  and a column of value block  305 , and stores the series data  101  in the form of so-called column store. This configuration allows the user  106  to obtain data by providing an instruction/operation to the data selection unit from the user  106 . 
         [0041]    The above configuration allows the user  106  to obtain in a short time a set of sequential labels and values corresponding to a desired sequential label range  303  and a series ID  304 . Further, a data compression technique applied to value blocks reduces the amount of stored series data  101 . 
         [0042]      FIG. 4  depicts a typical configuration example of the parallel processing infrastructure  104 . The parallel processing infrastructure  104  is configured to include a job and task management server  401  that monitors and manages the processing status, a data block management server  402  that manages data, data block processing servers  403  that process data blocks  203 , and block processing aggregation servers  404  that aggregate results of the data block processing servers  403 . One or more data block processing servers  403  and one or more block processing aggregation servers  404  are present, and a data block processing server  403  and a block processing aggregation server  404  may be installed on the same device. The user  106  provides an instruction/operation to the job and task management server  401  to obtain a processing result. This configuration allows writing of distributed analysis processing on various data with high flexibility and a few man-hours. 
         [0043]      FIG. 5  depicts a configuration of the series data parallel analysis infrastructure  102  according to Embodiment 1 of the present invention. The series data parallel analysis infrastructure  102  is configured to include the job and task management server  401  that monitors and manages the processing status, data blocks  509 , the data block processing servers  403  that process the data blocks  509 , the block processing aggregation servers  404  that aggregate results of the data block processing servers  403 , and a distributed series data management unit  501 . 
         [0044]    The distributed series data management unit  501  is configured to include a data input unit  502 , a data selection unit  503 , a distribution target determination unit  504 , a data search unit  505 , a data input/output unit  506 , and a distribution information management database  507 . The distribution information management database  507  is configured to include a column of sequential label  303 , a column of series ID  304 , and a column of meta-information  508 . The meta-information  508  contains information on servers actually holding the data blocks  509 . One data block  509  may be stored in one server or a plurality of servers. When one data block  509  is stored in a plurality of servers, the meta-information  508  associated with the data block  509  contains information on the plurality of servers. 
         [0045]    This configuration is based on the combination of the series data column store  301  illustrated in  FIG. 3  and the parallel processing infrastructure  104  illustrated in  FIG. 4 . A feature of the configuration is that the data blocks  509  are stored in devices with the data block processing servers  403  or devices where the data block processing servers  403  can obtain data with ease, without storing the data blocks  509  in the distribution information management database  507 . The data block  509  may be a file fog filed from the value block  305  illustrated in  FIG. 3 . 
         [0046]    The above described configuration allows the series data to be stored in a plurality of servers/devices, and facilitates increasing the holding capacity in data storing or expanding the holding capacity by adding devices. Further, when data increases, the configuration can improve the analysis performance by increasing devices 
         [0047]    Hereinafter, the outline of operations of the series data parallel analysis infrastructure  102  is described with reference to flowcharts illustrated in  FIG. 6  to  FIG. 11 . 
         [0048]      FIG. 6  depicts a method for distributed analysis performed by the series data parallel analysis infrastructure  102 . First, a processing execution request from the user  106  is issued to the series data parallel analysis infrastructure  102 . The processing execution request includes a sequential label range  303 , a series ID  304 , and a program describing the details of the processing (Step  1 ). Next, information indicating data blocks  509  associated with the sequential label range  303  and the series ID  304  included in the processing execution request is obtained from the distribution information management database  507  via the data search unit  505  (Step  2 ). 
         [0049]    The parallel processing infrastructure  104  selects data blocks for one time extraction processing based on the information on the data blocks  509  obtained from the distribution information management database, and creates a chunk consisting of the selected data blocks. It is desirable that the data blocks and the server to perform the extraction processing for the data blocks are on the same device; however, they may be on different devices. When they are on different devices, the server to perform the extraction processing may obtain data of the data blocks  509  from the device by ssh, scp and so on (Step  3 ). The job and task management server  401  creates an extraction task for each chunk and provides the data block processing server  403  with an instruction for the extraction task (Step  4 ). 
         [0050]    The data block processing server  403  obtains data from the chunk in the order of the sequential labels, and performs the extraction processing in accordance with the program describing the details of the processing specified in Step  1 . Result data of the extraction processing takes a form of values associated with sequential labels (called keys hereinafter) other than the sequential labels corresponding to the values in the data blocks  509 . The key may be the same as the sequential label in the data block  509  and a plurality of labels may be associated with a key (Step  5 ). The result of the extraction processing is transferred to the block processing aggregation servers  404  (Step  6 ). 
         [0051]    The transferred data is sorted by the keys in the block processing aggregation servers  404  (Step  7 ). The block processing aggregation servers  404  perform the aggregation processing for each group of values sharing a key (Step  8 ). Finally, the aggregation result is output, and the analysis processing ends. The output result may be stored in any type of file (tab separated text file or XML file, for example), or stored in the distributed series data management unit  501 . The result may be transferred on memory to another program outside the series data parallel analysis infrastructure  102  (Step  9 ). 
         [0052]    In the above described configuration, the sequential label range  303  appropriately specified guarantees the sequence of data in the extraction process in Step  5 . Thus, processing based on the sequence, such as moving average and furrier transform, can be described in the extraction procedure. 
         [0053]    Typically, the number of servers or tasks performing the aggregation processing is smaller than the extraction processing. Thus, processing based on the sequence can be described in the extraction procedure, resulting in a reduction in the transmitted amount of the result in Step  6 , and low-load and high-speed distributed analysis of the series data  101 . 
         [0054]      FIG. 7  depicts a method for data selection by the series data parallel analysis infrastructure  102 . 
         [0055]    First, a processing execution request from the user  106  is issued to the series data parallel analysis infrastructure  102 . The processing execution request includes a sequential label range  303  and a series ID  304  for the processing (Step  11 ). Next, information indicating data blocks  509  associated with the sequential label range  303  and the series ID  304  included in the processing execution request is obtained from the distribution information management database  507  via the data search unit  505  (Step  12 ). Data in the data blocks is transmitted to the distributed series data selection unit by ssh, scp and so on via the data input/output unit  506  and the data search unit  505  (Step  13 ). The data selection unit converts the data into a file (tab separated text file or XML file, for example) or a form of data which can be transferred on memory to another program outside the series data parallel analysis infrastructure  102 , and outputs the converted data to terminate the process (Step  14 ). In this way, the data can be obtained without passing through the parallel processing infrastructure  104 . 
         [0056]      FIG. 8  depicts a method for recording data by the series data parallel analysis infrastructure  102 . 
         [0057]    First, a processing execution request from the user  106  is issued to the series data parallel analysis infrastructure  102  via the data input unit  502 . The processing execution request includes series data  101  used in the processing and series IDs  304  used in the recording (Step  21 ). Next, the data input unit divides the input series data in the form such as binary tab separated text file and XML with the series IDs  304  and predetermined sequential label ranges  303  (Step  22 ). 
         [0058]    The data divided in Step  22  is blocked. For example, data blocks of data reduced by a compression technique are created (Step  23 ). It is determined which server/device holds each data block  509  in accordance with the distribution target determination method included in the distribution target determination unit  504 . A plurality of servers/devices may be selected in view of availability (Step  24 ). The sequential label ranges  303 , the series IDs  304  and the information on the distribution target servers/devices are recorded in the distribution information management database  507  (Step  25 ). Finally, the data blocks are disposed on the servers/devices via the data input/output unit  506  to terminate the process (Step  26 ). 
         [0059]    The above described configuration allows data recording that enables determination of distribution targets in a form customizable by the user  106 . 
         [0060]      FIG. 9  depicts an example method for determining distribution targets. The method illustrated in  FIG. 9  is based on the so called round robin algorithm. This method uses a list containing information on distribution target servers/devices, and a distribution target list pointer indicating one distribution target on the list. 
         [0061]    First, a determination number is set to be 0 (Step  31 ). A server/device indicated by the distribution target list pointer is determined to be a distribution target for each data block  509  (Step  32 ). If it is determined that the distribution target list pointer does not indicate the last distribution target on the list in Step  33 , the distribution target list pointer moves one step on the list (Step  34 ). If the distribution target list pointer indicates the last distribution target on the list in Step  33 , the distribution target list pointer moves to the top on the list (Step  35 ). Next, the determination number is incremented (Step  36 ). The above steps (Step  32  to Step  36 ) are repeated until the determination number reaches a predetermined multiplicity level, and then the process ends (Step  37 ). Thus, the data blocks  509  can be dispersed on servers/devices evenly without concentration on specific severs/devices. 
         [0062]      FIG. 10  depicts an example of a distribution target deter urination method. The method illustrated in  FIG. 10  equalizes the usage amounts of storage areas in servers/devices. The method includes steps of obtaining a distribution target list and the usage amounts of storage areas in the servers/devices on the distribution target list 
         [0063]    First, a determination number is set to be 0 (Step  41 ). The usage amount of storage area in each server/device is obtained and the server/device with the minimum usage amount of storage area is determined to be a distribution target (Step  42 ). Next, the determination number is incremented (Step  43 ). The above steps (Step  42  and Step  43 ) are repeated until the determination number reaches a predetermined multiplicity level, and then the process ends (Step  44 ). Thus, the usage amounts of storage areas in the servers/devices can be equalized. 
         [0064]      FIG. 11  depicts an example of a distribution target determination method. The method illustrated in  FIG. 11  specifies a distribution target for each series ID  304 . The method uses a function f to associates servers/devices on a distribution target list and series IDs  304 . 
         [0065]    First, a determination number is set to be 0 (Step  51 ). The function f is executed for each data block  509  (Step  52 ). A server/device on the distribution target list corresponding to the value of the function f executed for a data block  509  is determined to be a distribution target for the data block  509  (Step  53 ). Next, the determination number is incremented (Step  54 ). The above steps (Step  52  to Step  54 ) are repeated until the determination number reaches a predetermined multiplicity level, and then the process ends (Step  55 ). A hash function or a modulo function may be used as the function f, for example. The series ID  304  or a value calculated from the series ID  304  may be used as an argument of the function f, for example. The function f is defined such that when an entry, such as a server, is added to or deleted from the distribution target list, and the entry, such as a server, associated with a series ID  304  is not deleted, the same entry, such as a server, is selected for the series ID  304 . 
         [0066]    Thus, distributed processing tasks of a sequential label range  303  can be executed efficiently on a plurality of infrastructures for a plurality of series IDs  304 . 
         [0067]      FIG. 12  depicts an example of a distribution target determination method. The method illustrated in  FIG. 12  specifies a distribution target for each sequential label range  303 . The method uses a function g to associates servers/devices on a distribution target list and sequential label ranges  303 . 
         [0068]    First, a determination number is set to be 0 (Step  61 ). The function g is executed for each data block  509  (Step  62 ). A server/device on the distribution target list corresponding to the value of the function g executed for a data block  509  is determined to be a distribution target for the data block  509  (Step  63 ). Next, the determination number is incremented (Step  64 ). The above steps (Step  62  to Step  64 ) are repeated until the determination number reaches a predetermined multiplicity level, and then the process ends (Step  65 ). A hash function or a modulo function may be used as the function g, for example. The sequential label range  303  or a value calculated from the sequential label range  303  may be used as an argument of the function g, for example. The function g is defined such that when an entry, such as a server, is added to or deleted from the distribution target list, and the entry, such as a server, associated with a sequential label range  303  is not deleted, the same entry, such as a server, is selected for the sequential label range  303 . 
         [0069]    Thus, distributed processing tasks of a series ID  304  can be executed efficiently on a plurality of infrastructures for a plurality of sequential label ranges  303 . 
         [0070]      FIG. 13  depicts a method for recording data blocks in the extraction processing by the series data parallel analysis infrastructure  102 . 
         [0071]    First, a processing execution request from the user  106  is issued to the series data parallel analysis infrastructure  102 . The processing execution request includes sequential label ranges  303  and series IDs  304  used in the processing, series IDs  304  in the data block recording, and a program describing distribution target location information and details of processing (Step  71 ). Next, the information indicating the data blocks  509  associated with pairs of the sequential label ranges  303  and the series IDs  304  of the processing targets included in the processing execution request is retrieved from the distribution information management database  507  via the data search unit  505 . Further, the sever/devices to be the location targets of the data blocks  509  are determined, and the information is recorded in the distribution information management database in a manner similar to the manner in the data recording (Step  72 ). The parallel processing infrastructure  104  selects data blocks for one time extraction processing based on the information on the data blocks  509  obtained from the distribution information management database, and creates a chunk consisting of the selected data blocks in the job and task management server  401 . It is desirable that the data blocks and the server to perform the extraction processing are on the same device; however, they may be on different devices. When they are on different devices, the server to perform the extraction processing may obtain data of the data blocks  509  from the device by ssh, sep and so on (Step  73 ). 
         [0072]    The job and task management server  401  creates an extraction task for each chunk and provides the data block processing server  403  with an instruction for the extraction task (Step  74 ). The data block processing server  403  obtains data from the chunk in the order of the sequential labels, and performs the extraction processing in accordance with the program describing the details of the processing specified in Step  71  (Step  75 ). When data blocks are recorded in the extraction processing, the data of the extraction processing result is transformed into the fog in of the data block  509  with the time ranges the same as the chunks subjected to the extraction tasks and the series IDs  304  specified in Step  71 , and written to the servers/devices determined in Step  72  (Step  76 ). Then, the process ends. The data blocks may be placed by communication means such as ssh and scp; however, it is possible to reduce the communication traffic by placing the data blocks at the same servers/devices performing the extraction processing on the data blocks to the extent possible. 
         [0073]    Thus, when data is recorded after aggregation processing, data transmission or data writing for or after the aggregation processing is reduced, resulting in the reduction of workloads on the system and the higher speed of the system operations. 
         [0074]      FIG. 14  depicts the comparison of processing times for calculating a moving average between the series data parallel analysis infrastructure  102  and an existing distributed analysis infrastructure. They had the same system device configuration, and used five data block processing servers and five block processing aggregation servers. 
         [0075]    The both infrastructures took approximately 30 seconds in the calculation time for the initial setting of the parallel processing infrastructures. The present embodiment is approximately five times faster than the existing technique. This is because the present embodiment allows calculation of the moving average in the data block processing servers. As described above, the present embodiment allows flexible writing of various kinds of analysis processing for series data, and establishing a series data parallel analysis infrastructure capable of increasing the data storage area and the calculation performance for analysis processing with the increase of the number of devices. 
       Embodiment 2 
       [0076]    Embodiment 2 is different from Embodiment 1 in that Embodiment 2 stores a data block in a data block server  1501  instead of holding the data block in a file directly.  FIG. 15  depicts a configuration example of the series data parallel analysis infrastructure  102  according to Embodiment 2. 
         [0077]    The series data parallel analysis infrastructure  102  is configured to include the job and task management server  401  that monitors and manages the processing status, data block servers  1501 , the data block processing servers  403  that process data blocks  509  provided from the data block servers  1501 , the block processing aggregation servers  404  that aggregate results of the data block processing servers  403 , and the distributed series data management unit  501 . 
         [0078]    The distributed series data management unit  501  is configured to include the data input unit  502 , the data selection unit  503 , the distribution target determination unit  504 , the data search unit  505 , the data input/output unit  506 , and the distribution information management database  507 . The distribution information management database  507  is configured to include the column of sequential label  303 , the column of series ID  304 , and the column of meta-information  508 . The meta-information  508  contains information on the data block servers  1501  actually holding the data blocks  509 . One data block  509  may be stored in one data block server  1501  or a plurality of data block servers  1501 . When one data block  509  is stored in a plurality of data block servers  1501 , the meta-information  508  associated with the data block  509  contains information on the plurality of data block servers  1501 . 
         [0079]    This configuration is based on the combination of the series data column store  301  illustrated in  FIG. 3  and the parallel processing infrastructure  104  illustrated in  FIG. 4 . A feature of the configuration is that the data block servers  1501  are stored in the devices with data block processing servers  403  or devices where the data block processing servers  403  can obtain data with ease, without storing the data blocks  509  in the distribution information management database  507 . The data block  509  may be a file created from the value block  305  illustrated in  FIG. 3 , and the data block  509  is held in the data block server  1501 . 
         [0080]    The above described configuration allows the series data to be stored in a plurality of servers/devices, and facilitates increasing the holding capacity in data storing or expanding the holding capacity by adding devices. Further, when data increases, the configuration can improve the analysis performance by increasing devices 
         [0081]    Hereinafter, the outline of operations of the series data parallel analysis infrastructure  102  is described with reference to flowcharts illustrated in  FIG. 16  and  FIG. 17 . 
         [0082]      FIG. 16  depicts a method for distributed analysis performed by the series data parallel analysis infrastructure  102 . 
         [0083]    First, a processing execution request from the user  106  is issued to the series data parallel analysis infrastructure  102 . The processing execution request includes a sequential label range  303 , a series ID  304 , and a program describing the details of the processing (Step  81 ). Next, information indicating data blocks  509  associate with the sequential label range  303  and the series ID  304  included in the processing execution request are obtained from the distribution information management database  507  via the data search unit  505  (Step  82 ). 
         [0084]    The parallel processing infrastructure  104  selects data blocks for one time extraction processing based on the information on the data blocks  509  obtained from the distribution information management database, and creates a chunk consisting of the selected data blocks (Step  83 ). It is desirable that the data block server  1501  and the server to perform the extraction processing are on the same device; however, they may be on different devices. When they are on different devices, the server to perform the extraction processing may obtain data of the data blocks  509  from the device by ssh, scp and so on. 
         [0085]    Next, the job and task management server  401  creates an extraction task for each chunk and provides the data block processing server  403  with an instruction for the extraction task (Step  84 ). The data block processing server  403  obtains data from the chunk in the order of the sequential labels, and performs the extraction processing in accordance with the program describing the details of the processing specified in Step  81 . Result data of the extraction processing takes a form of values associated with sequential labels (called keys hereinafter) other than the sequential labels corresponding to the values in the data blocks  509 . The key may be the same as the sequential label in the data block  509  and a plurality of labels may be associated with a key (Step  85 ). 
         [0086]    Next, the result of the extraction processing is transferred to the block processing aggregation server  404  (Step  86 ). The transferred data is sorted on the keys in the block processing aggregation servers  404  (Step  87 ). The block processing aggregation servers  404  perform the aggregation processing for each group of values sharing a key (Step  88 ). 
         [0087]    Finally, the aggregation result is output, and the analysis processing ends. The output result may be stored in any type of file (tab separated text file or XML file, for example), or stored in the distributed series data management unit  501 . The result may be transferred on memory to another program outside the series data parallel analysis infrastructure  102  (Step  89 ). 
         [0088]    In the above described configuration, the sequential label range  303  appropriately specified guarantees the sequence of data in the extraction process in Step  85 . Thus, processing based on the sequence, such as moving average and furrier transform, can be described in the extraction procedure. 
         [0089]    Typically, the number of servers or tasks performing the aggregation processing is smaller than the extraction processing. Thus, processing based on sequence can be described in the extraction procedure, resulting in a reduction in the transmitted amount of the result in Step  86 , and low-load and high-speed distributed analysis of the series data  101 . 
         [0090]      FIG. 17  depicts a method for data selection by the series data parallel analysis infrastructure  102 . 
         [0091]    First, a processing execution request from the user  106  is issued to the series data parallel analysis infrastructure  102 . The processing execution request includes a sequential label range  303  and a series ID  304  for the processing (Step  91 ). Next, information indicating data blocks  509  associated with the sequential label range  303  and the series ID  304  included in the processing execution request is obtained from the distribution information management database  507  via the data search unit  505  (Step  92 ). Data in the data blocks is transmitted to the distributed series data selection unit by ssh, scp and so on via the data block servers  1501 , the data input/output unit  506  and the data search unit  505  (Step  93 ). The data selection unit converts the data into a file (tab separated text file or XML file, for example) or a form of data which can be transferred on memory to another program outside the series data parallel analysis infrastructure  102 , and outputs the converted data to terminate the process (Step  94 ). In this way, the data can be obtained without passing through the parallel processing infrastructure  104 . 
       REFERENCE SINGS LIST 
       [0092]      101  Series data 
         [0093]      102  Series data parallel analysis infrastructure 
         [0094]      103  Series data parallel store 
         [0095]      104  Parallel processing infrastructure 
         [0096]      105  Analysis result data 
         [0097]      106  User 
         [0098]      201  Sequential label 
         [0099]      202  Value 
         [0100]      203  Data block 
         [0101]      301  Data input unit 
         [0102]      302  Data base 
         [0103]      303  Sequential label range 
         [0104]      304  Series ID 
         [0105]      305  Value block 
         [0106]      306  Data selection unit 
         [0107]      401  Job task management server 
         [0108]      402  Data block management server 
         [0109]      403  Data block processing server 
         [0110]      404  Block processing aggregation servers 
         [0111]      501  Distributed series data management unit 
         [0112]      502  Data input unit 
         [0113]      504  Distribution target determination unit 
         [0114]      505  Data search unit 
         [0115]      506  Data input/output unit 
         [0116]      507  Distribution information management database 
         [0117]      508  Meta-information 
         [0118]      509  Data block 
         [0119]      1501  Data block server