Configuring a system to collect and aggregate datasets

• US 9,317,572 B2
• Filed: 09/08/2010
• Issued: 04/19/2016
• Est. Priority Date: 03/31/2010
• Status: Active Grant

• Current Assignee: Cloudera Incorporated
• Original Assignee: Cloudera Incorporated
• Inventors: Hsieh, Jonathan Ming-Cyn, Robinson, Henry Noel
• Primary Examiner(s): Tiv, Backhean
• Application Number:
  US12/877,902
• Publication Number:
  US 20110246816A1
• Time in Patent Office: 2,050 Days
• Field of Search: 709217-219, 709223-226
• US Class Current: 1/1
• CPC Class Codes: G06F 11/2023 Failover techniques
• CPC Class Codes: G06F 11/3476 Data logging G06F11/14, G06F11/2205 take precedence
• CPC Class Codes: G06F 16/254 Extract, transform and load [ETL] procedures, e.g. ETL data flows in data warehouses
• CPC Class Codes: H04L 41/12 Discovery or management of network topologies
• CPC Class Codes: H04L 67/1042 using topology management mechanisms
• CPC Class Codes: H04L 67/1087 using cross-functional networking aspects

Abstract

Methods for configuring a system to collect and aggregate datasets are disclosed. One embodiment includes, identifying a data source in the system from where dataset is to be collected, configuring a machine in the system that generates the dataset to be collected, to send the dataset to the data source, identifying an arrival location where the dataset that is collected is to be aggregated or written, and/or configuring an agent node by specifying a source for the agent node as the data source in the system and specifying a sink for the agent node as the arrival location.

Claims

1. A method for configuring a system to collect and aggregate datasets, wherein the system comprises agent nodes to collect the datasets, collector nodes to receive the datasets from the agent nodes, and master nodes configured to dynamically change topology among the nodes in the system, the method being executed by a master node and comprising:

identifying a data source in the system from which a dataset is to be collected;

configuring a machine in the system that generates the dataset to send the dataset to the data source;

identifying an arrival location where the dataset is to be aggregated or written;

dynamically configuring, based on system changes, an agent node in an agent tier by;

specifying a source for the agent node as the identified data source in the system;

specifying a sink for the agent node as a collector source of a collector node in a collector tier; and

dynamically configuring, based on the system changes, a collector node in a collector tier by;

specifying the collector source of the collector node in the collector tier as the identified arrival location; and

specifying a collector sink of the collector node in the collector tier as a distributed file system;

wherein the distributed file system is in a storage tier;

wherein the agent node and collector node function as peers in a peer-to-peer network.

2. The method of claim 1, wherein, the system includes multiple machines which generate datasets that are to be collected.

3. The method of claim 1, further comprising:

wherein, the distributed file system is the Hadoop distributed file system (HDFS).

4. The method of claim 1, wherein, the agent node is configurable to perform analytics on the dataset.

5. The method of claim 1, further comprising, launching a web page on a machine to access a master for the system using the master'"'"'s universal resource locator (URL).

6. The method of claim 5, further comprising, scaling the system by implementing multiple masters on multiple machines.

7. The method of claim 6, further comprising, maintaining consistency among the multiple masters using a consensus protocol for ordered state changes.

8. The method of claim 7, wherein, the ordered state changes include data flow configurations and node-to-machine mappings.

9. The method of claim 6, further comprising, maintaining consistency among the multiple masters using a gossip protocol.

10. The method of claim 9, wherein, the gossip protocol is used to maintain consistency among data including one or more of, node status updates, reporting information, and metrics information.

11. The method of claim 5, wherein, configuration of the agent node is performed by accessing a link on the web page for the master.

12. The method of claim 1, wherein, the data source includes, one or more of, a syslog port, an incoming network connection, and an IRC channel.

13. The method of claim 1, wherein, the data source includes, a static file source.

14. The method of claim 1, wherein, the data source includes, one or more of, a status update and a feed.

15. The method of claim 1, wherein, the data source includes, output from execution of a program.

