Resource optimization for parallel data integration

For optimizing resources for a parallel data integration job, a job request is received, which specifies a parallel data integration job to deploy in a grid. Grid resource utilizations are predicted for hypothetical runs of the specified job on respective hypothetical grid resource configurations. This includes automatically predicting grid resource utilizations by a resource optimizer module responsive to a model based on a plurality of actual runs of previous jobs. A grid resource configuration is selected for running the parallel data integration job, which includes the optimizer module automatically selecting a grid resource configuration responsive to the predicted grid resource utilizations and an optimization criterion.

CROSS-REFERENCE TO RELATED APPLICATIONS

This application is related to the following applications, which are owned in common with the present application and are hereby incorporated herein by reference:

X. Pu, “Apparatus, System and Method for Generating a Resource Utilization Description for a Parallel Data Processing System,” United States Published Patent Application 2008/0114870, filed Nov. 10, 2006; and

X. Pu and C. Thornton, “Managing Parallel Data Processing Jobs in Grid Environments,” United States Published Patent Application 2009/0193427, filed Jan. 30, 2008.

BACKGROUND

The present invention relates to job design for parallel data integration, and more particularly to automatic and dynamic creation of configurations for parallel data integration jobs.

Data integration in a parallel computing environment typically involves computations with large data sets, which may even exceed thousands of Gigabytes, distributed among computers or partitions. (Partitions may correspond to the computers in some instances and may exist as computer subdivisions in other instances.) This, of course, consumes both processing and storage resources.

Regarding the storage resources, the data sets include i) one or more originating data sets (also known as “data sources”), which consume disk space that may be associated with respective originating computers, and ii) one or more destination data sets (also known as “data sinks”), which consume computer disk space that may be associated with respective destination computers. Such data sets are for long term storage, which may be referred to as “permanent” storage.

In addition to consuming long term storage for originating and destination storage data sets, data integration requires temporary storage. That is, during a time while data sets are processed, some operations require memory, e.g., disk space, to store intermediate results. (This memory to store intermediate results of processing operations may be called “scratch space” or “intermediate memory.”) For example, a sorting operation may require memory for storing intermediate results during the time the sort is executing.

In a parallel computing environment, various operations on data sets can be performed across different computers and the operational steps for processing the data sets can be intertwined and complicated. Processing steps, i.e., data processing flow, to be performed on one or more data sets may be described by a “data flow graph” (also referred to, more simply, as a “data graph”), which may be a pictorial or string representation of relationships among operations in the data processing flow. A data graph describes data sources, operations to be performed and data sinks. The data flow graph may be coupled with other information—for example, a configuration file describing which operations are performed on which partitions and a data schema for each operator at the input and output for the given operator.

Based on the above, it should be appreciated that resource utilization may include data source and sink storage space, scratch space, and processor utilization.

SUMMARY OF THE INVENTION

Methods of optimizing resources for a parallel data integration job are provided, along with systems and computer program products therefor. In one implementation, a job request is received, which specifies a parallel data integration job to deploy in a grid. Grid resource utilizations are predicted for hypothetical runs of the specified job on respective hypothetical grid resource configurations. This includes automatically predicting grid resource utilizations by a resource optimizer module responsive to a model based on a plurality of actual runs of previous jobs. A grid resource configuration is selected for running the parallel data integration job, which includes the optimizer module automatically selecting a grid resource configuration responsive to the predicted grid resource utilizations and an optimization criterion.

In another aspect, resource utilization categories are generated, including the optimizer module automatically generating resource utilization categories responsive to the predicted grid resource utilizations, wherein the optimizer module automatically selecting the grid resource configuration responsive to the optimization criterion includes selecting the grid resource configuration responsive to the categories.

In another aspect, the job request includes operators specifying parallel data integration operations performed when the parallel data integration job is run. The optimizer module automatically generating resource utilization categories responsive to the predicted grid resource utilizations includes generating resource utilization indices for the respective operators responsive to the predicted grid resource utilizations; generating resource utilization indices for respective groups of the operators responsive to the resource utilization indices of the respective operator groups; and generating a first resource utilization index for the job responsive to the resource utilization indices of the respective operator groups.

In another aspect, a second resource utilization index for the job is generated, wherein generating the second resource utilization index includes selecting a first maximum of the resource utilization indices for a first subset of the operator groups; selecting a second maximum of the resource utilization indices for a second subset of the operator groups; and computing a ratio of the first and second maxima.

In another aspect, a third resource utilization index is generated for the job responsive to a sum of the predicted grid resource utilizations for all the operators.

In another aspect, correlation coefficients are generated for the model responsive to performance data for a plurality of previous jobs actually run on respective configurations of the grid resources.

In another aspect, selecting the grid resource configuration for running the parallel data integration job includes selecting an optimal number of physical compute nodes.

In another aspect, selecting the grid resource configuration for running the parallel data integration job includes adjusting estimated resource utilizations for operators responsive to a ratio of user-specified job execution time to estimated job execution time; and selecting a number of partitions for each operator responsive to a combined total of adjusted estimated resource utilizations on all the operator's partitions and a minimum of the adjusted estimated resource utilizations among all the operator's partitions.

In another aspect, the job request includes a data graph of linked operators specifying a sequence of parallel data integration operations performed when the parallel data integration job is run, such that each operator has one or more respective link mates. Selecting the grid resource configuration for running the parallel data integration job includes traversing the data graph and increasing numbers of partitions for operators having throughputs less than their respective link mates.

In another aspect, selecting the grid resource configuration for running the parallel data integration job includes selecting a number of partitions for each physical node responsive to the categories.

DETAILED DESCRIPTION

Headings herein are not intended to limit the invention, embodiments of the invention or other matter disclosed under the headings.

Predicting and optimizing resource utilization for a data processing job is important in designing the job and in designing hardware or planning for upgrades to hardware. Herein described are advances, including advances in automation with regard to dynamic creation of configuration files for parallel data integration jobs, resulting in more nearly optimal job configurations, which tend to improve execution and more closely match job configurations to job resource requirements.

U.S. Patent Publication 2008/0114870 A1 describes how to estimate and predict job resource utilization based on data flow and input data, and encompasses a resource estimation tool used at job design time. The tool allows a user to find a more nearly optimal data flow for data transformation logic, such as, for example, logic to merge multiple data streams into one. In one such application of the tool, a user designs three alternative jobs that merge multiple data streams, where the alternative jobs use join, merge, or lookup operators, respectively. The user directs the resource estimation tool to automatically estimate resource utilization (such as CPU, disk usage, scratch usage) for each job and compare the estimated results in order to find a more nearly optimal design, which depends upon the entire transformation logic of the job as well as the volume of input data, among other things. Once a job design has been finalized, if different input data arises the user can direct the tool to predict resource utilization, which is helpful for migrating a job among environments including development, test and production. This is particularly advantageous, because data volume may change greatly from one environment to another. Application of the tool can predict how much CPU processing, disk memory and scratch memory a job needs.

U.S. Patent application 2009/0193427 describes how to manage parallel data processing jobs in grid environments, which includes when and where to run each job. In one feature, the patent application discloses automatically generating a dynamic parallel configuration file responsive to a user specified criteria and available system resources. This is in contrast to past practices in which users manually creating parallel configuration files and attempting to leverage user effort merely by multiple users cooperatively storing such manually created files and sharing them with one another.

In the past, configuration files had to be manually created due to a variety of circumstances. For example, different environments may have different system characteristics. Consequently, configuration files created for one environment don't necessarily apply to another and cannot be migrated as part of job deployment, so new configuration files had to be created for each different environment. In addition, parallel configuration files had to be manually modified to accommodate any system changes, such as addition or removal of resource disks, nodes, and node pools etc.

