Geo-scale analytics with bandwidth and regulatory constraints

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 an analytical 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;

scheduling the query execution plan and the replication strategy based on combined bandwidth costs for data transfers between the geographically distributed data centers for replicating the at least one data partition and executing the plurality of tasks of the query execution plan;

controlling the geographically distributed data centers to execute the plurality of tasks of the query execution plan for the analytical query as scheduled and transfer intermediate query results between the geographically distributed data centers, wherein at least one of the intermediate query results is based on a subquery delta, wherein 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, 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;

receiving one or more of the intermediate query results from one or more of the geographically distributed data centers; and

generating a result responsive to the analytical query based on the one or more intermediate query results received from the one or more geographically distributed data centers.

2. The computing system of claim 1 , wherein the plurality of tasks of the query execution plan for the analytical query are scheduled to the geographically distributed data centers based on the bandwidth costs for data transfers between the geographically distributed data centers.

3. The computing system of claim 1 , the combined bandwidth costs for the data transfers between the geographically distributed data centers comprise:

replication bandwidth costs for data transfers between the geographically distributed data centers to replicate the data partitions; and

execution bandwidth costs for data transfers between the geographically distributed data centers to execute the tasks of the query execution plan for the analytical query.

4. The computing system of claim 1 , the computer-readable storage further stores computer-executable instructions that, when executed by the at least one processor, cause the at least one processor to perform acts comprising:

modifying 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 plurality of tasks during execution of the modified query execution plan; and

acquiring the data transfer measurements for each of the plurality of tasks during the execution of the modified query execution plan.

5. The computing system of claim 4 , wherein the replication strategy for the at least one data partition is determined and the plurality of tasks of the query execution plan for the analytical query are scheduled to the geographically distributed data centers further based on the data transfer measurements for each of the plurality of tasks.

6. The computing system of claim 1 , wherein the replication strategy for the at least one data partition is determined and the plurality of tasks of the query execution plan for the analytical query are scheduled to the geographically distributed data centers further 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 , wherein an integer linear program is solved to jointly determine the replication strategy for the at least one data partition and to schedule the plurality of tasks of the query execution plan for the analytical query to the geographically distributed data centers.

8. The computing system of claim 1 , wherein the replication strategy for the at least one data partition is determined and the plurality of tasks of the query execution plan for the analytical query are scheduled to the geographically distributed data centers using a heuristic algorithm.

9. The computing system of claim 1 , wherein the query execution plan for the analytical query is periodically regenerated and the replication strategy for the at least one data partition is periodically determined.

10. The computing system of claim 1 , 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.

11. 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 geographically distributed data centers.

12. The computing system of claim 1 being a data center in the plurality of geographically distributed data centers.

13. A method of controlling data transfer, comprising:

receiving an analytical 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;

scheduling the query execution plan and the replication strategy based on combined bandwidth costs for data transfers between the geographically distributed data centers for replicating the at least one data partition and executing the plurality of tasks of the query execution plan;

controlling the geographically distributed data centers to execute the plurality of tasks of the query execution plan for the analytical query as scheduled and transfer intermediate query results between the geographically distributed data centers, wherein at least one of the intermediate query results is based on a subquery delta, wherein 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, 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, wherein the subquery is executed by the given data center as part of a different analytical query to compute the first intermediate query result;

receiving one or more of the intermediate query results from one or more of the geographically distributed data centers; and

generating a result responsive to the analytical query based on the one or more intermediate query results received from the one or more geographically distributed data centers.

14. The method of claim 13 , 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.

15. The method of claim 13 , 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.

16. The method of claim 13 , 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 geographically distributed data centers.

17. The method of claim 13 , wherein the replication strategy for the at least one data partition is determined and the plurality of tasks of the query execution plan for the analytical query are scheduled to the geographically distributed data centers using a heuristic algorithm.

18. The method of claim 13 , wherein the query execution plan for the analytical query is periodically regenerated and the replication strategy for the at least one data partition is periodically determined.

19. 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 operations, comprising:

receiving an analytical 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;

scheduling the query execution plan and the replication strategy based on combined bandwidth costs for data transfers between the geographically distributed data centers for replicating the at least one data partition and executing the plurality of tasks of the query execution plan;

controlling the geographically distributed data centers to execute the plurality of tasks of the query execution plan for the analytical query as scheduled and transfer intermediate query results between the geographically distributed data centers, wherein at least one of the intermediate query results is based on a subquery delta, wherein 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, 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, wherein the subquery is executed by the given data center as part of a different analytical query to compute the first intermediate query result;

receiving one or more of the intermediate query results from one or more of the geographically distributed data centers; and

generating a result responsive to the analytical query based on the one or more intermediate query results received from the one or more geographically distributed data centers.

20. The computing system of claim 19 ,

wherein determining the replication strategy for the at least one data partition and scheduling the plurality of tasks of the analytical query to the geographically distributed data centers are based on data transfer measurements for each of the plurality of tasks.