SYSTEMS AND METHODS FOR PROCESSING DATA

• US 20110302583A1
• Filed: 02/22/2011
• Published: 12/08/2011
• Est. Priority Date: 06/04/2010
• Status: Active Grant

• Current Assignee: Yale University
• Original Assignee: Yale University
• Inventors: ABADI, Daniel, BAJDA-PAWLIKOWSKI, Kamil, ABOUZIED, Azza, SILBERSCHATZ, Avi
• Granted Patent: US 9,495,427 B2
• Time in Patent Office: Days
• US Class Current: 718/102
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 2209/5017 Task decomposition

Abstract

A system, method, and computer program product for processing data are disclosed. The system includes a data processing framework configured to receive a data processing task for processing, a plurality of database systems coupled to the data processing framework, wherein the database systems are configured to perform a data processing task, and a storage component in communication with the data processing framework and the plurality database systems, configured to store information about each partition of the data processing task being processed by each database system and the data processing framework. The data processing task is configured to be partitioned into a plurality of partitions and each database system is configured to process a partition of the data processing task assigned for processing to that database system. Each database system is configured to perform processing of its assigned partition of the data processing task in parallel with another database system processing another partition of the data processing task assigned to the another database system. The data processing framework is configured to perform at least one partition of the data processing task.

Claims

1. A data processing system, comprising:

a data processing framework configured to receive a data processing task for processing;

a plurality of database systems coupled to the data processing framework, wherein the database systems are configured to perform a data processing task;

wherein the data processing task is configured to be partitioned into a plurality of partitions and each database system in the plurality of database systems is configured to process a partition of the data processing task assigned for processing to that database system;

each database system in the plurality of database systems is configured to perform processing of its assigned partition of the data processing task in parallel with another database system in the plurality of database systems processing another partition of the data processing task assigned to the another database system;

wherein the data processing framework is configured to perform at least one partition of the data processing task;

a storage component in communication with the data processing framework and the plurality database systems, configured to store information about each partition of the data processing task being processed by each database system in the plurality of database systems and the data processing framework.

2. The system according to claim 1, further comprising

a database connector component configured to provide a communication interface between the plurality of database systems and the data processing framework;

a data loader component configured to partition data stored in at least one database system and further configured to manage loading of data into the at least one database system; and

a data processing interface in communication with the data processing framework and the plurality of database systems and configured to receive and translate the data processing task for processing by at least one of the plurality of database systems and data processing framework.

3. The system according to claim 2, wherein the data processing task is a query for processing of data.

4. The system according to claim 3, wherein upon receipt of the query, the database connector component is configured to

connect to at least one database system in the plurality of database systems;

execute the received query; and

output results of the executed query.

5. The system according to claim 1, wherein the information stored by the storage component is selected from a group consisting of:

size and location of data, size and location of each data partition, and size and location of at least one replica of data.

6. The system according to claim 3, wherein the storage component is configured to store meta-information about each database system;

wherein the meta-information is selected from a group consisting of;

information about at least one connection parameter of each database system, information about at least one data set contained within at least one database system, information about location of a replica of data, and information about at least one data partition property of data store in at least one database system.

7. The system according to claim 6, wherein the plurality of database systems are configured to access the storage component to obtain information for processing of a query.

8. The system according to claim 6, wherein the meta-information is an XML file.

9. The system according to claim 3, wherein each database system in the plurality of database systems is configured to be disposed on a node in a cluster containing a plurality of nodes, wherein each node in the cluster of nodes is configured to communicate with a master node disposed in a data processing framework.

10. The system according to claim 9, wherein the data loader component is configured to

re-partition data being loaded into the cluster into a plurality of partitions corresponding to the plurality of nodes in the cluster;

partition each portion in the plurality of partitions into a plurality of smaller partitions for storing in the at least one database system on the at least node;

perform bulk-loading of the plurality of smaller partitions of data into the at least one database system on the at least one node.

11. The system according to claim 10, wherein the data loader component further comprises

a global hasher component configured to re-partition the data into the plurality of partitions corresponding to the plurality of nodes in the cluster;

a local hasher component configured to partition each partition in the plurality of partitions into the plurality of smaller partitions using a predetermined maximum smaller partition size.

12. The system according to claim 10, further comprising a query submission interface in communication with the master node and configured to allow submission of a query to the data processing system.

13. The system according to claim 10, wherein upon receipt of a query, the query processing interface is configured to

parse the query;

communicate with the storage component to determine a predetermined data storage structure in the at least one database system;

generate a query execution plan using at least one relational operator;

optimize the query execution plan by generating a specific order in which the relational operators process the data;

execute the query in accordance with the optimized query execution plan.

14. The system according to claim 13, wherein the at least one relational operator is configured to process data and is selected from a group consisting of:

a filter operator, a select operator, a join operator, an aggregation operator, a repartition operator;

wherein each operator is configured to process the data prior to the next operator processing the data.

