System and method for scheduling jobs in distributed datacenters

FIELD

0001This disclosure relates to the field of distributed datacenters, and more particularly, to systems, devices, methods and computer-readable media for scheduling jobs in distributed datacenters.

BACKGROUND

0002In computing environments such as cloud computing or other distributed computing platforms, data and computational resources may be located at different networked or geo-distributed locations.

0003As the amount of data and the number of jobs scale, it can be a challenge to effectively distribute jobs across datacenters.

SUMMARY

0004In some situations, some embodiments of the present disclosure may consider tasks across multiple jobs when scheduling jobs for execution in a distributed computing environment.

0005In accordance with one aspect of the present disclosure, there is provided method for scheduling jobs in a distributed computing environment. The method includes: obtaining a set of task identifiers, each task identifier identifying a corresponding data processing task included in one of a plurality of jobs to be scheduled for execution at one of a plurality of data processing locations; from the jobs having unscheduled data processing tasks, selecting a job having a longest job completion time based on a shortest task completion time for the data processing tasks included in the selected job; selecting, from unscheduled data processing tasks for the selected job, a data processing task having a longest task completion time based on shortest task completion times for the unscheduled data processing tasks for the selected job; and scheduling the selected data processing task for execution at the data processing location of the plurality of data processing locations corresponding to the selected data processing task's shortest completion time and having available processing resources.

0006In some embodiments, the method includes: adjusting the completion times of all other unscheduled data processing tasks included in the selected job to have a completion time equal to the maximum of: the optimal completion time of the selected data processing task, and the completion time of the respective unscheduled data processing task.

0007In any of the previous embodiments, the method can include: until each data processing task identified in the set of task identifiers has been scheduled, repeating: updating the available processing resources to identify the resources to be consumed by the previously scheduled data processing task as unavailable; selecting the job of the set of jobs having unscheduled tasks and having the longest job completion time based on the shortest task completion time for the data processing tasks included in the selected job; selecting, from the unscheduled tasks for the selected job, the data processing task having the longest task completion time based on shortest task completion times for the unscheduled data processing tasks for the selected job; and scheduling the selected data processing task at the data processing location corresponding to the selected data processing task's shortest completion time.

0008In any of the previous embodiments, the method can include: for each data processing task identified in the set of task identifiers: determining input data transfer times to transfer input data for the data processing task to each of the plurality of data processing locations having available processing resources; and for each of the plurality of data processing locations having available processing resources, determining a completion time for the data processing task based on the corresponding input data transfer time.

0009In any of the previous embodiments, determining the input data transfer times for a single data processing task identified in the set of task identifiers can include: identifying a size and location of each input data for the single data processing task; determining a communication bandwidth between each input data location and each of the data processing locations; and for each of the data processing locations: determining the transfer time for each of the input data to the data processing location based on the size of the input data and the communication input data location and the data processing location; and selecting, from the transfer times for each of the input data to the data processing location, a largest transfer time as the input data transfer time for the single data processing task at the data processing location.

0010In any of the previous embodiments, the method can include: for each of the data processing tasks identified in the set of task identifiers, determining an execution time for the data processing task; and for each of the plurality of data processing locations having available processing resources, determining the completion time for the data processing task at the data processing location based on the corresponding input data transfer time and the corresponding execution time.

0011In any of the previous embodiments, identifying the execution time for the data processing task can include: identifying a type of the data processing task; based on a database of execution data for types of data processing tasks, determining the execution time for the data processing task based on the identified type, and a size of the input data for the data processing task.

0012In any of the previous embodiments, the method can include: populating one or more data structures representing completion time objectives and constraints based on the completion times for each of the data processing tasks, data processing location assignments parameters, and the available resources; wherein selecting the job having the longest optimal completion time includes solving a linear programming problem defined by the matrices.

0013In any of the previous embodiments, obtaining the set of task identifiers can include: receiving job execution requests until a scheduling trigger is detected; and upon detection of the scheduling trigger, identifying the set of data processing tasks to be scheduled from the job execution requests.

0014In any of the previous embodiments, the scheduling trigger can be detected when a defined time period has elapsed, or when a number of received jobs execution requests meets a defined threshold.

0015In any of the previous embodiments, the data processing tasks identified in the set of task identifiers can be executed in parallel.

0016In any of the previous embodiments, selecting the job having the longest job completion time can be based on the shortest task completion times at data processing locations having available resources; and selecting the data processing task having the longest task completion time can be based on the shortest task completion times for the unscheduled data processing tasks for the selected job at data processing locations having available resources.

0017In accordance with another aspect of the present disclosure there is provided a system comprising: at least one processor for scheduling jobs in a distributed computing environment. The at least one processor is configured for: obtaining a set of task identifiers, each task identifier identifying a corresponding data processing task included in one of a plurality of jobs to be scheduled for execution at one of a plurality of data processing locations; from the jobs having unscheduled data processing tasks, selecting a job having a longest job completion time based on a shortest task completion time for the data processing tasks included in the selected job; selecting, from unscheduled data processing tasks for the selected job, a data processing task having a longest task completion time based on shortest task completion times for the unscheduled data processing tasks for the selected job; and scheduling the selected data processing task for execution at the data processing location of the plurality of data processing locations corresponding to the selected data processing task's shortest completion time and having available processing resources.

0018In some embodiments, the at least one processor is configured for: adjusting the completion times of all other unscheduled data processing tasks included in the selected job to have a completion time equal to the maximum of: the optimal completion time of the selected data processing task, and the completion time of the respective unscheduled data processing task.

0019In any of the previous embodiments, the at least one processor is configured for: until each data processing task identified in the set of task identifiers has been scheduled, repeating: updating the available processing resources to identify the resources to be consumed by the previously scheduled data processing task as unavailable; selecting the job of the set of jobs having unscheduled tasks and having the longest job completion time based on the shortest task completion time for the data processing tasks included in the selected job; selecting, from the unscheduled tasks for the selected job, the data processing task having the longest task completion time based on shortest task completion times for the unscheduled data processing tasks for the selected job; and scheduling the selected data processing task at the data processing location corresponding to the selected data processing task's shortest completion time.

0020In any of the previous embodiments, the at least one processor can be configured for: for each data processing task identified in the set of task identifiers: determining input data transfer times to transfer input data for the data processing task to each of the plurality of data processing locations having available processing resources; and for each of the plurality of data processing locations having available processing resources, determining a completion time for the data processing task based on the corresponding input data transfer time.

0021In any of the previous embodiments, determining the input data transfer times for a single data processing task identified in the set of task identifiers can include: identifying a size and location of each input data for the single data processing task; determining a communication bandwidth between each input data location and each of the data processing locations; and for each of the data processing locations: determining the transfer time for each of the input data to the data processing location based on the size of the input data and the communication input data location and the data processing location; and selecting, from the transfer times for each of the input data to the data processing location, a largest transfer time as the input data transfer time for the single data processing task at the data processing location.

0022In any of the previous embodiments, the at least one processor can be configured for: for each of the data processing tasks identified in the set of task identifiers, determining an execution time for the data processing task; and for each of the plurality of data processing locations having available processing resources, determining the completion time for the data processing task at the data processing location based on a corresponding input data transfer time and the corresponding execution time.

