LOW LATENCY QUERY ENGINE FOR APACHE HADOOP

• US 20140280032A1
• Filed: 03/13/2013
• Published: 09/18/2014
• Est. Priority Date: 03/13/2013
• Status: Active Grant

• Current Assignee: Cloudera Incorporated
• Original Assignee: Cloudera Incorporated
• Inventors: Robinson, Henry Noel, Kornacker, Marcel, Erickson, Justin, Li, Nong, Kuff, Lenni, Choi, Alan, Behm, Alex
• Granted Patent: US 9,342,557 B2
• Time in Patent Office: Days
• US Class Current: 707/718
• CPC Class Codes: G06F 16/2453 Query optimisation
• CPC Class Codes: G06F 16/24535 of sub-queries or views
• CPC Class Codes: G06F 16/24542 Plan optimisation
• CPC Class Codes: G06F 16/24544 Join order optimisation
• CPC Class Codes: G06F 16/2471 Distributed queries
• CPC Class Codes: G06F 16/258 Data format conversion from or to a database

Abstract

A low latency query engine for Apache Hadoop that provides real-time or near real-time, ad hoc query capability, while completing batch-processing of MapReduce. In one embodiment, the low latency query engine comprises a daemon that is installed on data nodes in a Hadoop cluster for handling query requests and all internal requests related to query execution. In a further embodiment, the low latency query engine comprises a daemon for providing name service and metadata distribution. The low latency query engine receives a query request via client, turns the request into collections of plan fragments and coordinates parallel and optimized execution of the plan fragments on remote daemons to generate results at a much faster speed than existing batch-oriented processing frameworks.

Claims

1. A system for performing queries on stored data in a distributed computing cluster, comprising:

a plurality of data nodes, each data node having;

a query planner that parses queries from clients to create query fragments;

a query coordinator that distributes the query fragments among the plurality of data nodes; and

a query execution engine that executes query fragments to obtain intermediate results that are aggregated and returned to clients.

2. The system of claim 1, wherein the distributed computing cluster is a Hadoop cluster.

3. The system of claim 2, wherein a query coordinator and a query planner of one of the plurality of data nodes are selected as an initiating query coordinator and an initiating query planner, respectively, for a query from a client.

4. The system of claim 3, wherein the initiating query coordinator and the initiating query planner are selected by a routing component that uses a load balancing scheme to distribute queries from clients among the plurality of data nodes.

5. The system of claim 3, wherein the initiating query coordinator and the initiating query planner are selected based on the client targeting a specific data node from the plurality of data nodes to send the query.

6. The system of claim 3, wherein the query fragments are executed in parallel by query execution engines of data nodes from plurality of data nodes that have data relevant to the query.

7. The system of claim 6, wherein the initiating query coordinator aggregates query results from the query execution engines and provides the aggregated query results to the client.

8. The system of claim 7, wherein prior to sending the query results to the initiating query coordinator, intermediate query results are streamed between the query execution engines for pre-aggregation.

9. The system of claim 6, wherein the query execution engines execute the query fragments directly on Apache HBase data and Hadoop Distributed File System (HDFS) data that comprise the stored data.

10. The system of claim 9, wherein the query execution engines determines a schema-on-read to translate the stored data into an in memory format at run time.

11. The system of claim 2, further comprising a state store that tracks the status of each data node having a query planner, a query coordinator and a query execution engine and distributes metadata.

12. The system of claim 2, wherein the initiating query planner uses information from a name node in the Hadoop cluster to identify data nodes that have relevant data for the query.

13. The system of claim 2, further comprising a low level virtual machine component for run-time code generation and latency reduction.

14. A method of executing a query in a distributed computing cluster having multiple data nodes, comprising:

receiving, by a coordinating data node in the distributed computing cluster, a query; and

distributing, by the coordinating data node, fragments of the query to data nodes in the distributed computing cluster that have data relevant to the query.

15. The method of claim 14, wherein the data nodes execute the fragments of the query on a distributed file system or a data store of the distributed computing cluster.

16. The method of claim 15, wherein the distributed computing cluster is an Apache Hadoop cluster, the distributed file system is a Hadoop Distributed File System (HDFS) and the data store is a “

NoSQL”

(No Structured Query Language) data store.

17. The method of claim 16, wherein the NoSQL data store includes Apache HBase.

18. The method of claim 16, further comprising:

parsing and analyzing the query to determine tasks to be performed by query execution engines running on the data nodes in the Apache Hadoop cluster.

19. The method of claim 18, further comprising:

determining states of the data nodes from a state store, wherein the state store registers the data nodes at start up or after a loss of connection.

20. The method of claim 19, further comprising:

determining location of the data relevant to the query from the state store.

21. The method of claim 16, wherein during execution of the fragments of the query in parallel across the data nodes, intermediate results from the execution are streamed between query execution engines running on the data nodes.

22. The method of claim 16, further comprising:

sending, by the coordinating data node, results received from the data nodes to client.

23. The method of claim 16, further comprising:

receiving, by the coordinating data node, pre-aggregated results of the query from the data nodes; and

performing, by the coordinating data node, an operation on the pre-aggregated results to determine results of the query.

24. The method of claim 23, wherein the operation includes an aggregation operation or an TopN operation.

25. The method of claim 16, wherein the fragments of the query correspond to plans that include partitions along scan boundaries.

26. The method of claim 16, wherein the data node includes the coordinating data node.

27. The method of claim 18, wherein the query execution engines implement a low level virtual machine for run-time code generation to reduce latency.

28. The method of claim 15, further comprising:

receiving, by the coordinating data node, results corresponding to the execution of the fragments of the query from the data nodes.

29. A system of executing a query in a distributed computing cluster, comprising:

means for receiving a query; and

means for parsing and analyzing the query;

means for creating plan fragments of the query; and

means for distributing the plan fragments of the query to data nodes in the distributed computing cluster that have data relevant to the query.

30. The system of claim 29, further comprising:

means for receiving the plan fragments for execution.

31. The system of claim 30, further comprising:

means for executing the plan fragments of the query on a data storage means.

32. The system of claim 31, further comprising:

means for caching meta data for distribution to the means for executing the plan fragments of the query on a data storage means.

33. The system of claim 29, further comprising:

means for determining location of the data relevant to the query.

34. The system of claim 31, further comprising:

means for pre-aggregating results of the query; and

means for communicating the pre-aggregated results of the query.

35. The system of claim 34, further comprising:

means for receiving the pre-aggregated results of the query; and

means for performing an operation on the pre-aggregated results of the query to determine results of the query.