Processing of data using a database system in communication with a data processing framework

• US 9,495,427 B2
• Filed: 02/22/2011
• Issued: 11/15/2016
• Est. Priority Date: 06/04/2010
• Status: Active Grant

• Current Assignee: Yale University
• Original Assignee: Yale University
• Inventors: Silberschatz, Avi, Abadi, Daniel, Bajda-Pawlikowski, Kamil, Abouzied, Azza
• Primary Examiner(s): Kerzhner, Aleksandr
• Assistant Examiner(s): TORGRIMSON, TYLER J
• Application Number:
  US13/032,516
• Publication Number:
  US 20110302583A1
• Time in Patent Office: 2,093 Days
• Field of Search: None
• US Class Current: 1/1
• 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, and a storage component in communication with the data processing framework and the plurality database systems. The database systems perform a data processing task. The data processing task is partitioned into a plurality of partitions and each database system processes a partition of the data processing task assigned for processing to that database system. Each database system performs 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 performs at least one partition of the data processing task.

Claims

1. A data processing system, comprising:

a data processing framework having at least one processor and configured to receive a data processing task for processing;

a plurality of database systems having at least another processor, the plurality of database systems being distinct from and are coupled to the data processing framework, wherein the database systems are configured to perform a data processing task, the at least one processor of the data processing framework is distinct from the at least another processor of the plurality of database systems;

wherein the data processing task is configured to be partitioned into a plurality of partitions;

a distributed file system in communication with the data processing framework and the plurality of database systems, and being distinct from each of the data processing framework and the plurality of database systems, the distributed file system optionally stores at least one output data associated with at least one partition of the data processing task being processed by the data processing framework;

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 by the data processing framework based on at least a processing capacity of that database system, wherein the processing capacity is determined based on at least one of the following;

whether that database system is still processing a previously assigned partition of the data processing task, and having each database system determine and provide an indication of its processing capacity to the data processing framework while the data processing task is being processed;

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 process the 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, wherein the storage component stores at least one connection parameter specific to each database system and information about at least one data partition property of data stored in at least one database system, wherein, using the at least one connection parameter and the at least one data partition property, processing of each partition of the data processing task is optimized in accordance with at least one requirement of each database system; and

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

2. The system according to claim 1, further comprising:

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 data set contained within at least one database system, and information about location of a replica of data.

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, and 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 includes at least one map processing phase during which the received data processing task is partitioned into the plurality of partitions for distribution to at least one database system in the plurality of database systems, and at least one reduce processing phase during which data generated as a result of the at least one map processing phase is aggregated and, optionally, at least one operation is performed on the aggregated data.

18. The system according to claim 17, wherein the query is a query performed by the data processing framework having the at least one map processing phase and the at least one reduce processing phase.

38. The system according to claim 1, wherein the data processing framework assigns a second partition of the data processing task to the database system when the data processing task completes processing of a first partition of the data processing task assigned to the database system by the data processing framework, wherein the first partition is processed by the database system before the second partition.

19. A computer implemented method for processing data using a data processing system having a data processing framework having at least one processor, a plurality of database systems having at least another processor, the plurality of database systems being distinct from and are coupled to the data processing framework, the at least one processor of the data processing framework is distinct from the at least another processor of the plurality of database systems, a distributed file system in communication with the data processing framework and the plurality of database systems, and being distinct from each of the data processing framework and the plurality of database systems, the distributed file system optionally stores at least one output data associated with at least one partition of the data processing task being processed by the data processing framework, and a storage component in communication with the data processing framework and the plurality database systems, the method comprising:

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

partitioning the data processing task into a plurality of partitions;

determining and assigning, using the data processing framework, a partition of the data processing task to at least one database system in the plurality of database systems for processing, wherein the data processing framework assigns partitions of data processing task to database systems based on at least a processing capacity of a database system in the plurality of database systems, wherein the processing capacity is determined based on at least one of the following;

whether that database system is still processing a previously assigned partition of the data processing task, and having each database system determine and provide an indication of its processing capacity to the data processing framework while the data processing task is being processed;

processing, using the at least one database systems, the assigned partitions in parallel with processing of at least another partition of the data processing task by another database system in the plurality of database systems;

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, wherein the storage component stores at least one connection parameter specific to each database system and information about at least one data partition property of data stored in at least one database system, wherein, using the at least one connection parameter and the at least one data partition property, processing of each partition of the data processing task is optimized in accordance with at least one requirement of each database system;

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.

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

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 data set contained within at least one database system, and information about location of a replica of data.

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, and 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 includes at least one map processing phase during which the received data processing task is partitioned into the plurality of partitions for distribution to at least one database system in the plurality of database systems, and at least one reduce processing phase during which data generated as a result of the at least one map processing phase is aggregated and, optionally, at least one operation is performed on the aggregated data.

36. The method according to claim 35, wherein the query is a query performed by the data processing framework having the at least one map processing phase and the at least one reduce processing phase.

39. The method according to claim 19, wherein the data processing framework assigns a second partition of the data processing task to the database system when the data processing task completes processing of a first partition of the data processing task assigned to the database system by the data processing framework, wherein the first partition is processed by the database system before the second partition.

37. A non-transitory 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 being distinct from and are coupled to the data processing framework, the at least one processor of the data processing framework is distinct from the at least another processor of the plurality of database systems, a distributed file system in communication with the data processing framework and the plurality of database systems, and being distinct from each of the data processing framework and the plurality of database systems, the distributed file system optionally stores at least one output data associated with at least one partition of the data processing task being processed by 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 the data processing framework;

partitioning the data processing task into a plurality of partitions;

determining and assigning, using the data processing framework, a partition of the data processing task to at least one database system in the plurality of database systems for processing, wherein the data processing framework assigns partitions of data processing task to database systems based on at least a processing capacity of a database system in the plurality of database systems, wherein the processing capacity is determined based on at least one of the following;

whether that database system is still processing a previously assigned partition of the data processing task, and having each database system determine and provide an indication of its processing capacity to the data processing framework while the data processing task is being processed;

processing, using the at least one database systems, the assigned partitions in parallel with processing of at least another partition of the data processing task by another database system in the plurality of database systems;

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, wherein the storage component stores at least one connection parameter specific to each database system and information about at least one data partition property of data stored in at least one database system, wherein, using the at least one connection parameter and the at least one data partition property, processing of each partition of the data processing task is optimized in accordance with at least one requirement of each database system;

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.

40. The computer program product according to claim 37, wherein the data processing framework assigns a second partition of the data processing task to the database system when the data processing task completes processing of a first partition of the data processing task assigned to the database system by the data processing framework, wherein the first partition is processed by the database system before the second partition.