U.S. Patent application 2009/0193427 mitigates the complexity of various steps needed for manually generating a configuration file, including defining node pools and storage resources, describing processing nodes, and managing the transformation of configuration files to different parallel execution environments.

Job design for parallel data integration includes two major components: data flow and parallel configuration. A data flow contains stages and links, and represents logic in terms of how data is transformed while flowing through from source to target. Parallel configuration indicates the number of partitions needed for processing data in parallel, and is usually specified in a configuration file. Separating data flow from parallel configuration, which is taught herein, allows the user to concentrate on logic during job design without worrying about parallelism. Once the logic has been verified, the user may then want to specify a parallel configuration file to meet job performance requirements.

Choosing an appropriate parallel configuration is very important, as it not only directly determines job performance, but also has great impact on efficiency and resource utilization of a parallel execution environment, especially when the environment is used as a shared IT resource for data integration at the enterprise level. However, it is difficult for users to create an optimized parallel configuration file due to a lack of knowledge about job resource requirements, such as whether the job consumes high/medium/low CPU, and what is the minimum requirement of disk space/scratch space/memory. Job resource requirements also depend on the size of input data and the number of processing partitions, which makes it even more difficult for the user to determine appropriate parallel configurations in various situations. To further complicate matters, an efficient parallel configuration when the job is the only one executing on the system may not be the same as when the job has to share those system resources with other executing jobs.

An inappropriate parallel configuration often leads to a series of problems. A job aborts because the system runs out of disk space, scratch space, or memory. A highly CPU-intensive job does not perform well due to fewer numbers of partitions being defined in the parallel configuration file than the job actually needs, or the same job meets its performance requirements but uses more partitions than absolutely necessary, causing some resources to be wasted. For example, running a job in 8-way parallel on 2 physical compute nodes may only improve performance by 10% compared to running 4-way parallel on 1 physical compute node. If resource utilization is an overriding consideration, then an appropriate configuration in this situation is to use 1 physical compute node with 4 logical partitions.

CPU consumption of the same job can also vary widely depending upon the size of input data set. If the data set is small for one execution and the user chooses to use 4 logical partitions on 1 physical compute node, the machine will be underutilized and resources will be wasted. An appropriate configuration here is to allow multiple low CPU-intensive jobs to run on the same physical node, to better utilize machine resources. Since there are many dynamic factors which influence the efficiency of jobs, the definition of an optimized parallel configuration file is problematic.

Even if the user manages to find an appropriate parallel configuration with the right number of physical compute nodes and the right number of logical partitions to meet job performance requirements, the configuration may not be optimal considering the overall efficiency of the entire parallel execution environment. As mentioned above, in traditional parallel data integration, a parallel configuration file is manually created by the user and saved on disk, and often shared by other users for different jobs. Using similar parallel configuration for different jobs may cause jobs to run on some machines not others. The system becomes unevenly loaded, leaving some machines overstressed, which unavoidably causes job performance degradation. In a grid environment, this process is improved by dynamic configuration file generation where a parallel configuration file is created on the fly using dynamically allocated resources. However, the improvement may be limited if the dynamic parallel configuration file is still based on a user-specified number of physical and logical partitions. As discussed above, the user often cannot provide the right number of partitions, and even so the parallel configuration may not be optimal from the system perspective. Therefore, it is very important to automatically define an optimal parallel configuration for each job based on its resource requirements and the availability of system resources.

In embodiments of the present invention, an automated system, method, computer program, computer program product or combinations thereof automatically categorize jobs in terms of CPU utilization in three different classes: high, medium, and low CPU consumption. Other resources such as memory, disk space, and scratch space are estimated and compared with the amount of resources provided by the execution environment. Each physical compute node is configured to run jobs from different classes. For example, a physical compute node can simultaneously execute one high CPU consumption job, or two medium CPU consumption jobs, or four low CPU consumption jobs, or a combination such as one medium and two low CPU consumption jobs. An optimal parallel configuration is automatically generated for the user based on historical runtime statistics, job resource requirements and the availability of system resources. Responsive to the generated configuration, resources are automatically allocated, which frees users from manually specifying the number of physical and logical partitions and improves individual job performance. This also tends to better utilize system resources, which tends to improve overall efficiency of the entire parallel execution environment (e.g., SMP, MPP, cluster, or grid environment) for data integration.

With reference now toFIG. 1, a pictorial representation of a network data processing system100is presented in which the present invention may be implemented. Network data processing system100contains a network102, which is the medium used to provide communications links between various devices and computers connected together within network data processing system100. Network102may include connections, such as wire, wireless communication links, or fiber optic cables etc.

In the depicted example, server104is connected to network102along with storage unit106. In addition, clients108,110, and112are connected to network102. These clients108,110, and112may be, for example, personal computers or network computers. In the depicted example, server104provides data, such as boot files, operating system images, and applications to clients108,110and112. Clients108,110and112are clients to server104. Network data processing system100may include additional servers, clients, and other devices not shown. In the depicted example, network data processing system100is the Internet with network102representing a worldwide collection of networks and gateways that use the TCP/IP suite of protocols to communicate with one another.

Peripheral component interconnect (PCI) bus bridge214connected to I/O bus212provides an interface to PCI local bus216. A number of modems may be connected to PCI local bus216. Typical PCI bus implementations will support four PCI expansion slots or add-in connectors. Communications links to network computers108,110and112inFIG. 1may be provided through modem218and network adapter220connected to PCI local bus216through add-in boards. Additional PCI bus bridges222and224provide interfaces for additional PCI local buses226and228, from which additional modems or network adapters may be supported. In this manner, data processing system200allows connections to multiple network computers. A memory-mapped graphics adapter230and hard disk232may also be connected to I/O bus212as depicted, either directly or indirectly.

Server104may provide a suitable website or other internet-based graphical user interface accessible by users to enable user interaction for aspects of an embodiment of the present invention. In one embodiment, Netscape web server, IBM Websphere Internet tools suite, an IBM DB2-UDB database platform and a Sybase database platform are used in conjunction with a Sun Solaris operating system platform. Additionally, components such as JBDC drivers, IBM connection pooling and IBM MQ series connection methods may be used to provide data access to several sources. The term webpage as it is used herein is not meant to limit the type of documents and applications that might be used to interact with the user. For example, a typical website might include, in addition to standard HTML documents, various forms, Java applets, Javascript, active server pages (ASP), Java Server Pages (JSP), common gateway interface scripts (CGI), extensible markup language (XML), dynamic HTML, cascading style sheets (CSS), helper applications, plug-ins, and the like.

With reference now toFIG. 3, a block diagram illustrating a data processing system is depicted in which aspects of an embodiment of the invention may be implemented. Data processing system300is an example of a client computer. Data processing system300employs a peripheral component interconnect (PCI) local bus architecture. Although the depicted example employs a PCI bus, other bus architectures such as Accelerated Graphics Port (AGP) and Industry Standard Architecture (ISA) may be used. Processor302and main memory304are connected to PCI local bus306through PCI bridge308. PCI bridge308also may include an integrated memory controller and cache memory for processor302. Additional connections to PCI local bus306may be made through direct component interconnection or through add-in boards. In the depicted example, local area network (LAN) adapter310, Small computer system interface (SCSI) host bus adapter312, and expansion bus interface314are connected to PCI local bus306by direct component connection. In contrast, audio adapter316, graphics adapter318, and audio/video adapter319are connected to PCI local bus306by add-in boards inserted into expansion slots.

Expansion bus interface314provides a connection for a keyboard and mouse adapter320, modem322, and additional memory324. SCSI host bus adapter312provides a connection for hard disk drive326, tape drive328, and CD-ROM drive330. Typical PCI local bus implementations will support three or four PCI expansion slots or add-in connectors.