0023In any of the previous embodiments, determining the input data transfer times for a single data processing task identified in the set of task identifiers can include: identifying a size and location of each input data for the single data processing task; determining a communication bandwidth between each input data location and each of the data processing locations; and for each of the data processing locations: determining the transfer time for each of the input data to the data processing location based on the size of the input data and the communication input data location and the data processing location; and selecting, from the transfer times for each of the input data to the data processing location, a largest transfer time as the input data transfer time for the single data processing task at the data processing location.

0024In any of the previous embodiments, the at least one processor can be configured for: for each of the data processing tasks identified in the set of task identifiers, determining an execution time for the data processing task; and for each of the plurality of data processing locations having available processing resources, determining the completion time for the data processing task at the data processing location based on the a corresponding input data transfer time and the corresponding execution time.

0025In any of the previous embodiments, identifying the execution time for the data processing task can include: identifying a type of the data processing task; based on a database of execution data for types of data processing tasks, determining the execution time for the data processing task based on the identified type, and a size of the input data for the data processing task.

0026In any of the previous embodiments, the at least one processor can be configured for: populating one or more data structures representing completion time objectives and constraints based on the completion times for each of the data processing tasks, data processing location assignments parameters, and the available resources; wherein selecting the job having the longest optimal completion time includes solving a linear programming problem defined by the matrices.

0027In any of the previous embodiments, obtaining the set of task identifiers can include: receiving job execution requests until a scheduling trigger is detected; and upon detection of the scheduling trigger, identifying the set of data processing tasks to be scheduled from the job execution requests.

0028In any of the previous embodiments, scheduling trigger can be detected when a defined time period has elapsed, or when a number of received jobs execution requests meets a defined threshold.

0029In any of the previous embodiments, the data processing tasks identified in the set of task identifiers can be executed in parallel.

0030In any of the previous embodiments, selecting the job having the longest job completion time can be based on the shortest task completion times at data processing locations having available resources; and selecting the data processing task having the longest task completion time can be based on the shortest task completion times for the unscheduled data processing tasks for the selected job at data processing locations having available resources.

0031In accordance with another aspect of the present disclosure there is provided a non-transitory, computer-readable medium or media having stored thereon computer-readable instructions. The instructions, which when executed by at least one processor, configure the at least one processor for: obtaining a set of task identifiers, each task identifier identifying a corresponding data processing task included in one of a plurality of jobs to be scheduled for execution at one of a plurality of data processing locations; from the jobs having unassigned data processing tasks, selecting a job having a longest job completion time based on a shortest task completion time for the data processing tasks included in the selected job; selecting, from unscheduled data processing tasks for the selected job, a data processing task having a longest task completion time based on shortest task completion times for the unscheduled data processing tasks for the selected job; and scheduling the selected data processing task for execution at the data processing location of the plurality of data processing locations corresponding to the selected data processing task's shortest completion time and having available processing resources.

0032In some embodiments, the computer-readable instructions can configure the at least one processor for: adjusting the completion times of all other unscheduled data processing tasks included in the selected job to have a completion time equal to the maximum of: the optimal completion time of the selected data processing task, and the completion time of the respective unscheduled data processing task.

0033In any of the previous embodiments, the computer-readable instructions can configure the at least one processor for: until each data processing task identified in the set of task identifiers has been scheduled, repeating: updating the available processing resources to identify the resources to be consumed by the previously scheduled data processing task as unavailable; selecting the job of the set of jobs having unscheduled tasks and having the longest job completion time based on the shortest task completion time for the data processing tasks included in the selected job; selecting, from the unscheduled tasks for the selected job, the data processing task having the longest task completion time based on shortest task completion times for the unscheduled data processing tasks for the selected job; and scheduling the selected data processing task at the data processing location corresponding to the selected data processing task's shortest completion time.

0034In any of the previous embodiments, the computer-readable instructions can configure the at least one processor for: for each data processing task identified in the set of task identifiers: determining input data transfer times to transfer input data for the data processing task to each of the plurality of data processing locations having available processing resources; and for each of the plurality of data processing locations having available processing resources, determining a completion time for the data processing task based on the corresponding input data transfer time.

0035In any of the previous embodiments, determining the input data transfer times for a single data processing task identified in the set of task identifiers can include: identifying a size and location of each input data for the single data processing task; determining a communication bandwidth between each input data location and each of the data processing locations; and for each of the data processing locations: determining the transfer time for each of the input data to the data processing location based on the size of the input data and the communication input data location and the data processing location; and selecting, from the transfer times for each of the input data to the data processing location, a largest transfer time as the input data transfer time for the single data processing task at the data processing location.

0036In any of the previous embodiments, the computer-readable instructions can configure the at least one processor for: for each of the data processing tasks identified in the set of task identifiers, determining an execution time for the data processing task; and for each of the plurality of data processing locations having available processing resources, determining the completion time for the data processing task at the data processing location based on a corresponding input data transfer time and the corresponding execution time.

0037In any of the previous embodiments, determining the input data transfer times for a single data processing task identified in the set of task identifiers can include: identifying a size and location of each input data for the single data processing task; determining a communication bandwidth between each input data location and each of the data processing locations; and for each of the data processing locations: determining the transfer time for each of the input data to the data processing location based on the size of the input data and the communication input data location and the data processing location; and selecting, from the transfer times for each of the input data to the data processing location, a largest transfer time as the input data transfer time for the single data processing task at the data processing location.

0038In any of the previous embodiments, the computer-readable instructions can configure the at least one processor for: for each of the data processing tasks identified in the set of task identifiers, determining an execution time for the data processing task; and for each of the plurality of data processing locations having available processing resources, determining the completion time for the data processing task at the data processing location based on the a corresponding input data transfer time and the corresponding execution time.

0039In any of the previous embodiments, identifying the execution time for the data processing task can include: identifying a type of the data processing task; based on a database of execution data for types of data processing tasks, determining the execution time for the data processing task based on the identified type, and a size of the input data for the data processing task.

0040In any of the previous embodiments, the computer-readable instructions can configure the at least one processor for: populating one or more data structures representing completion time objectives and constraints based on the completion times for each of the data processing tasks, data processing location assignments parameters, and the available resources; wherein selecting the job having the longest optimal completion time includes solving a linear programming problem defined by the matrices.

0041In any of the previous embodiments, obtaining the set of task identifiers can include: receiving job execution requests until a scheduling trigger is detected; and upon detection of the scheduling trigger, identifying the set of data processing tasks to be scheduled from the job execution requests.

0042In any of the previous embodiments, scheduling trigger can be detected when a defined time period has elapsed, or when a number of received jobs execution requests meets a defined threshold.

0043In any of the previous embodiments, the data processing tasks identified in the set of task identifiers can be executed in parallel.

0044In any of the previous embodiments, selecting the job having the longest job completion time can be based on the shortest task completion times at data processing locations having available resources; and selecting the data processing task having the longest task completion time can be based on the shortest task completion times for the unscheduled data processing tasks for the selected job at data processing locations having available resources.

0045Many further features and combinations thereof concerning the present improvements will appear to those skilled in the art following a reading of the instant disclosure.

DESCRIPTION OF THE FIGURES

0046<figref idref="DRAWINGS">FIG. 1</figref> is a diagram showing aspects of an example computing environment and task assignment scenario;

0047<figref idref="DRAWINGS">FIG. 2</figref> is a diagram showing an example task assignment for the computing environment of <figref idref="DRAWINGS">FIG. 1</figref>;