15. The system according to claim 14, wherein the specific order of relational operator includes processing data using the repartition operator prior to processing data using the join operator.

16. The system according to claim 3, wherein the query is an SQL query.

17. The system according to claim 3, wherein the data processing framework is a MapReduce framework.

18. The system according to claim 17, wherein the query is a MapReduce query.

19. A method for processing data using a data processing system having a data processing framework, a plurality of database systems coupled to the data processing framework, and a storage component in communication with the data processing framework and the plurality database systems, the method comprising the steps of:

receiving a data processing task for processing using a data processing framework;

partitioning the data processing task into a plurality of partitions;

assigning a partition of the data processing task to a database system in the plurality of database systems for processing;

using the database systems to which partitions have been assigned for processing, processing the assigned partitions in parallel to one another;

processing at least one partition of the data processing task using the data processing framework; and

using the storage component, storing information about each partition of the data processing task being processed by each database system in the plurality of database systems and the data processing framework.

20. The method according to claim 19, wherein the data processing system further includes

a database connector component configured to provide a communication interface between the plurality of database systems and the data processing framework;

a data loader component configured to partition data stored in at least one database system and further configured to manage loading of data into the at least one database system; and

a data processing interface in communication with the data processing framework and the plurality of database systems and configured to receive and translate the data processing task for processing by at least one of the plurality of database systems and data processing framework.

21. The method according to claim 20, wherein the data processing task is a query for processing of data.

22. The method according to claim 21, further comprising

using the database connector component, upon receipt of the query

connecting to at least one database system in the plurality of database systems;

executing the received query; and

outputting results of the executed query.

23. The method according to claim 19, wherein the information stored by the storage component is selected from a group consisting of:

size and location of data, size and location of each data partition, and size and location of at least one replica of data.

24. The method according to claim 21, further comprising

using the storage component, storing meta-information about each database system;

wherein the meta-information is selected from a group consisting of;

information about at least one connection parameter of each database system, information about at least one data set contained within at least one database system, information about location of a replica of data, and information about at least one data partition property of data store in at least one database system.

25. The method according to claim 24, further comprising

using the plurality of database systems, accessing the storage component to obtain information for processing of a query.

26. The method according to claim 24, wherein the meta-information is an XML file.

27. The method according to claim 21, wherein each database system in the plurality of database systems is configured to be disposed on a node in a cluster containing a plurality of nodes, wherein each node in the cluster of nodes is configured to communicate with a master node disposed in a data processing framework.

28. The method according to claim 27, further comprising

using the data loader component

re-partitioning data being loaded into the cluster into the plurality of partitions corresponding to the plurality of nodes in the cluster;

partitioning each portion in the plurality of partitions into a plurality of smaller partitions for storing in the at least one database system on the at least node;

bulk-loading of the plurality of smaller partitions of data into the at least one database system on the at least one node.

29. The method according to claim 28, wherein the data loader component further comprises

a global hasher component configured to re-partition the data into the plurality of partitions corresponding to the plurality of nodes in the cluster;

a local hasher component configured to partition each partition in the plurality of partitions into the plurality of smaller partitions using a predetermined maximum smaller partition size.

30. The method according to claim 28, wherein the data processing system further includes a query submission interface in communication with the master node and configured to allow submission of a query to the data processing system.

31. The method according to claim 28, further comprising

using the query processing interface, upon receipt of a query

parsing the query;

communicating with the storage component to determine a predetermined data storage structure in the at least one database system;

generating a query execution plan using at least one relational operator;

optimizing the query execution plan by generating a specific order in which the relational operators process the data;

executing the query in accordance with the optimized query execution plan.

32. The method according to claim 31, wherein the at least one relational operator is configured to process data and is selected from a group consisting of:

a filter operator, a select operator, a join operator, an aggregation operator, a repartition operator;

wherein each operator is configured to process the data prior to the next operator processing the data.

33. The method according to claim 32, wherein the specific order of relational operator includes processing data using the repartition operator prior to processing data using the join operator.

34. The method according to claim 21, wherein the query is an SQL query.

35. The method according to claim 21, wherein the data processing framework is a MapReduce framework.

36. The method according to claim 35, wherein the query is a MapReduce query.

37. A computer program product, tangibly embodied in a computer-readable medium, the computer program product being operable to cause a data processing system having a data processing framework, a plurality of database systems coupled to the data processing framework, and a storage component in communication with the data processing framework and the plurality database systems, to perform operations comprising:

receiving a data processing task for processing using a data processing framework;

partitioning the data processing task into a plurality of partitions;

assigning a partition of the data processing task to a database system in the plurality of database systems for processing;

using the database systems to which partitions have been assigned for processing, processing the assigned partitions in parallel to one another;

processing at least one partition of the data processing task using the data processing framework; and

using the storage component, storing information about each partition of the data processing task being processed by each database system in the plurality of database systems and the data processing framework.