Collecting and aggregating datasets for analysis

• US 9,082,127 B2
• Filed: 09/08/2010
• Issued: 07/14/2015
• 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): To, Baoquoc N
• Application Number:
  US12/877,878
• Publication Number:
  US 20110246460A1
• Time in Patent Office: 1,770 Days
• Field of Search: 707/661, 707/672
• US Class Current: 1/1
• CPC Class Codes: G06F 16/182 Distributed file systems
• CPC Class Codes: G06F 16/24556 Aggregation; Duplicate elimination
• CPC Class Codes: G06Q 30/02 Marketing; Price estimation or determination; Fundraising

Abstract

Systems and methods of facilitating collecting and aggregating datasets that are machine or user-generated for analysis are disclosed. One embodiment includes, collecting a dataset on a machine on which the dataset is received or generated, wherein, the dataset is collected from a data source on the machine, aggregating the dataset collected from the data source at a receiving location, performing analytics on the dataset upon collection or aggregation, and/or writing the dataset aggregated at the receiving location to a storage location.

Claims

1. A method of facilitating collection and aggregation of machine or user generated dataset for analysis, the method comprising:

collecting, by a compute node, the dataset from a data source on a machine,

wherein the dataset is received or generated on the machine;

transmitting the dataset from the data source toward a receiving location by steps including;

recording the dataset as an event using a data model;

extracting a timestamp from the dataset;

specifying, based on the timestamp, a priority of the event in a priority field included in the data model;

specifying, based on the priority, the event in the data model with an attribute in a metadata table included in the data model, wherein the attribute includes a map that directs how the event is to be streamed to a subsequent machine, wherein the metadata table included in the data model is extensible to add additional attributes to the event by the subsequent machine which is configured to further process the dataset as the event is streamed from the data source to the receiving location;

aggregating the dataset collected from the data source at the receiving location,

wherein the receiving location is dynamically updated by the compute node responsive to receiving the configuration information from a master node; and

performing analytics on the dataset responsive to collecting or aggregating the dataset on the machine;

wherein the dataset aggregated at the receiving location is written to a storage location, and

wherein the dataset is stored redundantly on a distributed file system at the storage location.

2. The method of claim 1, wherein, the analytics are performed in near real time relative to collection of the dataset.

3. The method of claim 1, wherein, the storage location is distributed among multiple machines.

4. The method of claim 3, wherein, the distributed file system is the Hadoop Distributed File System (HDFS).

5. The method of claim 1, wherein, mapping of the data source to the receiving location is controlled by a master.

6. The method of claim 5, wherein, the master is executed on a different machine than the machine on which the dataset is received or generated.

7. The method of claim 5, wherein, the master is executed on the machine on which the dataset is received or generated.

8. The method of claim 5, wherein, data flow of the dataset from the data source to the receiving location is configurable and reconfigurable at the master by a user.

9. The method of claim 1, wherein the data source is mapped to a receiving location via a logical node.

10. The method of claim 9, wherein, the logical node contacts the master to obtain configuration information.

11. The method of claim 9, wherein, the logical node adds encryption or a digital signature.

12. The method of claim 9, wherein, the logical node performs batching and unbatching.

13. The method of claim 9, wherein, the logical node performs compression and uncompression.

14. The method of claim 9, wherein, the logical node generates a checksum of the dataset.

15. The method of claim 14, wherein, the logical node verifies the checksum of the dataset.

16. The method of claim 9, wherein, the logical node performs sampling.

17. The method of claim 9, wherein, the logical node performs benchmarking.

18. The method of claim 17, wherein, the logical node injects benchmark tags to measure benchmarks.

19. The method of claim 9, wherein, the logical node performs analytics.

20. The method of claim 19, wherein, the analytics include, one or more of, feature extraction, filtering, and transformation.

21. The method of claim 19, wherein, the analytics include, generation of aggregate counts.

22. The method of claim 19, wherein, the analytics include, generation of statistics.

23. The method of claim 1, wherein, the dataset includes logs from clicks.

24. The method of claim 1, wherein, the dataset includes logs from social networking sites.

25. The method of claim 1, wherein, the dataset includes logs from, one or more of, feeds, status updates.