0048<figref idref="DRAWINGS">FIG. 3</figref> is a diagram showing another example task assignment for the computing environment of <figref idref="DRAWINGS">FIG. 1</figref>;

0049<figref idref="DRAWINGS">FIG. 4</figref> is a diagram showing aspects of an example computing environment;

0050<figref idref="DRAWINGS">FIG. 5</figref> is a schematic showing aspects of an example system and process for scheduling jobs;

0051<figref idref="DRAWINGS">FIG. 6</figref> is a flowchart showing aspects of an example method for scheduling jobs;

0052<figref idref="DRAWINGS">FIGS. 7 and 8</figref> are diagrams showing two different example task assignments for a computing environment having three data processing locations;

0053<figref idref="DRAWINGS">FIG. 9</figref> is a diagram showing an example geodistributed data processing environment;

0054<figref idref="DRAWINGS">FIGS. 10, 11 and 12</figref> include graphs showing example job completion time results.

0055These drawings depict aspects of example embodiments for illustrative purposes. Variations, alternative configurations, alternative components and modifications may be made to these example embodiments.

DETAILED DESCRIPTION

0056It is increasingly common for large volumes of data to be generated and processed in a distributed fashion at multiple datacenters spread across a network and/or geographically around the world.

0057A data analytic job can often be separated into tens to thousands of data processing tasks. Some tasks may be dependent on the results of previous tasks in the job. In some approaches, a job can be divided into a number of consecutive computation stages, each of which includes a number of data processing tasks which can be executed in parallel. To start a new computation stage, intermediate data from the preceding stage may need to be fetched from another location, which may initiate multiple network flows.

0058When input data is located across multiple datacenters, one approach is to gather all the data for a single job to be processed locally within a single datacenter. In some instances, transferring large amounts of data between datacenters may be slow and inefficient because bandwidth on inter-datacenter links may be limited.

0059In some approaches, better performance may be achieved if tasks in a job are distributed across datacenters to be located closer to the data being processed. Different task assignment architectures can results in different data flow patterns, and may ultimately affect job completion times and throughput.

0060In one task scheduling architecture, all of the tasks of an individual job may be considered to determine an optimal task assignment for that particular job. However, because multiple jobs are competing for limited resources, optimizing the performance for a single job may assign resources to the detriment of other jobs waiting to be executed.

0061In some instances, task scheduling architectures which consider tasks across multiple jobs may result in increased fairness in the distribution of resources, and may reduce execution times for concurrently executing jobs. In some instances, this may improve the overall throughput of the system.

0062<figref idref="DRAWINGS">FIG. 1</figref> illustrates an example distributed computing environment <b>100</b> including three datacenters <b>110</b> labelled DC<b>1</b>, DC<b>2</b>, and DC<b>3</b>. Each datacenter <b>110</b> includes data storage device(s) <b>120</b> which contain input data A<b>1</b>, A<b>2</b>, B<b>1</b>, or B<b>2</b> as labelled. Each datacenter <b>110</b> includes a number of processing slots <b>130</b> with each slot representing a set of available computing resource(s) which can accommodate a single task assignment. The arrows represent the available communication bandwidth between the datacenters <b>110</b> in MB/s.

0063In this example scenario, two jobs, Job A and Job B, have been received for scheduling. Job A includes two tasks tA<b>1</b> and tA<b>2</b>, each of which require input data A<b>1</b> and A<b>2</b>. Job B includes two tasks tB<b>1</b> and tB<b>2</b>, each of which require input data B<b>1</b> and B<b>2</b>. For Job A, both of its tasks, tA<b>1</b> and tA<b>2</b>, require 100 MB of data from input data A<b>1</b> stored in DC<b>1</b>, and 200 MB of data from A<b>2</b> located at DC<b>3</b>. For Job B, the amounts of data to be read by task tB<b>1</b> from data B<b>1</b> in DC<b>2</b> and B<b>2</b> in DC<b>3</b> are both 200 MB; and task tB<b>2</b> needs to read 200 MB of data from B<b>1</b> and 300 MB from B<b>2</b>.

0064<figref idref="DRAWINGS">FIG. 2</figref> shows an example task assignment for the example computing environment <b>100</b> and scenario illustrated in <figref idref="DRAWINGS">FIG. 1</figref>. This example assignment was scheduled by a scheduler configured to optimize task assignments on a job-by-job basis. The arrows illustrate the transfer times for each input data in seconds (input data size in MB/bandwidth in MB per second).

0065To optimize the assignment of Job A, task tA<b>2</b> is assigned to the only available computing slot in DC<b>3</b>, and tA<b>1</b> is assigned to DC<b>2</b>, which when considering only data transfer times results in a job completion time of max {100/80, 200/160, 100/150}=1.25 seconds. Then, the scheduler continues to optimize the assignment of Job B. DC<b>1</b> and DC<b>2</b> are selected to distribute tasks tB<b>1</b> and tB<b>2</b>, respectively, resulting in the completion time for Job B of max {200/80, 200/100, 200/160}=2.5 seconds. This task placement may not be optimal when considering the performance of the jobs jointly.

0066<figref idref="DRAWINGS">FIG. 3</figref> shows another example task assignment for the example computing environment <b>100</b> and scenario illustrated in <figref idref="DRAWINGS">FIG. 1</figref>. This example assignment was scheduled by a scheduler configured to consider task assignments across different jobs. In this assignment, task tB<b>2</b> is assigned to the computing slot in DC<b>3</b>, which avoids the transfer of 300 MB of data from dataset B<b>2</b>. Task tB<b>1</b> is assigned to DC<b>2</b> rather than DC<b>1</b>, which takes advantage the high bandwidth of the DC<b>3</b>-DC<b>2</b> link. In this assignment, the completion times of Job A and Job B are max {200/100, 100/80, 200/160}=2 seconds, and max {200/160, 200/120}=1.67 seconds, respectively. Compared to the independent assignment in <figref idref="DRAWINGS">FIG. 2</figref> where the worst completion time across both jobs is 2.5 seconds, the assignment in <figref idref="DRAWINGS">FIG. 3</figref> results in a worst completion time of 2 seconds.

0067Aspects of embodiments described herein may, in some instances, provide a task scheduler which considers task assignments across multiple jobs.

0068<figref idref="DRAWINGS">FIG. 4</figref> is a schematic showing aspects of an example distributed computing environment <b>400</b> to which aspects of the present disclosure may be applied. In some embodiments, the distributed computing environment <b>400</b> can include any combination of hardware and software components, and may be referred to as a system. For example, a distributed computing environment <b>400</b> may include a combination of computing devices, processors, networks, memories, storage devices, network devices, power sources, license servers (e.g., software license servers), swap space, and the like. In some embodiments, the computing environment <b>400</b> may be a heterogeneous or a homogenous environment, and may have different or similar hardware components running different or similar operating systems.

0069In some embodiments, the distributed computing environment <b>400</b> may be a computing system including one or more processors in a single device or split across any number of devices. These processors can be configured to manage the resources <b>450</b> and/or schedule tasks in the computing system. The computing system can be configured as a distributed resource management (DRM) system. In some embodiments, the DRM system is an infrastructure middleware which can run on top of a distributed environment. The distributed computing environment <b>400</b> can include different kinds of hardware and software. The distributed computing environment <b>400</b> includes a number of data processing locations <b>410</b>. A data processing location <b>410</b> can include one or more devices having computing resources <b>450</b>. In some embodiments, a data processing location <b>410</b> can be a single device physically or logically, such as an individual computer or server. In some embodiments, a data processing location <b>410</b> can include multiple devices such as a computing system, mainframe or datacenter.