As another example, data processing system300may be a stand-alone system configured to be bootable without relying on some type of network communication interface, whether or not data processing system300comprises some type of network communication interface. As a further example, data processing system300may be a Personal Digital Assistant (PDA) device, which is configured with ROM and/or flash ROM in order to provide non-volatile memory for storing operating system files and/or user-generated data.

The depicted example inFIG. 3and above-described examples are not meant to imply architectural limitations. For example, data processing system300may also be a notebook computer or hand held computer as well as a PDA. Further, data processing system300may also be a kiosk or a Web appliance. Further, the present invention may reside on any data storage medium (i.e., floppy disk, compact disk, hard disk, tape, ROM, RAM, etc.) used by a computer system. (The terms “computer,” “system,” “computer system,” and “data processing system” and are used interchangeably herein.)

One or more databases may be included in a host for storing and providing access to data for the various implementations. One skilled in the art will also appreciate that, for security reasons, any databases, systems, or components of the present invention may include any combination of databases or components at a single location or at multiple locations, wherein each database or system includes any of various suitable security features, such as firewalls, access codes, encryption, de-encryption and the like.

The database may be any type of database, such as relational, hierarchical, object-oriented, and/or the like. Common database products that may be used to implement the databases include DB2 by IBM (White Plains, N.Y.), any of the database products available from Oracle Corporation (Redwood Shores, Calif.), Microsoft Access by Microsoft Corporation (Redmond, Wash.), or any other database product. The database may be organized in any suitable manner, including as data tables or lookup tables.

Association of certain data may be accomplished through any data association technique known and practiced in the art. For example, the association may be accomplished either manually or automatically. Automatic association techniques may include, for example, a database search, a database merge, GREP, AGREP, SQL, and/or the like. The association step may be accomplished by a database merge function, for example, using a key field in each of the manufacturer and retailer data tables. A key field partitions the database according to the high-level class of objects defined by the key field. For example, a certain class may be designated as a key field in both the first data table and the second data table, and the two data tables may then be merged on the basis of the class data in the key field. In this embodiment, the data corresponding to the key field in each of the merged data tables is preferably the same. However, data tables having similar, though not identical, data in the key fields may also be merged by using AGREP, for example.

Modules implemented in software for execution by various types of processors may, for instance, comprise one or more physical or logical blocks of computer instructions which may, for instance, be organized as an object, procedure, or function. Nevertheless, the executables of an identified module need not be physically located together, but may comprise disparate instructions stored in different locations which, when joined logically together, comprise the module and achieve the stated purpose for the module. Indeed, a module of executable code may be a single instruction, or many instructions, and may even be distributed over several different code segments, among different programs, and across several memory devices. Similarly, operational data may be identified and illustrated herein within modules, and may be embodied in any suitable form and organized within any suitable type of data structure. The operational data may be collected as a single data set, or may be distributed over different locations including over different storage devices, and may exist, at least partially, merely as electronic signals on a system or network.

Referring now toFIG. 4, a resource optimization system400is shown, according to an embodiment of the present invention. System400includes grid resources401. Grid resources401include a file server405, a database server404and first and second blade centers402and403, each having a plurality of servers, also referred to as server blades. Blade centers402and403may include more or less server blades in other implementations.

Each server blade has all essential functional components of a computer (e.g., processor(s), memor(ies), and so forth). Non-essential components, such as power, cooling, networking, and so forth, are provided through an enclosure (not depicted) of each blade center. Each blade center may include more or less server blades in other implementations.

System400(which may also be referred to herein as a “grid” or a “grid environment”) may include more or less blade centers, server blades, file servers, and database servers. Additionally, the grid may include other resources, such as storages (not depicted).

Each server, server blade, or the like in grid environment400may be referred to as a ‘physical node.’ Within each ‘physical node’, there may be one or more ‘logical nodes’ or ‘partitions’. For example, a server blade with dual processors can include two logical nodes, one node running on each processor. The term ‘resource node’ will be also used herein to refer to any physical or logical node in a grid environment.

In an embodiment of the invention, some of the physical nodes in grid environment400, such as certain server blades, are designated as ‘compute nodes’ to be used exclusively for computing purposes. Compute nodes are considered to be dynamic grid resources as they are dynamically allocated to jobs at runtime. In contrast, fixed-name servers, such as file server106and database server108, are considered to be static resources, also referred herein to as ‘static nodes’. Disk information in each compute node is also considered to be a static resource.

Typically, configuration files are manually created prior to deployment (e.g., before runtime), such as while designing the parallel data processing jobs. Creating configuration files before runtime, however, is not possible in a grid environment because certain system resources are dynamically allocated at runtime. Hence, not all system resources to be used for executing a parallel data processing job in the grid environment will be known prior to deployment. As a result, there is a very short window of time in which to create a configuration file for the parallel data processing job (e.g., after resource allocation and before execution).

System resources specified in configuration files are usually machine names of the system resources. Consequently, a configuration file created for a parallel data processing job in one environment generally cannot be used in another environment. To given an example, suppose a configuration file is created for a parallel data processing job in an environment with one set of servers. If the parallel data processing job is to be executed in another environment with a different set of servers, then a new configuration file will have to be created because the machine names for the servers will be different.

System400also includes an application406for managing parallel data processing jobs in system400. In an embodiment of the invention, application406runs on one or more of the servers of one or more of the blade centers402or403. Application406includes a job run request module410, a job controller420, a resource optimizer430, a resource manager440, a parallel engine450, and a repository460. In one implementation, job run request module410, job controller420, resource optimizer430, resource manager440, parallel engine450, and repository460may be stand-alone applications rather than being included in a single application406. The invention includes embodiments in which repository460stores information on a computer readable medium of one or more of the following: file server405, database server404, and one or more of the servers of blade center402or403, possibly including the same server running application406.

