Geo-scale analytics with bandwidth and regulatory constraints

Abstract

Various technologies described herein pertain to controlling geo-scale analytics with bandwidth and regulatory constraints. An analytical query (e.g., a recurrent analytical query, a non-recurrent analytical query, etc.) to be executed over distributed data in data partitions stored in a plurality of data centers can be received. Moreover, a query execution plan for the analytical query can be generated, where the query execution plan includes tasks. Further, replication strategies for the data partitions can be determined. A replication strategy for a particular data partition can specify one or more data centers to which the particular data partition is to be replicated if the particular data partition is to be replicated. The tasks of the query execution plan for the analytical query can further be scheduled to the data centers based on the replication strategies for the data partitions. The analytical query can be part of a workload of analytical queries.

What is claimed is: 
     
         1 . A computing system, comprising:
 at least one processor; and   computer-readable storage that stores computer-executable instructions that, when executed by the at least one processor, cause the at least one processor to perform acts comprising:
 receiving a query to be executed over distributed data in data partitions stored in geographically distributed data centers; 
 generating a query execution plan for the analytical query, wherein the query execution plan comprises a plurality of tasks; 
 determining a replication strategy for at least one of the data partitions, wherein the replication strategy specifies one or more of the geographically distributed data centers to which the at least one data partition is to be replicated; 
 causing a given data center of the geographically distributed data centers to execute at least one of the plurality of tasks to output at least one intermediate query result, wherein the at least one intermediate query result is based on a subquery delta, wherein the subquery delta comprises a difference between:
 a first intermediate query result for a subquery executed during a first time period by a given data center; 
 a second intermediate query result for the subquery executed during a second time period by the given data center, the second time period being subsequent to the first time period, wherein the second intermediate query result is computed based on the subquery delta and a stored copy of the first intermediate query result for the subquery; and 
 
 generating a result responsive to the analytical query based on the at least one intermediate query result received from the given data center of the geographically distributed data centers. 
   
     
     
         2 . The computing system of  claim 1 , the query planner component configured to generate the query execution plan for the analytical query and the workload optimization component configured to determine the replication strategies for the data partitions and schedule the tasks of the query execution plan for the analytical query to the data centers based on bandwidth costs for data transfers between the data centers. 
     
     
         3 . The computing system of  claim 2 , the bandwidth costs for the data transfers between the data centers comprise:
 replication bandwidth costs for data transfers between the data centers to replicate the data partitions; and   execution bandwidth costs for data transfers between the data centers to execute the tasks of the query execution plan for the analytical query.   
     
     
         4 . The computing system of  claim 1 , the components further comprising:
 a measurement component configured to:
 modify the query execution plan for the analytical query to generate a modified query execution plan, the query execution plan being modified to collect data transfer measurements for each of the tasks during execution of the modified query execution plan; and 
 acquire the data transfer measurements for each of the tasks during the execution of the modified query execution plan. 
   
     
     
         5 . The computing system of  claim 4 , the workload optimization component further configured to determine the replication strategies for the data partitions and schedule the tasks of the query execution plan for the analytical query to the data centers based on the data transfer measurements for each of the tasks. 
     
     
         6 . The computing system of  claim 1 , the workload optimization component further configured to determine the replication strategies for the data partitions and schedule the tasks of the query execution plan for the analytical query to the data centers based on fault-tolerance constraints for the data partitions and regulatory constraints for the data partitions, a fault-tolerance constraint for a corresponding data partition specifies a minimum number of replicas of the corresponding data partition, and a regulatory constraint for a given data partition specifies a data center to which the given data partition is restricted from being replicated. 
     
     
         7 . The computing system of  claim 1 , the workload optimization component further configured to solve an integer linear program to jointly determine the replication strategies for the data partitions and schedule the tasks of the query execution plan for the analytical query to the data centers. 
     
     
         8 . The computing system of  claim 1 , the workload optimization component further configured to determine the replication strategies for the data partitions and schedule the tasks of the query execution plan for the analytical query to the data centers using a heuristic algorithm. 
     
     
         9 . The computing system of  claim 1 , the query planner component further configured to periodically regenerate the query execution plan for the analytical query and the workload optimization component further configured to periodically determine the replication strategies for the data partitions and schedule the tasks of the query execution plan for the analytical query to the data centers. 
     
     
         10 . The computing system of  claim 1 , the components further comprising:
 a query control component configured to:
 receive the analytical query to be executed over the distributed data in the data partitions stored in the data centers; 
 control the data centers to execute the tasks of the query execution plan for the analytical query as scheduled and transfer intermediate query results between the data centers; 
 receive one or more of the intermediate query results from one or more of the data centers; and 
 generate a result responsive to the analytical query based on the one or more intermediate query results received from the one or more data centers. 
   
     
     
         11 . The computing system of  claim 10 , wherein the one or more intermediate query results received from the one or more data centers comprises a subquery delta, the subquery delta comprises a difference between:
 a second intermediate query result for a subquery executed during a second time period by a given data center; and   a first intermediate query result for the subquery executed during a first time period by the given data center, the second time period being subsequent to the first time period.   
     
     
         12 . The computing system of  claim 11 , the query control component further configured to compute the second intermediate query result for the subquery executed during the second time period based on:
 the subquery delta; and   a stored copy of the first intermediate query result for the subquery executed during the first time period retained by the computing system.   
     
     
         13 . The computing system of  claim 11 , the subquery being executed by the given data center as part of the analytical query to compute the second intermediate query result, and the subquery being executed by the given data center as part of a differing analytical query to compute the first intermediate query result. 
     
     
         14 . The computing system of  claim 1 , wherein the analytical query is comprised in a core workload of analytical queries executed over the distributed data in the data partitions stored in the data centers. 
     
     
         15 . The computing system of  claim 1  being a data center in the plurality of data centers. 
     
     
         16 . A method of controlling data transfer, comprising:
 receiving, at a data center, a first request for a first intermediate query result for a first subquery to be executed during a first time period, an analytical query comprises the first subquery, the analytical query to be executed over distributed data stored in a plurality of data centers, the plurality of data centers comprises the data center;   executing, at the data center, the first subquery to compute the first intermediate query result during the first time period;   responsive to the first request, transmitting the first intermediate query result for the first subquery from the data center;   storing the first intermediate query result for the first subquery in storage at the data center;   receiving, at the data center, a second request for a second intermediate query result for the first subquery to be executed during a second time period, the second time period being subsequent to the first time period;   executing, at the data center, the first subquery to compute the second intermediate query result during the second time period;   computing, at the data center, a subquery delta for the first subquery, the subquery delta for the first subquery being computed as a difference between:
 the second intermediate query result for the first subquery; and 
 the first intermediate query result for the first subquery as stored; and 
   responsive to the second request, transmitting the subquery delta for the first subquery from the data center.   
     
     
         17 . The method of  claim 16 , the first subquery being executed over data partitions stored in the data center, the data partitions comprise at least:
 a first data partition generated locally at the data center; and   a second data partition generated locally at a disparate data center, the second data partition being replicated to the data center.   
     
     
         18 . The method of  claim 16 , further comprising:
 receiving, at the data center, a third intermediate query result for a second subquery executed by a disparate data center during a third time period, the analytical query comprises the second subquery;   storing the third intermediate query result for the second subquery in the storage of the data center;   receiving, at the data center, a subquery delta for the second subquery, the subquery delta for the second subquery comprising a difference between:
 a fourth intermediate query result for the second subquery executed by the disparate data center during a fourth time period; and 
 the third intermediate query result for the second subquery executed by the disparate data center during the third time period, the fourth time period being subsequent to the third time period; and 
   computing, at the data center, the fourth intermediate query result for the second subquery based on:
 the subquery delta for the second subquery; and 
 the third intermediate query result for the second subquery stored at the data center. 
   
     
     
         19 . A computing system, comprising:
 at least one processor; and   computer-readable storage comprising components, the components being executable by the at least one processor, the components comprising:
 a query control component configured to receive an analytical query to be executed over distributed data in a plurality of data centers; 
 a query planner component configured to generate a query execution plan for the analytical query, the query execution plan comprising tasks; 
 a measurement component configured to generate a modified query execution plan for the analytical query based on the query execution plan, the query execution plan being modified to collect data transfer measurements for each of the tasks during execution of the modified query execution plan; 
 the query control component further configured to control the data centers to execute the tasks of the modified query execution plan and transfer intermediate query results between the data centers; and 
 the measurement component further configured to acquire the data transfer measurements for each of the tasks during the execution of the modified query execution plan. 
   
     
     
         20 . The computing system of  claim 19 , the components further comprising a workload optimization component configured to determine replication strategies for the distributed data and schedule the tasks of the analytical query to the data centers based on the data transfer measurements for each of the tasks.