0070The distributed computing environment <b>400</b> can include one or more resources <b>450</b> which can be utilized to perform data processing tasks. Resources <b>450</b> can include, but are not limited to, processors (e.g. central processing units, graphics processing units, processing units or cores, integrated circuits, field-programmable gate arrays, any other computational devices, and the like), memories, network resources (ports, bandwidth allocations, etc.), software licenses, data storage devices, and any other hardware or software component, or combination thereof which can be assigned to tasks/jobs in a distributed computing environment. These resources <b>450</b> can be at any number of data processing locations <b>410</b>.

0071Some resources <b>450</b> may be physically or logically associated with a single device, and other resources <b>450</b> may be shared resources which may be utilized by multiple devices in the distributed computing environment <b>400</b>. In some embodiments, resources <b>150</b> in the computing environment <b>400</b> can have or otherwise be associated with one or more attributes or characteristics. In some embodiments, attributes can include resource type or class, resource state/status, resource location, resource identifier/name, resource value, resource capacity, resource capabilities, or any other resource information that can be used as criteria for selecting or identifying a suitable resource.

0072The distributed computing environment <b>400</b> can include one or more data storage devices <b>420</b> located at data processing locations <b>410</b> or elsewhere in the environment <b>400</b>. The data storage devices <b>420</b> may store input, output and/or intermediate data for one or more tasks or jobs. Data may be in the form of datasets, databases, files, or any suitable format.

0073The data processing locations <b>410</b>, resources <b>450</b>, and data storage devices <b>420</b> may be connected by one or more data networks <b>430</b>. The data networks <b>430</b> can include direct communication links, physical networks, logical networks, wired networks and/or wireless networks. These networks may include any number of network devices, communication interfaces, or other components for data communication between elements of the distributing computing environment. In some embodiments, data networks <b>430</b> can include communication connections between different systems or devices, and/or interconnections within a device or system (e.g. within a datacenter, within a rack, within a backplane, etc.). In some embodiments, aspects of the networks <b>430</b> themselves can be resources. For example, physical or logical connections/ports/addresses, communication bandwidths, protocols, etc. In some embodiments, various elements of the data processing locations <b>410</b> or other elements of the environment include communication interfaces and/or other suitable hardware and/or software components for communicating data across the network(s) <b>430</b>.

0074In some embodiments, processors and/or other computer elements at the data processing locations <b>410</b> and/or elsewhere in the distributed computing environment can be configured to operate one or more distributed resource management (DRM) agents or other hardware and/or software elements for monitoring and reporting resource capabilities and/or availabilities, file sizes, network bandwidths, and/or any other information for the job/task schedulers described herein.

0075In some embodiments, processors and/or other computer elements in the system may be configured to operate at least portions of a DRM system including, for example, DRM agents, master daemons, data processing engines, job managers, directed acyclic graph schedulers, task schedulers, resources managers and/or any other DRM component or portion thereof. These DRM components may be software components, hardware components, or any combination thereof. In some embodiments, the DRM components may be implemented by machine-readable instructions being executed on one or more processors.

0076Aspects of the present disclosure may be applied to various DRM systems such as the IBM™ Platform Load Sharing Facility (LSF), Apache™ Hadoop™, YARN (Yet Another Resource Negotiator), Apache™ Spark™, PBS (Portable Batch Scheduler), OpenStack™, Huawei™ FusionSphere, FusionCube and FusionInsight.

0077<figref idref="DRAWINGS">FIG. 5</figref> illustrates a schematic showing aspects of an example system <b>500</b> and process <b>501</b> for scheduling jobs in a distributed computing environment.

0078Jobs can refer to any process, service, or any other computing operation or set of operations to be run on a computing system. For example, jobs may include batch jobs (e.g., high performance computing (HPC) batch jobs), Message Passing Interface (MPI) processes, serial and/or parallel batch tasks, real time analytics, elastic applications (e.g., MapReduce), long running services, virtual machines, containers, etc.

0079In some instances, jobs include a number of tasks, or smaller operations which are to be completed in order to complete the job. In some embodiments, some tasks may be dependent on other tasks because they may utilize the output of a previous task as an input. In some embodiments, some tasks may be independent of and may be executed in parallel with others.

0080In some embodiments, the system <b>500</b> includes a task scheduler <b>510</b>, a job manager <b>515</b>, and a resource manager <b>525</b>. In some embodiments, the functions performed by these DRM components can be executed on different processors, devices, and/or data processing locations. In some embodiments, some or all the functions performed by these DRM components can be executed on the same processors, devices and/or data processing locations as other DRM components. In some embodiments, there may be no separation between some or all aspects of the job manager, task scheduler and/or resource manager; and some or all of their functionalities may all be executed by a single data process.

0081The task scheduler <b>510</b> obtains a set of task identifiers identifying data processing tasks to be scheduled. In some embodiments, a task identifier may be a unique identifier assigned to a task by the DRM system. In some embodiments, the task identifier may be mapped or otherwise associated with additional task data such as a job identifier identifying a job to which the task belongs, a task function call, task parameters, inputs, outputs, task types, data dependencies and/or the like. In some embodiments, task identifiers may include or may be included in some or all of the additional task data. In some embodiments, this task data is obtained from the job manager or other element of the DRM system.

0082In some instances, the data processing tasks identified in the set of task identifiers can be from multiple jobs. For example, a set of data processing tasks identified by the set of task identifiers may include two tasks from Job A, and two tasks from Job B.

0083In some embodiments, the task scheduler <b>510</b> obtains task identifiers from a job manager <b>515</b> or other element in the DRM system. In some embodiments, a job manager <b>515</b> receives job execution requests and communicates some or all tasks from a single job to the task scheduler <b>510</b>. In some embodiments, the job manager <b>515</b> may only send, for scheduling, tasks which are ready for execution, for example, if they have no outstanding data dependencies.

0084In some embodiments, the job manager <b>515</b> may be or may include a directed acyclic graph (DAG) scheduler. In some embodiments, the job manager <b>515</b> separates a job's tasks into stages which can be provided to the task scheduler on a stage by stage basis. In some embodiments, all tasks obtained by the task scheduler for a particular iteration of scheduling can be executed in parallel.

0085In some embodiments, the task scheduler <b>510</b> can store the task identifiers and/or task data in a scheduling pool <b>520</b>. For example, the task identifiers and/or task data may be stored in a buffer, memory, storage device and the like in a queue or other suitable arrangement until a scheduling trigger is detected.

0086In some embodiments, a scheduling trigger may be detected when the number of tasks or jobs represented in the scheduling pool <b>520</b> meets a defined threshold (e.g. the number of tasks or jobs is greater than X).

0087In some embodiments, a scheduling trigger may be detected when a defined time period has elapsed. In some embodiments, a scheduling trigger may be detected when a defined number or percentage of resources in the system become available.

0088In some embodiments, a scheduling trigger may be detected when the scheduling of a previous set of tasks has been completed. In some embodiments, a scheduling trigger may be detected when the execution of a previous set of tasks has been completed.

0089In some embodiments, a scheduling trigger may be related to any suitable event, or any combination thereof.

0090Upon detection of a scheduling trigger, the task scheduler <b>510</b>, in some embodiments, may identify some or all of the tasks in the scheduling pool and/or their corresponding identifiers as a set of task identifiers corresponding to tasks to be scheduled.

