Low latency query engine for Apache Hadoop

• US 9,342,557 B2
• Filed: 03/13/2013
• Issued: 05/17/2016
• Est. Priority Date: 03/13/2013
• Status: Active Grant

• Current Assignee: Cloudera Incorporated
• Original Assignee: Cloudera Incorporated
• Inventors: Kornacker, Marcel, Erickson, Justin, Li, Nong, Kuff, Lenni, Robinson, Henry Noel, Choi, Alan, Behm, Alex
• Primary Examiner(s): Waldron, Scott A
• Assistant Examiner(s): Wang, Dongming
• Application Number:
  US13/800,280
• Publication Number:
  US 20140280032A1
• Time in Patent Office: 1,161 Days
• Field of Search: 707/718
• US Class Current: 1/1
• 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 HADOOP™

distributed computing cluster having a plurality of data nodes, each data node being a computing device having processing circuitry and memory circuitry, the system comprising;

a state store that tracks a status of each data node, wherein the state store is separate from the data nodes and is further coupled to a name node that tracks where file data are stored across the cluster; and

a plurality of data nodes forming a peer-to-peer network for the queries, each data node functioning as a peer in the peer-to-peer network and being capable of interacting with components of the HADOOP™

cluster, each peer having an instance of a query engine running in memory, each instance of the query engine having;

a query planner configured to;

receive queries from clients;

obtain, from the state store and the name node, (1) membership information regarding all query engine instances that are running in the cluster, and (2) location information regarding where data blocks relevant to the queries are distributed among the plurality of data nodes;

parse queries from clients to create query fragments based on data obtained from the state store and the name node; and

construct a query plan based on the data obtained from the state store;

a query coordinator configured to distribute the query fragments among the plurality of data nodes according to the query plan; and

a query execution engine configured to execute the query fragments, to obtain intermediate results from other data nodes that receive the query fragments, and to aggregate the intermediate results for the clients.

2. The system of claim 1, wherein the distributed computing cluster is configured to store unstructured data.

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 2, wherein

the state store is further coupled to a metadata store that stores metadata relevant to a database management engine implemented in the cluster, and

wherein the query planner is configured to:

obtain, from the state store, metadata associated with the queries.

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

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

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

14. The system of claim 1, wherein the location information includes a plurality of replicas of the data blocks relevant to the queries, and

wherein the query planner or the query coordinator is configured to select one or more, but not all, of the plurality of replicas for execution of the query fragments.

15. The system of claim 1, wherein, when the state store fails, the system is configured to continue to operate based on last information received from the state store.

16. The system of claim 1, wherein all instances of the query engine, at start up, register with the state store and obtain the membership information.

17. The system of claim 1, wherein the membership information is suitable for devising information about all the query engine instances that are running in the cluster.

18. The system of claim 1, wherein the state store caches metadata for running queries and distributes the metadata to query engine instances at start up and/or at a time when the metadata is updated.

19. The system of claim 1, wherein, when the state store fails, rest of the system continues to operate based on last information received from the state store.

20. The system of claim 1, wherein the name node includes details of distribution of files across the data nodes to optimize local reads.

21. The system of claim 1, wherein the name node includes information concerning disk volumes where files are located, on an individual data node.

22. The system of claim 1, wherein the query planner is further configured to use a select number of operators to construct the query plan, and wherein each operator can either generate data or combine data.

23. A method of executing a query in a HADOOP™

distributed computing cluster having multiple data nodes forming a peer-to-peer network for the query, each data node functioning as a peer in the peer-to-peer network and being capable of interacting with components of HADOOP™

cluster, each peer having an instance of a query engine running in memory, each instance of the query engine is configured to perform;

the method comprising;

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

designating the one data node that receives the query as a coordinating data node;

obtaining, by the coordinating data node and through a state store and a name node, (1) membership information regarding all query engine instances that are running in the cluster, and (2) location information regarding where data blocks relevant to the query are distributed among the plurality of data nodes,

wherein the state store is separate from the data nodes;

parsing the query to create fragments of the query based on data obtained from the state store and the name node;

constructing a query plan based on the data obtained from the state store;

distributing, by the coordinating data node and according to the query plan, the fragments of the query to data nodes in the distributed computing cluster that have data relevant to the query;

receiving, from the data nodes having data relevant to the query, intermediate results corresponding to execution of the fragments of the query; and

generating a final result based on the intermediate results for a client.

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

25. The method of claim 24, 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.

26. The method of claim 25, wherein the NoSQL data store include APACHE HBASE™

.

27. The method of claim 25, 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.

28. The method of claim 27, 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.

29. The method of claim 28, further comprising:

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

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

31. The method of claim 25, 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.

32. The method of claim 25, 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.

33. The method of claim 32, wherein the operation includes an aggregation operation or an TopN operation.

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

35. The method of claim 25, wherein the data node includes the coordinating data node.

36. The method of claim 23, further comprising:

sending, by the coordinating data node, the results to the client.

37. The method of claim 23, further comprising:

obtaining, from the state store, metadata associated with the query.

38. The method of claim 23, wherein the location information includes a plurality of replicas of the data blocks relevant to the queries, and the method further comprising:

selecting one or more, but not all, of the plurality of replicas for execution of the fragments of the query.

39. The method of claim 23, further comprising:

upon determining that the state store has failed, continuing to operate based on last information received from the state store.