Application406generates a user interface, which may be either a graphical user interface (“GUI” or a command line interface. Responsive to user input, job run request module410causes a job and the job's associated properties to be stored in a computer readable media. Likewise, job run request module410causes storage of job execution information, such as an indication of whether system400should perform parallel execution of a job, and, if so, parallel configuration information. In one implementation in a non-grid environment, parallel information is defined in a pre-created static configuration file. In one implementation in a grid environment that does not support any resource optimization mechanisms; parallel information includes the number of requested compute nodes and the number of logical partitions. In one grid environment embodiment of the present invention, parallel information includes a flag indicating that an optimal parallel configuration needs to be dynamically generated prior to job execution. The user can still choose to either use a predefined static configuration file or specify job resource requirements, as mentioned above. This makes system400flexible enough to handle different needs.

Job run request module410sends job execution information to front end module424of job controller420responsive to a command from either a GUI or command line interface. If the job execution information includes a pre-created static parallel configuration file, module424parses the configuration file to determine the physical machine names and explicitly reserves those machines via resource manager440. If the job uses a dynamic configuration file, module424asks resource manager440to allocate the requested number of compute nodes and generates a configuration file after the resources have become available.

In one implementation, a user wishes a job to use resource optimization mechanisms and initiates this via user interaction with job run request module410. (In one implementation, resource optimization may be a default condition.) In response to such an indication of resource optimization, job controller420calls resource optimizer430to obtain system resources utilization and optimized resource requirements of each job. Before job controller420submits a job to resource manager430, resource requirement module428reviews job resource requirements and compares those requirements with system400resources that are available. Module428issues a warning to job run request module410if the minimum resources required to run the job exceed the available system capability. The job run request is returned in this case. The user can submit another run request after having fixed system resource problems. If system400is capable of accommodating the job resource requirements, module428submits the job to resource manager440for dynamic resource allocation and job execution.

Resource optimizer430creates statistical models using performance data collected from previous job runs to determine the optimized job resource requirements. In an implementation, the user has three options while working with resource optimizer430: pre-creating models before running jobs, creating models while jobs are running, and updating models in a batch process every so often (e.g. every week). Each job can have multiple models associated with it, and those models are saved in a model repository. At run time, resource optimizer430receives resource optimization requests sent from job controller420and retrieves the model associated with the job, estimates job resource utilization, determines CPU categories and optimal number of partitions based on estimated resource utilization, and returns the statistics to job controller420. Resource optimizer430also checks the capacity of the system resources, and sends the information back to job controller420along with the statistics that represents job resource requirements.

Resource manager440dynamically determines and allocates system resources for each job based on job resource requirements. This includes resource manager440placing a job into a waiting queue until the resources needed to run the job become available. Resource manager440then starts the job and locks the resources for the running job until the job has finished. Once a job has finished, resource manager440frees the resources allocated and gives those resources to a next job that has similar resource requirements. Because resource manager440can handle resource contention more efficiently, job performance and overall efficiency of the entire parallel execution environment is improved.

Parallel engine module450provides rich functionality for data integration, transformation, and movement. It is built on top of a parallel framework where jobs can run with both pipelined and data-partitioned parallelism.

The following describes how job resource requirements may be defined and what may be included in a job run request that is produced by module410, according to an embodiment of the present invention. This is followed by a description of how both job controller420and resource optimizer430may be implemented and integrated to more nearly optimize resource utilization for parallel data integration on grid, according to an embodiment of the present invention.

Job Resource Requirements

Concerning job resource requirements, a parallel job requires a certain amount of machine resources in order to run successfully. These resources include CPU, memory, disk space, and scratch space, as previously mentioned. The amount of resources needed may vary depending on the type of the job and the amount of input data to be processed. In a non-grid environment, or a grid environment lacking resource optimization mechanisms, job resource requirements must be submitted as part of a job run request. In an embodiment of the present invention, resource requirements are dynamically determined and allocated for each job. In such an embodiment, the user no longer needs to submit those requirements as part of a job run, but may do so. This manual feature provides backward compatibility and usage flexibility, an embodiment of the current invention continues to support user-defined job resource requirements.

In one implementation, a GUI generated by application406for user interaction with job run request module410presents a screen to a user having a checkbox for enabling or disabling resource optimization and automatic configuration file generation at a project level. In one implementation, the checkbox is selected by default, meaning the resource requirements of each job in the same project are automatically determined by application406based on automatic resource optimization mechanisms described herein.

The GUI also presents a job properties screen. The default status of the checkbox can be overwritten at the job level by enabling a grid properties tab on the job properties screen and specifying resource requirements on the resulting grid properties screen. A CPU properties tab is included on the grid properties screen to let the user manually assign the job to an appropriate CPU utilization category, i.e., high, medium, or low CPU utilization, in the particular embodiment of the invention that is illustrated herein. The user can de-select the checkbox, which causes application406to revert back to the automatic resource optimization mechanisms.

Job resource requirements include:

number of physical compute nodesnumber of logical partitions on each physical compute nodetype of CPU utilization: high/medium/low CPU consumptionCPU timejob execution timeamount of disk spaceamount of scratch spaceamount of memory
Job Run Request

Concerning a job run request, the request can be sent from job run request module410to job controller420responsive to a command via a GUI or command line interface. As mentioned earlier, in an embodiment of the present invention, a job run request contains data-flow description but not parallel configuration information about job resource requirements, since resource optimization mechanisms described herein automatically determine the resources that are needed to run the job and dynamically generate a parallel configuration file.

Job Control

Concerning job control, job controller420provides job compilation, job validation, job execution, configuration file generation, and job report. Mechanisms for job compilation, job validation, and configuration file generation are described in detail in one or more of the above referenced patent applications. These mechanisms apply regardless whether resource optimization features are enabled, i.e., features disclosed herein.

For automatic configuration file generation in an embodiment of the present invention, if the user chooses to use sequence-level resource configuration for all parallel jobs (e.g., one configuration file for all parallel jobs within a user-specified sequence of jobs), resource optimizer430automatically designates a number of physical compute nodes for the job sequence, where the designated number is equal to the maximum number of physical compute nodes determined among all the jobs in the sequence. Likewise, resource optimizer430automatically designates a number of partitions, scratch space, and disk space required by a job sequence based on the maximum numbers of these resources requested among the jobs.

For each job, resource optimizer430generates a job report stored in repository460, which includes a message indicating a CPU category for the job, e.g., high, medium or low CPU consumption.

Job controller410reads in the job report generated by resource optimizer430and requests the indicated number of physical nodes from resource manager440. Once the resources become available, job controller410generates a parallel configuration file. This configuration file contains the number of physical nodes and the number of logical partitions per physical node determined by resource optimizer430.

FIG. 5shows a flow chart500of job execution sequences when resource optimization is enabled on grid401, according to an embodiment of the present invention.

Referring to bothFIGS. 4 and 5, job controller420receives and parses502a job run request410and determines a next step based on actions specified in request410. Responsive to determining504that job run request410says to use resource optimizer430, job controller420calls resource optimizer430, which, in turn, determines506job resource requirements. Once job controller410receives job resource requirements returned from resource optimizer430, job controller410compares510the requirements with available system resources. If job resource requirements can be met, job controller420then calls resource manager440, which, in turn, asks516for resources. Once the requested resources become available, job controller420obtains resources518and generates520parallel configuration file. Resource manager440then starts job execution522.

If the job run request410does not say to use resource optimizer430, then responsive to job controller420determining508that job run request410indicates resource manager440should be used, job controller420calls resource manager440, which, in turn, asks516for resources as specified in job run request410, and flow500proceeds as described above.

If job controller420determines504that job run request410does not indicate to use resource optimizer430, and determines508that job run request410does not indicate to use resource manager440, job controller420then starts job execution522.

If job controller420determines504that job run request410indicates to use resource optimizer430and if estimated job resource requirements cannot be met, as determined by job controller420comparing510job resource requirements to availability, job controller420warns512the user. If job controller420determines512that job run request410says to run the job anyway, job controller420starts job execution522(without calling resource manager440). If job controller420determines512that job run request410says not to run if there are not enough resources, job controller420simply returns524job run request410without executing the job.

Resource Optimizer

According to an embodiment of the present invention, resource optimizer430provides features that operate in cooperation with, and in addition to, the previously disclosed resource estimator tool in order to provide resource optimization at a job level. (Resource optimization at a system level is handled by resource manager440.) Particularly, the resource estimation tool is shown in and described as resource estimation tool 124 in FIGS. 1 and 2 of related United States Published Patent Application 2008/0114870, among other places. Accordingly, it should be understood that resource optimizer430of the present patent application is herein added to what is shown as resource estimation tool 124 in FIGS. 1 and 2 of the 2008/0114870 application. Consequently, in addition to what is disclosed herein, resource optimizer430also receives the inputs of and provides the functions and outputs of the earlier disclosed resource estimation tool 124. References herein to resource optimizer430should be understood to include reference to resource estimation functions described in the 2008/0114870 application, so that FIGS. 1 and 2 of the 2008/0114870 application are particularly relevant for understanding inputs and outputs of resource optimizer430.

FIG. 6illustrates work flow600of resource optimizer430, according to an embodiment of the present invention.

Resource optimizer430work flow600begins at receiving602a user request for a job. (The user request for a job is initiated via user interaction with job run request module410in one embodiment of the invention.) Three major modes of action are included in work flow600, which a user can choose to use one at a time. The modes include build model604, optimize resources606, and check available system resources608. When a user selects to build an analytic model604, the user indicates at610whether the model should first collect performance data. If no, resource optimizer430builds the model using pre-collected performance data, so resource optimizer430first imports data612, analyzes data614, then creates the model616based on data analysis614and saves the model618. If performance data is not provided, resource optimizer430first runs job620, collecting and saving some performance data622. Resource optimizer430then analyzes the data614from collecting and saving622, creates616the model based on data analysis614and saves618the model. The performance data collected in this mode can be used later together with other performance data to refine a model.

If the user selects to optimize resources606, a model is imported630. The user may specify an existing model that can be used at630to perform such optimizations. If a model is not explicitly specified, a model that has been built most recently is obtained at630. A report on job resource requirements is generated634once resource optimization632has been performed. A third mode is to check system resources608. In this mode, resource optimizer430may invoke an external component/tool (e.g. resource tracker640) to analyze system resource statistics642and generate a report644, which provides a snapshot of the system and its capacity.

Resource Optimization Request

A request602can contain some or all of the following directives:

action mode: build, optimize, or checkmodel type: analyticmodel nameperformance data: the location where performance data is storedjob report namesystem report namedata flow description

Not all directives and options are required for each request602. When a directive or an option is omitted, a default value is assumed. The default action mode is optimize, default modeling type is analytic, default model name and job report name are generated based on the job name, and default system report name is stdout. The data flow description must be present if the mode is build.

Performance data can be normally stored in a file.

Performance Data

Performance data is used as sample data by resource optimizer430, in an embodiment of the present invention, for building a model that is capable of estimating job resource utilization when presented with information for a new job. Resource optimizer430collects the performance data at three different levels while a job is running on each partition: link level, operator level, and job level.

Performance data at the link level per partition includes:

Performance data can be collected from run to run with the same number of partitions, or with increasingly varying number of partitions. Performance data collected from different parallel configurations is useful for determining the optimal parallelism that helps best utilize system resources. There are many different ways to store performance data. Serializing data and saving data to disk is one way.

Analytic Model

Resource optimization module430is capable of creating a number of different models in a hierarchical structure as illustrated inFIG. 7, including a base model710at the simplest level of the hierarchy. More complex models include a static model720, a dynamic model730and an analytic model740, which is a particular type of model provided according to an embodiment of the present invention. Analytic model740differs from other models in that it analyzes job run-time characteristics based on performance data collected from previous runs over time, whereas static model720does not use any performance data and dynamic model730uses segments of performance data from one job run. Compared to models720and730, analytical model740is more flexible and accurate.

Model Description

Analytic model740extends dynamic model730, in part because model740is based on performance data from job runs with a complete set of input data, not segments of input data. Resource optimizer430can determine correlation coefficients for analytic model740using a multiple linear regression and least squares method, which is discussed for a dynamic model730in one of the incorporated patent applications. To produce the coefficients for a model740, resource optimizer740considers multiple job runs as being one job run having multiple segments, where each segment contains a complete set of input data.

In an embodiment of the invention, if performance data contains statistics collected from different parallel configurations, resource optimizer430uses analytic model740to compute an optimal number of physical compute nodes based on the collected performance data and a predetermined optimization criterion, which may be user-selected, such as shortest execution time, least amount of CPU time per partition, or shortest runtime of the batch of jobs.

In one instance of a high CPU job, which is a category explained herein below, using model740resource optimizer430estimates that execution time would be 20 minutes if the job is run on 1 physical compute node, 10 minutes on 2 physical compute nodes, and 6 minutes on 4 physical compute nodes. If specified performance requirements include a requirement to finish the job within 6 minutes, resource optimizer430determines that the configuration with 4 nodes is optimal, because it meets the specified execution time performance requirement.

If execution time is not so limited by specified performance requirements, resource optimizer430determines that the optimal number of compute nodes is 2, in an embodiment of the invention. This is because in the embodiment of the invention, if there is no specified maximum execution time, resource optimizer430seeks to optimize resource usage tradeoffs between resource utilization and job performance by minimizing a particular measure of resource utilization, e.g., execution time×number of nodes (node-minutes), wherein if there are configurations having equal overall execution time, resource optimizer430selects the configuration with the least number of nodes. In the illustrated instance, the number of nodes×execution time=20 node-minutes for the 1-node configuration, 20 node-minutes for the 2-node configuration, and 24 node-minutes for the 4-node configuration. However, since the job is a high CPU job, resource optimizer430does not select a single node configuration, in this embodiment of the invention. Therefore, in the illustrated instance, resource optimizer430selects the 2-node configuration, which achieves a little better node-minute execution time than the 4-node configuration and does so with fewer nodes.

In an alternative, resource optimizer430seeks to optimize the tradeoff between resource usage and job performance by minimizing node-minutes, wherein if there are configurations having equal node-minutes, resource optimizer430selects the configuration with the least number of nodes. For this alternative rule, in the illustrated instance, resource optimizer430would select the 1-node configuration, which achieves the same, 20 node-minute execution time as the 2 node configuration, but does so with less nodes.

Major steps associated with creating an analytical model (616FIG. 6) include:

Step: Determine segments for holding data from all available performance data files.

Step: For each performance data file:

Read performance data.

Parse performance data.

Store parsed performance data as data objects in respective segments.

Step: Calculate resource usage using parsed performance data for each one of the segments. For example, in one embodiment of the invention resource usage includes respective CPU times and execution times for the respective segments.

Step: Determine correlation coefficients using the multiple linear regression and least square method discussed in one of the incorporated patent applications for all statistics collected in performance data. That is, the collected performance data provides instances of observed performance, and includes data that the model assumes are for performance causes, e.g., input data size and number of input records on each link, etc., and data that the model assumes are for performance effects, e.g., data throughput, CPU time, etc. A correlation coefficient is calculated for each performance cause in the model responsive to the cause and effect performance data. (See list of performance data herein above.)

Step: Save model by serializing the correlation coefficients to disk.

Like other types of models (static model720and dynamic mode730), analytic model740can be used to predict, i.e., estimate, job resource utilization for any given input data size based on known correlation coefficients. (Refer to list of performance data herein above for statistics the analytic model can predict.)

Analytic model740and its use by resource optimizer430are distinguishable from static model720and dynamic model730in that analytic model740can optimize resources based on the estimated resource utilization that the model predicts. In particular, resource optimizer430uses analytic model740to predict resource utilization and performance, e.g., CPU times and execution times for a job, operators and groups of operators, then computes CPU indices for the same. Then resource optimizer430uses model740to determine CPU categories of the respective CPU indices and uses this category information to optimize resources, e.g., determine an optimal number of partitions for each operator, group of operators or job. Further details about this resource optimization methodology are described herein below.

Model Repository

A job can have more than one analytic model740. But only one of the models740can be used for the job each time the job's resource requirements are estimated. Models740can be stored in different ways. One way is to store to disk correlation coefficients and other model-related objects in a series for a corresponding series of models.

Model Deployment

Models740are exported as part of job export for deployment purposes. Then, when jobs are imported into a different environment, models740are moved into this environment as part of this process. A model740built in one environment can be used in another environment running the same configuration (both hardware and software).Resource Optimization Methodology

In the exemplary work flow600ofFIG. 6, resource optimization632is a second major functionality provided by resource optimizer430, according to an embodiment of the present invention. Resource optimizer430module predicts job resource utilization using an analytic model740that is most recently built or that provided by the user, as previously mentioned. Data can be imported from various sources, including file, datasets, file sets, database tables, or the output from another program. Resource optimization632for a job includes determining input data size for each data source used by the job. Resource optimization632next includes analytic model740using this information as an input to compute resource requirements.

Resource optimization632not only predicts job resource utilization (e.g., CPU time and execution time, in one embodiment of the invention) but also makes intelligent decisions by classifying predicted job resource utilization (e.g., high, medium or low CPU usage, in one embodiment of the invention) and then using this information to determine resource requirements (e.g., number and allocation of nodes and partitions, in one embodiment of the invention).

Regarding the classifying, in one implementation, resource optimization632uses a ratio of CPU time to execution time of the job. If the ratio is greater than 0.6, resource optimization632categorizes the job as a high CPU job. If the ratio is greater than 0.3 but less than 0.6, resource optimization632categorizes the job as a medium CPU job. Otherwise, resource optimization632categorizes the job as a low CPU job. In another implementation, resource optimization632defines a range of the CPU time for jobs in different categories. For low CPU jobs, the range of the CPU time varies from 0 to 300 seconds, in one instance. For medium CPU jobs, the range varies from 300 to 900 seconds, in one instance. Any jobs that consume more than 900 seconds of CPU are considered high CPU jobs in one instance.

In the job resource report634generated by resource optimizer430, there is an attribute called “class” which shows the class the job belongs to in terms of its CPU consumption. The value of this attribute is one of high, medium, or low, as described above.

For low or medium CPU jobs, it is optimal to configure each such job to run on one physical compute node. Therefore, for each low or medium CPU job, resource optimization632indicates on the generated job report that the job should run on only one physical computer node.

Further, since a single low or medium CPU job cannot consume all CPU of a physical compute node, it is optimal to run multiple such jobs on the same physical compute node. One example is to run 4 low CPU jobs, 2 medium CPU jobs, or 1 high CPU job on one physical compute node. For an extremely high CPU job (the ratio between CPU time and execution time is greater than 0.95), the job needs to run across multiple physical compute nodes.

Referring now toFIG. 9, a process900of resource optimization632is illustrated, according to an embodiment of the invention. Resource optimization632determines910the number of physical nodes and determines920the number of logical partitions on each physical node for a job and outputs this in a job report. In one embodiment of the present invention, an algorithm to determine the optimal number of physical compute nodes for an extremely high CPU job includes estimating CPU time for a job running on 1, 2, and 4 nodes, respectively. If the CPU ratio (also referred to herein as “CPU index”) is greater than 0.8, meaning the job is high CPU on each compute node, resource optimizer430then adds more nodes in estimating job resources until the CPU ratio is lower than 0.8.

Resource optimization632also determines930a number of partitions per physical node for each operator and also outputs this in the job report.

Attributes in the job report indicate, respectively, the number of physical nodes for the job, the number of logical partitions for the job on each physical node, and the number of logical partitions for each operator of the job on each physical node.

To determine920the number of logical partitions allocated to the job for each physical node, resource optimization632uses approach 3 described herein below under “Optimal Number of Partitions.” To determine930the number of logical partitions for each operator, resource optimization632uses approaches 1 and 2. The number of logical partitions determined930at the operator level overrides that determined920at the job level, except that if the value of an attribute for a number of logical partitions is zero for a particular operator on a particular node, then process930uses for this operator on that particular node the number of logical partitions determined920for the job on that particular node.

In one of the related patent applications, a data flow is described that has one or more of three types of runable groups: input group, vertical group, and scratch group. In an embodiment of the present invention, a different perspective is introduced in which a data graph is also considered to include runable groups of four types: data source group, processing group, scratch group, and data sink group. A data source group contains all data sources. A data sink group contains all data sinks. There is only one data source group and one data sink group per job. A scratch group contains sort and buffer operators where data may end up being saved on disk temporarily during run time. A processing group contains operators that perform more CPU intensive operations. A scratch group is similar to a data source or a data sink group in that it generates a lot of I/O operations. A scratch group is also similar to a processing group in that it also needs a lot of CPU for processing data. In an embodiment of the present invention, processing groups are the same as the vertical groups described previously. Accordingly, methods for constructing both processing and scratch groups are as described previously.

In an embodiment of the present invention, both ways of defining groups are used, i.e., the first way of defining groups that may include the three group types mentioned above, and the second way of defining groups that may include the four group types mentioned above. Resource optimization632uses the first way for predicting resource utilization, as described in one of the related patent applications. Resource optimization632uses the second way for optimizing resources, e.g., determining the CPU category and the optimal number of logical partitions on each physical compute node. Input groups in the first way are referred to as data sources in the second. Data sinks are part of vertical groups in the first way, but are a stand-alone group in the second. Vertical groups other than data sinks in the first way are referred to as processing groups in the second.

Execution Time

The elapsed time for running a job is the sum of three parts: startup time, execution time, and shutdown time. The startup time is the time for the job to set up the parallel execution environment, including creating section leaders on each partition, instantiating operators on each partition, and setting up data connections among operators on all partitions, etc. The shutdown time is the time from the moment all operators from all partitions finish processing data until the job finishes cleaning up the parallel execution environment. The execution time is the time from the moment the data source operators start to process the first record until the moment the data sink operators have finished processing all records. Since the data sink group is the last group to run to completion, the maximum execution time among operators in the data sink group represents the execution time of the job. Because a job runs in parallel, the number of records needed to be processed on each partition is different, which means the execution time can be different on each partition. The execution time of the job is the maximum execution time among all partitions. The startup time and shutdown time are usually short, so only the execution time is considered here.

CPU Category

As previously mentioned, resource optimization module632will automatically classify a job in one of three CPU categories: high CPU, medium CPU, and low CPU. The user can also explicitly assign a CPU category to a job. (The user may do this by the user interface to job run request module410ofFIG. 4.) If the user specifies a CPU category, job controller420uses this category. Otherwise, resource optimization module632automatically determines a job's CPU category based on estimated resource utilization, which is determined using analytic model740. Additional details for automatically determining a job's CPU category are next described.

In various embodiments of the present invention, there are various different ways of determining a job's CPU category. In one embodiment of the present invention, resource optimization module632uses a CPU index to help determine whether or not a job is high CPU, medium CPU, or low CPU. Three ways to define the CPU index of a job are described below. Resource optimization module632uses all three ways to determine the CPU category of the job, compares the results produced, and designates the highest result as the CPU category for the job. Resource manager440uses this category to determine the number of jobs on a node and resource optimizer430uses this category to determine the number of physical nodes for a job.

Approach 1—CPU Index at Operator Level

In a first way of determining CPU classification for a job, resource optimization module632uses the CPU index of the most CPU-intensive operator to determine the CPU index of the job. Let CTiprepresent the estimated CPU time of operator i on partition p, ETiprepresents the estimated execution time of operator i on partition p, the CPU index for operator i on partition p can be written as:
Icpuip=CTip/ETip(1)

The CPU index for operator i is:
Icpui=max{CTi0/ETi0, . . . , CTip/ETip, . . . , CTip-1/ETip-1}  (2)
where P is the number of logical partitions that the operator i runs on.

The CPU index of a runable group is represented by the maximum CPU index among all operators of that group.
Icpug=max{Icpu0g, . . . , Icpuig, . . . , IcpuI-1g}  (3)
where I is the number of operators in group g.

The CPU index of the job is represented by the maximum CPU index among all runable groups.
Icpu=max{Icpu0, . . . , Icpug, . . . , IcpuG-1}  (4)
where G is the number of runable groups of the job.

The algorithm to determine the CPU index of the job is given as follows:

for each runable group gfor each operator i inside gfor each partition p among all partitions of operator icalculate Icpuipusing Eq. (1)determine Icpuiusing Eq. (2)determine Icpugusing Eq. (3)determine Icpu using Eq. (4)
determine the CPU category based on the CPU index of the job as follows:

If Icpu ≧ 0.6 , the job is a high CPU job.If 0.3 ≦ Icpu ≦ 0.6 , the job is a medium CPU job.If 0 ≦ Icpu ≦ 0.3 , the job is a low CPU job.
Approach 2—CPU Index at Group Level

It is possible that the job does not have one very CPU intensive stage, but one or multiple runable groups may be CPU intensive. In this case, it is necessary to use a different approach to define the CPU index and further to determine the CPU type of the job based on its CPU index. In a second way of determining CPU classification for a job, resource optimization module632defines the CPU index of the job as the ratio between the maximum estimated CPU time among processing and scratch groups and the maximum estimated CPU time among data source and data sink groups.

Let CTigrepresent the estimated CPU time of operator i in group g on all partitions:

Let CTgrepresent the estimated CPU time of group g on all partitions:

The maximum estimated CPU time among all processing and scratch groups is:
CTpsgmax=max{CTpsgo, . . . , CTpsgn, . . . , CTpsgM-1}  (7)
where psg represents processing and scratch groups, M the number of total processing and scratch groups of the job.

The maximum estimated CPU time among data source and data sink groups is:
CTssgmax=max{CTssgo, . . . , CTssgn, . . . , CTssgN-1}  (8)
where ssg represents data source and data sink groups, N the number of total data source and data sink groups of the job. The CPU index of the job is defined as:
Icpu=CTpsgmax/CTssgmax(9)

The algorithm to determine the CPU index of the job is given as follows:

for each runable group gfor each operator i inside gfor each partition p among all partitions of operator icalculate CTigusing Eq. (5)calculate CTgusing Eq. (6)determine CTpsgmaxusing Eq. (7)determine CTssgmaxusing Eq. (8)determine Icpu using Eq. (9)
determine the CPU category based on the CPU index of the job as follows:

If Icpu ≧ 2 , the job is a high CPU job.If 1 ≦ Icpu ≦ 2 , the job is a medium CPU job.If 0 ≦ Icpu ≦ 1 , the job is a low CPU job.
Approach 3—CPU Index at Job Level

It could also be possible that the job does not have any CPU intensive runable groups, but the job has a large number of running processes and the total CPU percentage of all processes is large. For this case, there is a third way of determining CPU classification for a job, in which resource optimization module632defines the CPU index of the job as the total CPU percentage of all processes.

Let CPiprepresent the estimated CPU percentage of operator i on partition p, CPigrepresent the estimated CPU percentage of operator i in group g on all partitions,

CPgthe estimated CPU percentage of group g on all partitions,

The CPU index of the job is the total estimated CPU percentage of all groups, as follows:

An algorithm to determine the CPU index of the job is as follows:

for eachrunable group gfor each operator i inside gfor each partition p among all partitions of operator icalculate CPigusing Eq. (10)calculate CPgusing Eq. (11)calculate Icpu using Eq. (12)
determine the CPU category based on the CPU index of the job as follows:

If Icpu ≧ 0.6 , the job is a high CPU job.If 0.3 ≦ Icpu ≦ 0.6 , the job is a medium CPU job.If 0 ≦ Icpu ≦ 0.3 , the job is a low CPU job.

Referring now toFIG. 10, process910ofFIG. 9is illustrated for an embodiment of the present invention, according to what is described herein above. Process910includes initializing1005a hypothetical grid configuration having a number of physical nodes N, predicting1010grid resource utilizations using analytical model740for a hypothetical run on the hypothetical configuration, as described herein above. Next, process910generates1015resource utilization indices for the respective operators responsive to the predicted grid resource utilizations. Then process910generates1020resource utilization indices for respective groups of the operators responsive to the resource utilization indices of the respective operator groups. Then process910generates1025a first resource utilization index for the job responsive to the resource utilization indices of the respective operator groups. Process910then classifies1027the first resource utilization index in a high, medium or low CPU usage category.

Next, process910generates a second resource utilization index for the job, which includes selecting1030a first maximum of the resource utilization indices for a first subset of the operator groups, selecting1035a second maximum of the resource utilization indices for a second subset of the operator groups, and computing1040a ratio of the first and second maxima. The ratio is designates as the second resource utilization index. Process910then classifies1042the second resource utilization index in a high, medium or low CPU usage category.

Next process910generates1045a third resource utilization index for the job responsive to a sum of the predicted grid resource utilizations for all the operators. Process910then classifies1047the third resource utilization index in a high, medium or low CPU usage category, and then selects1050the highest category from among the first, second and third CPU usage categories.

Next, process910determines1055whether the selected resource utilization index is high. If yes, process910increments1060the number of hypothetical nodes, and returns to once again predict1010grid resource utilizations, etc. If no, process910accepts the current number of nodes.

Optimal Number of Partitions

The user can explicitly specify the number of logical partitions per physical compute node for a job. If so, the user-specified number of logical partitions is applied to the job. Otherwise, the analytic model automatically determines the optimal number of logical partitions based on estimated resource utilization, in accordance with one or more of the following embodiments of the invention.

In a first approach, if the user specifies the expected execution time for the job, the number of logical partitions that each operator runs on is determined based on the expected execution time. In a second approach, if the expected execution time is not user specified, the analytic model uses the estimated data throughput of each link to determine the number of logical partitions that each operator runs on. In this approach, the data throughput on either end of the link is adjusted so that the link has a unified data throughput which helps prevent a bottleneck. In a third approach, if each link in the job already has a unified data throughput, the CPU category is used to determine the number of logical partitions for the job, so all operators in the job have the same number of logical partitions.

Approach 1—User-Specified Expected Job Execution Time

The user specifies the expected time window for processing a given amount of data. The expected execution time of the job is shorter than the estimated execution time of the job. Parallel configuration needs to be changed so that the job can finish within the expected execution time. The number of logical partitions for each operator is determined based on the expected execution time of the job as follows:Determine estimated execution time and CPU time for each operator on each partition based on the given input data size using the methodology from one of the incorporated applications.Find the maximum estimated execution time of the data sink group. Use this as the estimated execution time of the job.Calculate the ratio between the expected execution time of the job and the estimated execution time of the job:
r=E{tilde over (T)}/ET  (13)
where E{tilde over (T)} represents the expected execution time of the job and ET the estimated execution time of the job.Start from the data source group, iterate through every group.Apply Eq. (13) to each operator in each group to determine the expected execution time and expected CPU time for each operator on each partition:
E{tilde over (T)}ip=ETip*r(14)
C{tilde over (T)}ip=Icpuip*E{tilde over (T)}ip(15)
where E{tilde over (T)}iprepresents the expected execution time and C{tilde over (T)}ipthe expected CPU time for operator i on partition p.Find the minimum expected CPU time of operator i among all partitions, C{tilde over (T)}i,minpCalculate the total estimated CPU time of operator i on all partitions using Eq. (5)Determine the optimal number of logical partitions using the following formula:
{tilde over (P)}i=CTig/C{tilde over (T)}i,minp(16)
where {tilde over (P)}irepresents the optimal number of logical partitions for operator i. If {tilde over (P)}iis not an integer, it is rounded up to the closest integer.
Approach 2—Unifying Data Throughput

In a data flow graph, a link connects two operators, referred to herein as “link mates.” The operator on the upstream end of the link is called a producing operator, the operator on the downstream end of the link is called a consuming operator. The data throughput of the producing operator on this link is equal to the number of output records produced by the producing operator on this link divided by the estimated execution time of the producing operator. Similarly, the data throughput of the consuming operator on this link is equal to the number of input records consumed by the consuming operator on this link divided by the estimated execution time of the consuming operator.

Although the total number of records produced by the producing operator is the same as the total number of records consumed by its link mate, the consuming operator, the data throughputs of the link mates can be different because the estimated execution times of the operators can be different. A bottleneck may occur on this link if the data throughput of its consuming operator is less than the data throughput of its producing operator. The analytic model unifies the data throughput of the link by adjusting the number of logical partitions of the operator that has the smaller data throughput.

The number of logical partitions for each operator is determined based on the estimated data throughput as follows:Determine estimated execution time and CPU time for each operator on each partition based on the given input data size using the methodology from one of the incorporated applications.Determine expected execution time and CPU time for each operator on each partition using the following algorithm:

for each runable group gfor each operator i inside gfor each output link k of producing operator ifind consuming operator j of link kfind the input port number h at which link k connects to consumingoperator jdetermine expected execution time of consuming operator j on its input hE{tilde over (T)}jh= ETi             (17)where E{tilde over (T)}jhrepresents the expected execution time of consuming operatorj on its input hfor each runable group gfor each operator i inside gdetermine the expected execution time of operator i based on E{tilde over (T)}jhE{tilde over (T)}i,minp= min{E{tilde over (T)}i0,...,E{tilde over (T)}ih,...,E{tilde over (T)}iH−1}       (18)calculate minimum expected CPU time per partition based on CPU indexC{tilde over (T)}i,minp= Icpui* E{tilde over (T)}i,minp(19)calculate the total estimated CPU time for all partitions using Eq. (5)determine the optimal number of logical partitions using Eq. (16)

Data flow traverse can start from any group in both upstream and downstream directions following the algorithm given in step 2. When it starts from the data source group, data sink group, or one of scratch groups, data throughput is better balanced between I/O and processing operators which further improves the overall performance of the job.

Approach 3—CPU Category

If the estimated data throughput of each link is unified on both ends, the resource optimizer uses the CPU category in which the job is classified to determine the number of logical partitions for each operator.

If Icpui≧0.8, {tilde over (P)}iis set to the number of CPUs in the system.

If 0.4≦Icpui≦0.8, {tilde over (P)}iis set to half of the number of CPUs in the system (rounded up or down to the next whole number, depending on the embodiment of the invention).

If 0≦Icpui≦0.4, {tilde over (P)}iis set to 1.

Referring now toFIG. 11, processes920and930ofFIG. 9are illustrated for an embodiment of the present invention, according to what is described herein above. First, process920determines1110a number of partitions for each physical node responsive to the highest category selected (1050FIG. 10) for the job. Next, process930determines1115whether a there is user-specified expected execution time. If yes, then process930adjusts1120estimated resource utilizations for operators responsive to a ratio of user-specified job execution time to estimated job execution time. Then, process930determines1125a number of partitions for each operator responsive to a combined total of adjusted estimated resource utilizations on all the operator's partitions and a minimum of the adjusted estimated resource utilizations among all the operator's partitions.

If process930determines1115no, then (given that the job request includes a data graph of linked operators specifying a sequence of parallel data integration operations performed when the parallel data integration job is run, such that each operator has one or more respective link mates) process930traverses1130the data graph and increases1135numbers of partitions for operators having throughputs less than their respective link mates.

Next, process930determines1140if the number of logical partitions is zero for any particular operator on any particular node. If yes, then process930designates1145for this operator on that particular node the number of logical partitions determined920for the job on that particular node.

System Resource Utilization

System resource utilization is collected by resource optimizer430in terms of the following attributes:CPU percentagememoryswapnumber of running processesdisk spacescratch space

Resource optimizer430collects statistics on these system resources each time a job run request is submitted, and writes them to a system resource report. Job controller420compares the job resource requirements to the system resources that are available as indicated by the system resource report, and indicates in the job resource report whether or not the job has enough resources to run to complete.

FIG. 8depicts a flowchart of a deployment of a software process that includes an embodiment of the present invention. Step800begins the deployment software process. The first thing is to determine if there are any programs that will reside on a server or servers when the process software is executed802. If this is the case then the servers that will contain the executables are identified842. The process software for the server or servers is transferred directly to the servers' storage via FTP or some other protocol or by copying through the use of a shared file system844. The software process is then installed on the servers846.

Next, a determination is made on whether the software process is to be deployed by having users access the software process on a server or servers804. If the users are to access the software process on servers then the server addresses that will store the invention are identified806. A determination is made as to whether a proxy server is to be built820to store the software process. A proxy server is a server that sits between a client application, such as a Web browser, and a real server. It intercepts all requests to the real server to see if it can fulfill the requests itself. If not, it forwards the request to the real server. The two primary benefits of a proxy server are to improve performance and to filter requests. If a proxy server is required then the proxy server is installed824.

The software process is sent to the servers either via a protocol such as FTP or it is copied directly from the source files to the server files via file sharing822. Another embodiment would be to send a transaction to the servers that contained the software process and have the server process the transaction, then receive and copy the software process to the server's file system. Once the software process is stored at the servers, the users via their client computers, then access the software process on the servers and copy to their client computers file systems826. Another embodiment is to have the servers automatically copy the process software to each client and then run the installation program for the process software at each client computer. The user executes the program that installs the software process on his client computer840then exits the process816.

In step808a determination is made as to whether the software process is to be deployed by sending the software process to users via e-mail. The set of users where the software process will be deployed are identified together with the addresses of the user client computers810. The process software is sent via e-mail to each of the users' client computers in step528. The users then receive the e-mail530and then detach the process software from the e-mail to a directory on their client computers832. The user executes the program that installs the process software on his client computer840then exits the process816.

Lastly a determination is made on whether the process software will be sent directly to user directories on their client computers812. If so, the user directories are identified814. The process software is transferred directly to the user's client computer directory834. This can be done in several ways such as but not limited to sharing of the file system directories and then copying from the sender's file system to the recipient user's file system or alternatively using a transfer protocol such as File Transfer Protocol (FTP). The users access the directories on their client file systems in preparation for installing the process software836. The user executes the program that installs the process software on his client computer840then exits the process816.

It should be appreciated from what has been described herein above that one or more embodiments of the invention automate the dynamic creation of configuration files for parallel data integration, which results in more nearly optimal job configurations. This, in turn, tends to improve execution and more closely match job configurations to job resource requirements. In one aspect, parallel configuration is responsive to dynamic factors including job characteristics (data-flow graph, metadata, and input data frequency distribution), input data volume, batch processing requirements, other concurrent running jobs, and the availability of system resources.

Regarding job category, jobs are placed into different groups based on CPU consumption: extremely high CPU, high CPU, medium CPU, and low CPU. Resource allocation is based on a group the job belongs to and resources the job actually needs. System resources are better managed in this way, as well, which prevents the system from being either overloaded or under-utilized. Job resource requirements are no longer merely part of a job run request. Job resource requirements are dynamically determined and optimized. This allows the user to focus on job design and not worry about defining resource requirements merely to run the job.

By adding resource optimization logic to job run mechanisms, job resource requirements are determined after a job is invoked but before the job is submitted to a resource manager in at least one embodiment of the present invention. This is in contrast to a job running immediately once it is invoked by the user either from the GUI or the command line. This is also in contrast to a job being submitted directly to a resource manager immediately upon being invoked. Mechanisms are provided to check on system resources before the job run request is executed to make sure that the system has enough resources to run the job, which avoids the job aborting due to lack of disk space, scratch space, or memory.

In an embodiment of the present invention, an automated process analyzes job run-time characteristics according to an analytic model that is based on performance data collected from previous job runs over time. This is in contrast to a static model that does not use any performance data and a dynamic model that uses segments of performance data from one job run. The analytic model is created prior to job run, or when the job is running. The user can also schedule to update the analytic model in a batch process. Collected performance data is stored in a repository along with job resource requirements and dynamic factors that produce performance data.

Benefits, advantages and solutions to problems have been described above with regard to specific embodiments. However, the benefits, advantages, solutions to problems, and any element(s) that may cause any benefit, advantage, or solution to occur or become more pronounced are not to be construed as critical, required, or essential features or elements of any or all the claims.

Those skilled in the art having read this disclosure will recognize that changes and modifications may be made to the embodiments without departing from the scope of the present invention.

It should be appreciated that the particular implementations shown and described herein are illustrative of the invention and its best mode and are not intended to otherwise limit the scope of the present invention in any way. Other variations are within the scope of the following claims.

As used herein, the terms comprises, comprising, or any other variation thereof, are intended to cover a non-exclusive inclusion, such that a process, method, article, or apparatus that comprises a list of elements does not include only those elements but may include other elements not expressly listed or inherent to such process, method, article, or apparatus. Further, no element described herein is required for the practice of the invention unless expressly described as essential or critical.