0091In some embodiments, the task scheduler <b>510</b> receives resource availability data for resources in the DRM system. This data may include static and/or runtime/dynamic resource information. In some examples, the resource availability data may include resource types, resource names, device identifiers, network addresses, resource capacities or capabilities, consumed/available resource information, resource status, resource relationships, topologies, and/or any other information associated with one or more resources.

0092In some embodiments, resource availability data can include a number of available processing cores, a total number of processing cores, an amount of free RAM, an up/down status of a resource, a type of processor, a processor speed, an operating system, and the like.

0093In some embodiments, the DRM system may include a resource manager <b>525</b> or cluster manager which receives resource availability data from different agents or other DRM components in the system which monitor and report local resource availability data. In some embodiments, the resource availability information may be obtained by the task scheduler from a resource manager <b>525</b> and/or directly from DRM components (e.g. agents) in the system.

0094In some embodiments, the task scheduler <b>510</b> receives or otherwise obtains network and task data such as input data sizes, input data locations, network bandwidths, and the like. Some or all of this data may be received with the task data and/or the resource data.

0095The task scheduler performs task scheduling operations <b>530</b> on a set of tasks to be scheduled from the scheduling pool <b>520</b>. In some embodiments, the task scheduler <b>510</b> is configured to consider optimal job completion times as well as fairness in resource allocation between jobs. In some instances, a task scheduling/assignment problem can be an NP-hard integer optimization problem.

0096However, in some embodiments, the task scheduling processes <b>530</b> described herein may reduce the processor computations into a linear programming problem which, in some instances, may be computed efficiently.

0097In some embodiments, the task scheduling processes <b>530</b> assign the tasks to available resources, and the assigned tasks are scheduled for execution. In some embodiments, the scheduled task data <b>540</b> including task identifiers and their assigned data processing locations/resources are stored in a buffer or other data storage device until the instructions or other signals for executing the tasks are transmitted. In some embodiments, the signals for executing the scheduled tasks are transmitted after all tasks in the scheduling pool have been scheduled.

0098In some embodiments, instructions or other signals for executing the tasks are transmitted as soon as a task is scheduled.

0099<figref idref="DRAWINGS">FIG. 6</figref> is a flowchart showing aspects of an example method <b>600</b> for scheduling jobs in a distributed computing environment.

0100At <b>610</b>, as described herein or otherwise, one or more processors in the DRM system obtain a set of task identifiers. In some embodiments, each task identifier identifies a corresponding data processing task to be scheduled for execution. The set of task identifiers may define a set of data processing tasks which are from at least two different jobs. As described herein or otherwise, data processing jobs can include one or more data processing tasks.

0101In some embodiments, the set of task identifiers may be grouped in a data structure such as an array or database table. In other examples, the set of task identifiers may not necessarily be stored in a separate data structure but may be referenced by links or otherwise logically grouped together in a set.

0102At <b>620</b>, the processor(s) determines transfer times to transfer input data to each data processing location for each of the data processing tasks. In some instances, the processors may generate, for each data processing task, an array, vector or other arrangement of input data transfer times. In some embodiments, the arrangement can include transfer times for each input data from the input data's original location to each data processing location having available resources.

0103In some embodiments, the processor(s) identify data processing locations having available resources as described herein or otherwise.

0104In some instances, input data for a task may be located at multiple locations in the distributed computing environment. In some embodiments, the processors determine the input data transfer times to transfer each of the respective input data to each data processing location.

0105In some embodiments, determining an input data transfer time includes identifying a size and location of each input data for a task. In some embodiments, this data may be obtained from a job manager or other DRM component. In some embodiments, input data size and location information can be included in job data and/or task data which may be received when tasks are added to the scheduling pool or at any other time.

0106In some embodiments, the processor(s) may obtain input data size and location information from one or more file system processes.

0107In some embodiments, determining an input data transfer time includes determining a communication bandwidth between each input data location and each of the data processing locations. In some embodiments, the processors determine communication bandwidths through measurements using the iperf2 or other similar utility. In some embodiments, the processors may determine communication bandwidths by monitoring data transfer times of previous data transmissions between the respective input data locations and data processing locations.

0108In some embodiments, the input data transfer time to transfer all input data for a particular task to a particular data processing location is determined to be the largest transfer time selected from the corresponding transfer times for each input data.

0109In some embodiments, the processors determine the input data transfer times based on the following equation:

0000<maths id="MATH-US-00001" num="00001"><math overflow="scroll"><mrow><msubsup><mi>c</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo>=</mo><mrow><mo>{</mo><mtable><mtr><mtd><mrow><mn>0</mn><mo>,</mo></mrow></mtd><mtd><mrow><mrow><mi>when</mi><mo></mo><mstyle><mspace width="0.8em" height="0.8ex" /></mstyle><mo></mo><msubsup><mi>S</mi><mi>i</mi><mi>k</mi></msubsup></mrow><mo>=</mo><mrow><mo>{</mo><mi>j</mi><mo>}</mo></mrow></mrow></mtd></mtr><mtr><mtd><mrow><mrow><munder><mi>max</mi><mrow><mrow><mi>s</mi><mo>∈</mo><msubsup><mi>S</mi><mi>i</mi><mi>k</mi></msubsup></mrow><mo>,</mo><mrow><mi>s</mi><mo>≠</mo><mi>j</mi></mrow></mrow></munder><mo></mo><mrow><mo>(</mo><mrow><msubsup><mi>d</mi><mi>i</mi><mrow><mi>k</mi><mo>,</mo><mi>s</mi></mrow></msubsup><mo></mo><mstyle><mtext>/</mtext></mstyle><mo></mo><msub><mi>b</mi><mi>sj</mi></msub></mrow><mo>)</mo></mrow></mrow><mo>,</mo></mrow></mtd><mtd><mi>otherwise</mi></mtd></mtr></mtable></mrow></mrow></math></maths>

0110Where c_i,j^kis the input data transfer time for task i of job k at data processing location j; s is a input data source location from the set S of all input data source location for the particular task i of the particular job k; d is the amount of data to be transferred; and b is the communication bandwidth from location s to location j. In some embodiments, the data transfer time is zero when input data is already at the data processing location.

0111At <b>630</b>, the processor(s) determine a task completion time for each data processing task for each data processing location. In some embodiments, the task completion time is determined based on the corresponding input data transfer times.

0112In some embodiments, the task completion time can be based on the time required for each task to be executed. In some embodiments, the task completion time is based on a sum of the data transfer time and the execution time for a particular data processing location.

0113In some embodiments, the processor(s) determine an execution time for each data processing task. In some embodiments, the execution time can be determined based on the type of task and/or the commands/processes invoked by the task. For example, in some embodiments, the type of task may be matched with a database of execution data for types of data processing tasks. In some embodiments, the database may include historical or defined task execution times, common numbers of computations performed by a task (e.g. number of floating point operations, number of floating point operations per input data unit, etc.) and the like. In some embodiments, the processor(s) determine the execution time for a task based on the size of the input data for the task.

0114In some embodiments, the processors determine an execution time for a task at each data processing location. In some embodiments, the execution time for a task is based on the capabilities of the available resources at the data processing locations (e.g. processor speed, number of processing cores, floating point operations per second, etc.).

0115Beginning at <b>640</b>, the processors iteratively assign tasks to available resources until all tasks identified in the set of task identifiers have been assigned. At <b>640</b>, the processors select a job having unscheduled tasks and having the longest optimal job completion time based on the task completion times and the data processing locations having available resources. In some embodiments, the processors select the job with unscheduled tasks having the longest job completion time based on the shortest task completion times for the job.

0116In some embodiments, the completion time τ_kof job k can be identified as the largest job completion time among all of its tasks and across all available data processing locations D:

0000<maths id="MATH-US-00002" num="00002"><math overflow="scroll"><mrow><msub><mi>τ</mi><mi>k</mi></msub><mo>=</mo><mrow><munder><mi>max</mi><mrow><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow><mo>,</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></mrow></munder><mo></mo><mrow><msubsup><mi>x</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo></mo><mrow><mo>(</mo><mrow><msubsup><mi>c</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo>+</mo><msubsup><mi>e</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup></mrow><mo>)</mo></mrow></mrow></mrow></mrow></math></maths>

0117Where x_i,j^kis a binary variable indicating whether the i-th task of job k is assigned to data location j; and where e_i,j^kis the execution time for the i-th task of job k at data location j. In embodiments where the execution time is not considered, the e_i,j^kterm can be omitted.

0118In some embodiments, the processors select the job corresponding to the largest value from the set of shortest completion times for all the jobs {τ₁, τ₂, . . . , τ_K}. In some embodiments, to select the job corresponding to the largest value from the set of shortest completion times for the jobs, the processors calculate the shortest completion times for each job based on the task completion times and the data processing locations having available resources, and select the job which corresponds to the largest optimal completion time.

0119At <b>650</b>, the processors select the unscheduled task(s) having the longest optimal completion time from the tasks of the selected job. In some embodiments, when more than one task has the same optimal completion time or must be assigned in a particular location in order to achieve the optimal completion time, the processors select all of these tasks. In some embodiments, the processors select the task having the longest task completion time based on the shortest task completion times for the unscheduled tasks included in the selected job.

0120At <b>660</b>, the processors schedule the selected data processing task(s) for execution at the data processing location corresponding to their shortest completion times. In some embodiments, scheduling tasks includes assigning the task to the data processing location. In some embodiments, this includes updating a set of x values, such that the x value corresponding to the selected task at the corresponding data processing location is set to 1, true, or a functionally equivalent value; and the x values corresponding to the selected task at other data processing locations are set to 0, false or a functionally equivalent value.

0121At <b>670</b>, the processors update the available processing resources to identify the resources to be consumed by the scheduled task as unavailable. In some embodiments, the processors may represent available resources at each data processing location as an array of values or other suitable dataset. In some embodiments, the values may be binary to represent a slot or position which may or may not accommodate a task. In some embodiments, the values may represent available or unavailable amounts of a resource (e.g. available memory). In some such embodiments, these values are updated to reflect the resources that will be consumed by the assigned task.

0122While each task has not yet been scheduled <b>680</b>, the processors repeat the process starting at <b>640</b>.

0123At <b>690</b>, when each task has been scheduled, the processors generate and/or transmit instructions or other signals to execute the scheduled tasks. In some embodiments, the processors provide the scheduled task's data location assignment to another DRM component for execution. In other embodiments, the processors may generate instructions and/or datasets to instruct the data processing locations to execute their assigned tasks.

0124In some embodiments, rather than waiting for all tasks to be scheduled, the processors can initiate the execution of tasks as they are scheduled.

0125Because the slowest task for a particular job will define the optimal completion time, it does not matter how much faster other tasks in the job are completed. Therefore, in some embodiments, in order to prevent unnecessary optimization of faster tasks in a job, after determining a task assignment, the processors are can update objective parameters for all unassigned tasks belonging to the same job. This is illustrated in <figref idref="DRAWINGS">FIGS. 7 and 8</figref>.

0126In some embodiments, after determining the resource/location assignment for the selected task, the processor(s) adjust the completion times of all other unscheduled data processing tasks included in the selected job. In some embodiments, the completion times of all other unscheduled data processing tasks in the selected job are set to have an optimal completion time which is not faster than the completion time of the selected data processing task. In some embodiments, if the optimal completion time of the unscheduled data processing task in the selected job is faster than the completion time of the selected data processing task, the completion time of the unscheduled data processing task is set to the completion time of the selected data processing task.

0127In some embodiments, the completion time of an unscheduled data processing task in the selected job is set to the maximum between its own completion time and the completion time of the selected data processing task.

0128<figref idref="DRAWINGS">FIGS. 7 and 8</figref> show two example task assignments for two jobs each having two tasks in a computing environment <b>700</b> with three data processing locations: datacenter DC<b>1</b>, datacenter DC<b>2</b> and datacenter DC<b>3</b>. DC<b>1</b> can accommodate two tasks based on its available resource data, and DC<b>2</b> and DC<b>3</b> can accommodate one task each.

0129In this example, after the calculation in the first round, task tA<b>1</b> is assigned to resources in DC<b>2</b>, which has a completion time of four seconds at this location. This assignment is selected because assigning task tA<b>1</b> to DC<b>2</b> achieves the worst optimal completion time amongst the jobs. In <figref idref="DRAWINGS">FIG. 7</figref>, the completion times for task tA<b>2</b> are not updated, so in the second round, the process would maximize the next worst completion time for task tA<b>2</b> which is in the same job. This results in the completion times (in seconds) for the tasks as tA<b>1</b>=4, tA<b>2</b>=3, tB<b>1</b>=2, and tB<b>2</b>=2.5. Job A has a completion time of 4 seconds, and Job B has a completion time of 2.5 seconds.

0130In contrast, in <figref idref="DRAWINGS">FIG. 8</figref>, after the processors determine tA<b>1</b> is to be assigned to DC<b>2</b>, the completion time for tA<b>2</b> is updated to be 4 seconds. This results in the task assignment shown in <figref idref="DRAWINGS">FIG. 8</figref> and the completion times of tA<b>1</b>=4, tA<b>2</b>=3.5, tB<b>1</b>=2, and tB<b>2</b>=2. Job A still has a completion time of 4 seconds, and Job B has a completion time of 2 seconds. In some embodiments, the selection in a subsequent round is solved over a decreased set of variables with updated constraints and objectives such that, in some instances, a next worst job completion time can be optimized without impacting the worst job performance in the current round.

0131In some instances, task assignment problems can be defined as integer optimization problems which are generally considered to be NP-hard problems. In some situations, based on a number of transformations, the problem can be solved as an equivalent problem with a separable convex objective. In some examples, this may allow the task assignment to be performed more effectively by the DRM system.

0132Based on the terminology above, in some situations, determining the optimal worst performance across jobs may be represented by a lexicographical minimization problem f:

0000<maths id="MATH-US-00003" num="00003"><math overflow="scroll"><mrow><mrow><munder><mi>lexmin</mi><mi>x</mi></munder><mo></mo><mstyle><mspace width="0.3em" height="0.3ex" /></mstyle><mo></mo><mi>f</mi></mrow><mo>=</mo><mrow><mo>(</mo><mrow><msub><mi>τ</mi><mn>1</mn></msub><mo>,</mo><msub><mi>τ</mi><mn>2</mn></msub><mo>,</mo><mi>…</mi><mo></mo><mstyle><mspace width="0.3em" height="0.3ex" /></mstyle><mo>,</mo><msub><mi>τ</mi><mi>K</mi></msub></mrow><mo>)</mo></mrow></mrow></math></maths><maths id="MATH-US-00003-2" num="00003.2"><math overflow="scroll"><mrow><mrow><mrow><mi>s</mi><mo>.</mo><mi>t</mi><mo>.</mo><mstyle><mspace width="0.6em" height="0.6ex" /></mstyle><mo></mo><msub><mi>τ</mi><mi>k</mi></msub></mrow><mo>=</mo><mrow><munder><mi>max</mi><mrow><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow><mo>,</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></mrow></munder><mo></mo><mrow><msubsup><mi>x</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo></mo><mrow><mo>(</mo><mrow><msubsup><mi>c</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo>+</mo><msubsup><mi>e</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup></mrow><mo>)</mo></mrow></mrow></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></mrow></mrow></math></maths>

0000With the constraints that the total number of tasks to be assigned to a data processing location j does not exceed its resource capacity a_j:

0000<maths id="MATH-US-00004" num="00004"><math overflow="scroll"><mrow><mrow><mrow><munder><mo>∑</mo><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></munder><mo></mo><mrow><munder><mo>∑</mo><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow></munder><mo></mo><msubsup><mi>x</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup></mrow></mrow><mo>≤</mo><msub><mi>a</mi><mi>j</mi></msub></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></mrow></mrow></math></maths>

0000Each task is only assigned to a single data processing location:

0000<maths id="MATH-US-00005" num="00005"><math overflow="scroll"><mrow><mrow><mrow><munder><mo>∑</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></munder><mo></mo><msubsup><mi>x</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup></mrow><mo>=</mo><mn>1</mn></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></mrow></mrow></math></maths>

0000Where each task assignment is represented by a binary variable x:

0000<br /><i>x</i>_i,j^kϵ{0,1}∀<i>iϵτ</i>_k<i>,∀jϵD,∀kϵK. </i>

0133The objective of this problem is a vector f with K elements, each representing the optimal completion time for a particular job.

0134Based on the above equations and constraints, a potentially NP-hard, integer programming problem can be defined as:

0000<maths id="MATH-US-00006" num="00006"><math overflow="scroll"><mrow><munder><mi>min</mi><mi>x</mi></munder><mo></mo><mrow><munder><mi>max</mi><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></munder><mo></mo><mrow><mo>(</mo><mrow><munder><mi>max</mi><mrow><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow><mo>,</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></mrow></munder><mo></mo><mrow><msubsup><mi>x</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo></mo><mrow><mo>(</mo><mrow><msubsup><mi>c</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo>+</mo><msubsup><mi>e</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup></mrow><mo>)</mo></mrow></mrow></mrow><mo>)</mo></mrow></mrow></mrow></math></maths>

0135In some embodiments, to impart fairness across jobs, the job having the slowest optimal completion time can be assigned first, and in subsequent iterations, the job having the next slowest optimal completion time can be scheduled.

0136An optimal solution for the job having the slowest optimal completion time may be obtained by solving the following linear programming problem:

0000<maths id="MATH-US-00007" num="00007"><math overflow="scroll"><mrow><munder><mi>min</mi><mrow><mi>x</mi><mo>,</mo><mi>λ</mi></mrow></munder><mo></mo><mrow><munder><mo>∑</mo><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></munder><mo></mo><mrow><munder><mo>∑</mo><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow></munder><mo></mo><mrow><munder><mo>∑</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></munder><mo></mo><mrow><mo>(</mo><mrow><msubsup><mi>λ</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mrow><mi>k</mi><mo>,</mo><mn>0</mn></mrow></msubsup><mo>+</mo><mrow><msup><mi>M</mi><mrow><msubsup><mi>c</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo>+</mo><msubsup><mi>e</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup></mrow></msup><mo></mo><msubsup><mi>λ</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mrow><mi>k</mi><mo>,</mo><mn>1</mn></mrow></msubsup></mrow></mrow><mo>)</mo></mrow></mrow></mrow></mrow></mrow></math></maths><maths id="MATH-US-00007-2" num="00007.2"><math overflow="scroll"><mrow><mrow><mrow><mi>s</mi><mo>.</mo><mi>t</mi><mo>.</mo><mstyle><mspace width="0.6em" height="0.6ex" /></mstyle><mo></mo><msubsup><mi>x</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup></mrow><mo>=</mo><msubsup><mi>λ</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mrow><mi>k</mi><mo>,</mo><mn>1</mn></mrow></msubsup></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></mrow></mrow></math></maths><maths id="MATH-US-00007-3" num="00007.3"><math overflow="scroll"><mrow><mrow><mrow><msubsup><mi>λ</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mrow><mi>k</mi><mo>,</mo><mn>0</mn></mrow></msubsup><mo>+</mo><msubsup><mi>λ</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mrow><mi>k</mi><mo>,</mo><mn>1</mn></mrow></msubsup></mrow><mo>=</mo><mn>1</mn></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></mrow></mrow></math></maths><maths id="MATH-US-00007-4" num="00007.4"><math overflow="scroll"><mrow><msubsup><mi>λ</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mrow><mi>k</mi><mo>,</mo><mn>0</mn></mrow></msubsup><mo>,</mo><msubsup><mi>λ</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mrow><mi>k</mi><mo>,</mo><mn>1</mn></mrow></msubsup><mo>,</mo><mrow><msubsup><mi>x</mi><mrow><mi>i</mi><mo>,</mo><mi>j</mi></mrow><mi>k</mi></msubsup><mo>∈</mo><msup><mi>ℝ</mi><mo>+</mo></msup></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>k</mi><mo>∈</mo><mi>K</mi></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>i</mi><mo>∈</mo><msub><mi>τ</mi><mi>k</mi></msub></mrow></mrow><mo>,</mo><mrow><mo>∀</mo><mrow><mi>j</mi><mo>∈</mo><mi>D</mi></mrow></mrow></mrow></math></maths>

0000Where M is the dimension of <br /> g=(ϕ(x_1,1¹), . . . , ϕ(x_i,j^k), . . . ϕ(x_n_{<sub2>K</sub2>}_,J^K), <br /> and ϕ(x_i,j^k) is the convex objective function.

0000<br />ϕ(<i>x</i>_i,j^k)=<i>x</i>_i,j^k(<i>c</i>_i,j^k<i>+e</i>_i,j^k), ∀<i>iϵτ</i>_k<i>,∀iϵD,∀kϵK. </i>

0137In some embodiments, the processors are configured to populate one or more data structures representing parameters for solving some or all aspects of the above-noted linear programming model. In some embodiments, the data structures by matrices, vectors, or any other suitable data format.

0138In some embodiments, the data structures encapsulate task assignment parameters (e.g. x), available resources (e.g. a), input data transfer times (e.g. c), execution times (e.g. e), job completion times and/or task completion times.

0139In some embodiments, the processors executed one or more linear programming processes on the populated data structures to identify optimal completion times and/or task assignments. In some embodiments, the linear programming processes may be implemented with MOSEK™ or other linear programming solver packages.

0140After solving the linear programming minimization sub-problem above, the processors continue to solve the lexicographical minimization problem f by iteratively minimizing the next worst completion time.

0141In some embodiments, the processors populate and manipulate data structures representing data processing tasks, jobs, data processing location assignment parameters (e.g. x), available resources, and/or completion times. In some embodiments, the data structures in conjunction with the linear programming solver encapsulate the completion time objectives and constraints described or similarly represented herein.

0142After solving the initial subproblem, it is known that the optimal worst job completion time is achieved by job k*, whose slowest task i* is assigned to processing location j*. In some embodiments, the processor(s) assign the slowest task and update the data processing location assignment parameters to remove the assignment parameter x_i*,j*^k*from the data structures for the next iteration. Also, because task i* is to be assigned to data location j*, the processor(s) update the data structures to set all other assignment parameters to 0 or false (x_i*,j*^k*=0, ∀j≠j*). The corresponding available resource parameters are also updated in the data structures.

0143In some embodiments, in order to obtain an assignment which can prevent an earlier selected job from receiving task assignments which could be unnecessarily detrimental to other jobs (as illustrated in the examples in <figref idref="DRAWINGS">FIGS. 7 and 8</figref>), the processors are configured to update the data structures to change the completion times. In some embodiments, after the slowest task i* of job k* has been assigned, for the tasks in k* which have not yet been assigned (x_i,j^k*, ∀i≠i*, ∀j≠j*), the processors update their completion times as x_i,j^k*(c_i*,j*^k*+e_i*,j*^k*).

0144<figref idref="DRAWINGS">FIG. 9</figref> shows an example data processing environment <b>900</b> having six geodistributed data processing locations (datacenters: Vir, Ore, Ire, Sin, Sy, and SP) with the arrows indicating the available bandwidth between each datacenter. Each datacenter has resources for two task assignments each, and the input data for jobs A, B, C and D are illustrated beside the datacenter at which they are location.

0145In an example test scenario, 4 jobs A, B, C and D are to be scheduled, each having three tasks each. The example data processing environment obtained the worst and second worst job completion times illustrated in <figref idref="DRAWINGS">FIG. 10</figref>. The upper graph <b>1000</b> shows, for ten separate runs, the worst job completion times of the four jobs. The dotted line (Baseline) represents the worst job completion times when the system is implemented using a default scheduling process in Spark™. In contrast, the solid line (Fair) represents the corresponding worst job completion times when the system is implemented using a scheduling process as described herein.

0146The lower graph <b>1010</b> similarly shows the job completion times for the second slowest job. In every run, the slowest and the second slowest jobs both completed faster when the “Fair” system than when using the “Baseline” system.

0147<figref idref="DRAWINGS">FIG. 11</figref> similarly shows the results for the worst job completion time <b>1100</b>, and the second worst job completion time <b>1110</b> for a scenario when three jobs are received for execution on the example data processing environment <b>900</b>.

0148<figref idref="DRAWINGS">FIG. 12</figref> similarly shows the results for the worst job completion time <b>1200</b>, and the second worst job completion time <b>1210</b> for a scenario when five jobs are received for execution on the example data processing environment <b>900</b>.

0149Embodiments disclosed herein may be implemented using hardware, software or some combination thereof. Based on such understandings, the technical solution may be embodied in the form of a software product. The software product may be stored in a non-volatile or non-transitory storage medium, which can be, for example, a compact disk read-only memory (CD-ROM), USB flash disk, a removable hard disk, flash memory, hard drive, or the like. The software product includes a number of instructions that enable a computing device (computer, server, mainframe, or network device) to execute the methods provided herein.

0150Program code may be applied to input data to perform the functions described herein and to generate output information. The output information is applied to one or more output devices. In some embodiments, the communication interface may be a network communication interface. In embodiments in which elements are combined, the communication interface may be a software communication interface, such as those for inter-process communication. In still other embodiments, there may be a combination of communication interfaces implemented as hardware, software, and/or combination thereof.

0151Each computer program may be stored on a storage media or a device (e.g., ROM, magnetic disk, optical disc), readable by a general or special purpose programmable computer, for configuring and operating the computer when the storage media or device is read by the computer to perform the procedures described herein. Embodiments of the system may also be considered to be implemented as a non-transitory computer-readable storage medium, configured with a computer program, where the storage medium so configured causes a computer to operate in a specific and predefined manner to perform the functions described herein.

0152Furthermore, the systems and methods of the described embodiments are capable of being distributed in a computer program product including a physical, non-transitory computer readable medium that bears computer usable instructions for one or more processors. The medium may be provided in various forms, including one or more diskettes, compact disks, tapes, chips, magnetic and electronic storage media, volatile memory, non-volatile memory and the like. Non-transitory computer-readable media may include all computer-readable media, with the exception being a transitory, propagating signal. The term non-transitory is not intended to exclude computer readable media such as primary memory, volatile memory, RAM and so on, where the data stored thereon may only be temporarily stored. The computer useable instructions may also be in various forms, including compiled and non-compiled code.

0153The present disclosure may make numerous references to servers, services, interfaces, portals, platforms, or other systems formed from hardware devices. It should be appreciated that the use of such terms is deemed to represent one or more devices having at least one processor configured to execute software instructions stored on a computer readable tangible, non-transitory medium. One should further appreciate the disclosed computer-based algorithms, processes, methods, or other types of instruction sets can be embodied as a computer program product comprising a non-transitory, tangible computer readable media storing the instructions that cause a processor to execute the disclosed steps.

0154Various example embodiments are described herein. Although each embodiment represents a single combination of inventive elements, the inventive subject matter is considered to include all possible combinations of the disclosed elements. Thus, if one embodiment comprises elements A, B, and C, and a second embodiment comprises elements B and D, then the inventive subject matter is also considered to include other remaining combinations of A, B, C, or D, even if not explicitly disclosed.

0155The embodiments described herein are implemented by physical computer hardware embodiments. The embodiments described herein provide useful physical machines and particularly configured computer hardware arrangements of computing devices, servers, processors, memory, networks, for example. The embodiments described herein, for example, are directed to computer apparatuses, and methods implemented by computers through the processing and transformation of electronic data signals.

0156The embodiments described herein may involve computing devices, servers, receivers, transmitters, processors, memory(ies), displays, networks particularly configured to implement various acts. The embodiments described herein are directed to electronic machines adapted for processing and transforming electromagnetic signals which represent various types of information. The embodiments described herein pervasively and integrally relate to machines and their uses; the embodiments described herein have no meaning or practical applicability outside their use with computer hardware, machines, a various hardware components.

0157Substituting the computing devices, servers, receivers, transmitters, processors, memory, display, networks particularly configured to implement various acts for non-physical hardware, using mental steps for example, may substantially affect the way the embodiments work.

0158Such hardware limitations are clearly essential elements of the embodiments described herein, and they cannot be omitted or substituted for mental means without having a material effect on the operation and structure of the embodiments described herein. The hardware is essential to the embodiments described herein and is not merely used to perform steps expeditiously and in an efficient manner.

0159Although the present invention and its advantages have been described in detail, it should be understood that various changes, substitutions and alterations can be made herein without departing from the invention as defined by the appended claims.

0160Moreover, the scope of the present application is not intended to be limited to the particular embodiments of the process, machine, manufacture, composition of matter, means, methods and steps described in the specification. As one of ordinary skill in the art will readily appreciate from the disclosure of the present invention, processes, machines, manufacture, compositions of matter, means, methods, or steps, presently existing or later to be developed, that perform substantially the same function or achieve substantially the same result as the corresponding embodiments described herein may be utilized according to the present invention. Accordingly, the appended claims are intended to include within their scope such processes, machines, manufacture, compositions of matter, means, methods, or steps.