26. The method of claim 1, wherein, the dataset includes logs from, local or remote applications.

27. The method of claim 1, further comprising:

collecting the dataset from multiple machines in a cluster.

28. The method of claim 1, wherein, the machine includes a web server.

29. The method of claim 1, wherein, the data source includes, one or more of, a syslog port, an incoming network connection, and an Internet Relay Chat (IRC) channel.

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

31. The method of claim 1, wherein, the data source includes, a text file.

32. The method of claim 1, wherein, the data source includes, a Hadoop sequence file formatted file.

33. The method of claim 1, wherein, the receiving location includes a network connection or a network.

34. The method of claim 1, wherein, the receiving location includes a physical disk.

35. The method of claim 1, wherein, the receiving location includes a console.

36. The method of claim 1, wherein, the receiving location includes a text file.

37. The method of claim 1, wherein, the receiving location includes a file in Hadoop sequence file format.

38. The method of claim 1, wherein, the receiving location includes a Hadoop Distributed File System (HDFS) path.

39. The method of claim 1, wherein, the receiving location includes a port.

40. The method of claim 1, wherein, the receiving location includes a monitoring or alerting application.

41. The method of claim 1, wherein, the receiving location includes an Internet Relay Chat (IRC).

42. A system for collecting and aggregating datasets that are machine or user generated, the system comprising:

multiple machines that are simultaneously specifiable by a master,

wherein the multiple machines generate log data and are each associated with an agent node that is dynamically reconfigurable by the master to collect the log data responsive to receiving configuration instructions from the master;

wherein each agent node forwards the log data to a collector tier having a collector node that aggregates the log data from the multiple machines;

wherein, when the log data is forwarded, the collector node processes the log data by steps including;

recording the log data as an event using a data model;

extracting a timestamp from the log data;

specifying, based on the timestamp, a priority of the event in a priority field included in the data model;

annotating, based on the priority, the event in the data model with an attribute in a metadata table included in the data model, wherein the attribute includes a map that directs how the event is to be streamed to a subsequent node, wherein the metadata table included in the data model is extensible to add additional attributes to the event by the subsequent node which is configured to further process the log data as the event is streamed from the data source to a storage system;

wherein the agent node or the collector node performs analytics on the log data responsive to aggregation of the log data from the multiple machines; and

the storage system coupled to the multiple machines;

wherein the storage system includes a distributed file system to which the collector node in the collector tier stores the log data aggregated from the multiple machines.

43. The system of claim 42, wherein, the collector tier is executed on the multiple machines or a different set of machines.

44. The system of claim 43, wherein, the collector node further outputs the log data to multiple destinations.

45. The system of claim 44, wherein, the collector node further outputs the log data to multiple destinations in different formats.

46. The system of claim 43,

wherein, the collector tier comprises multiple collectors combine streams of the log data into larger streams for writing to the storage tier.

47. The system of claim 42,

further comprising, one or more masters running on the multiple machines or another set of machines.

48. The system of claim 47, wherein, the master controls the agent and stores configuration information of the agent node and the collector node;

wherein, the configuration information for the agent is dynamically changed at the master.

49. The system of claim 47, wherein, the agent node and the collector node contact the master to retrieve configuration information.

50. The system of claim 47, wherein, consistency is maintained across the one or more masters using a consensus protocol.

51. The system of claim 47, wherein, consistency is maintained across the one or more masters using a gossip protocol.

52. The system of claim 47, wherein, consistency is maintained across the masters using a combination of a consensus protocol and a gossip protocol.

53. The system of claim 42, further comprising,

wherein, a master of the one or more masters specifies an agent data source and an agent data sink for the agent.

54. The system of claim 42, wherein, the master specifies a configuration of the system.

55. The system of claim 54,

wherein, the master further specifies that an agent data sink for the agent node is a collector data source for the collector node in the collector tier;

wherein, the master further specifies that a collector data sink for the collector node is the distributed file system.

56. The system of claim 55,

wherein, the master is used by a user to configure or reconfigure the agent data sink and agent data source and the master is used by the user to configure or reconfigure the collector data source and the collector data sink.