16. The method of claim 1, wherein, the data sink includes a network connection.

17. The method of claim 1, wherein, the data sink includes a physical disk.

18. The method of claim 1, wherein, the data sink includes a console.

19. The method of claim 1, wherein the dynamically configuring comprises:

performing failover by changing the sink for the agent node to another collector source of another collector node when original collector fails; and

performing fan-out increase by assigning the collector node to receive datasets from another collector node.

20. The method of claim 1, wherein the system changes comprise occurrence of one or more of:

load imbalance, node failure, machine failure, new hardware, or a new collector node.

21. A method for configuring a system having multiple machines to collect datasets from the multiple machines and to perform analytics on the datasets, wherein the system comprises agent nodes to collect the datasets, collector nodes to receive the datasets from the agent nodes, and master nodes configured to dynamically change topology among the nodes in the system, the method being executed by a master node and comprising:

identifying data sources on the multiple machines wherein datasets are to be collected from;

configuring the multiple machines in the system that generate the datasets to send the datasets to the data sources;

identifying an arrival location where dataset is to be logged; and

accessing a master through a web page to dynamically specify, based on the system changes, configurations for the multiple machines simultaneously,

wherein the specifying of configurations comprises;

specifying sources for agent nodes in an agent tier as the identified data sources;

wherein each agent node is associated with one of the multiple machines;

specifying a sink for each of the agent nodes in the agent tier as a collector source of a collector node in a collector tier;

specifying the collector source of the collector node in the collector tier as the identified arrival location;

specifying a collector sink of the collector node as a distributed file system in a storage tier;

wherein the agent nodes and collector nodes function as peers in a peer-to-peer network; and

wherein at least one of the agent nodes of the agent tier or the collector node of the collector tier performs analytics on the datasets.

22. The method of claim 21, wherein, the master is used to configure and dynamically reconfigure the agent nodes.

23. The method of claim 21, wherein, statuses of the agent nodes are visible via the master.

24. The method of claim 21, wherein, the distributed file system is the Hadoop distributed file system (HDFS).

25. The method of claim 21, further comprising,

launching a web page on a machine to access the master using the master'"'"'s universal resource locator (URL);

wherein, configuration of the agent nodes for the multiple machines is performed by accessing a link on the web page for the master.

26. The method of claim 21, further comprising, configuring the sink to be a fan-out sink that sends incoming datasets to child sinks coupled to the sink for data replication or for splitting off a CPU intensive tap.

27. The method of claim 21, further comprising, configuring the sink as a failover sink that specifies alternate collectors in case the collector fails.

28. The method of claim 27, wherein, the sink is a failover sink that specifies a local disk to store data until the sink that has failed recovers.

29. A non-transitory machine-readable storage medium having stored thereon instructions which, when executed causes a method for configuring a system having multiple machines to collect datasets from the multiple machines and to perform analytics on the datasets to be performed by a master node, wherein the system comprises agent nodes to collect the datasets, collector nodes to receive the datasets from the agent nodes, and master nodes configured to dynamically change topology among the nodes in the system, the method comprising:

identifying data sources on the multiple machines wherein datasets are to be collected from;

configuring the multiple machines in the system that generate the datasets to send the datasets to the data sources;

identifying an arrival location where dataset is to be logged; and

accessing a master through a web page to dynamically specify, based on the system changes, configurations for the multiple machines simultaneously,

wherein the specifying of the configurations comprises;

specifying sources for agent nodes in an agent tier as the identified data sources;

wherein each agent node is associated with one of the multiple machines;

specifying a sink for each of the agent nodes in the agent tier as a collector source of a collector node in a collector tier;

specifying the collector source of the collector node in the collector tier as the identified arrival location; and

specifying a collector sink of the collector node as a distributed file system in a storage tier;

wherein the agent nodes and collector nodes function as peers in a peer-to-peer network; and

wherein at least one of the agent nodes of the agent tier or the collector node of the collector tier performs analytics on the datasets.

30. The method of claim 29, wherein, the distributed file system is the Hadoop distributed file system (HDFS).