Method and system for collecting and analyzing time-series data

• US 7,979,439 B1
• Filed: 03/14/2006
• Issued: 07/12/2011
• Est. Priority Date: 03/14/2006
• Status: Expired due to Fees

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Thompson, Aaron C., Nordstrom, Paul G.
• Primary Examiner(s): Robinson; Greta L
• Assistant Examiner(s): Hu; Jensen
• Application Number:
  US11/375,636
• Time in Patent Office: 1,946 Days
• Field of Search: 707/102, 707/741
• US Class Current: 707/741
• CPC Class Codes: G06F 16/22 Indexing; Data structures therefor; Storage structures

Abstract

A computer-implemented data processing method comprises receiving an index specification, storing data in a data repository, and indexing the data to create an index of the date stored in the data repository. The index specification comprises a user-specific index parameter. The data is indexed along a dimension of the data specified by the user-specified index parameter. The is received from data source computers and may be indexed as the data is received from the source computers.

Claims

1. A computer-implemented data collection and analysis method comprising:

receiving an index specification comprising an index parameter specified by a user, wherein the index parameter corresponds to at least one of a product identifier, a session identifier, and a visitor identifier;

receiving, from a plurality of data source computers, time-series data relating to contents of web pages provided to users of a website, wherein the time-series data is in the form of data messages;

converting the data messages into datapoints;

storing the time-series data in a data repository;

indexing the time-series data to create an index of the time-series data stored in the data repository, the time-series data being indexed along a dimension of the time-series data specified by the user-specified index parameter; and

storing information concerning the user-specified index parameter in a calculation table, the calculation table comprising calculation descriptors received from a plurality of host computers, the calculation descriptors describing desired at least one of data analysis datapoints and data index datapoints of the system, to perform analysis of the data on the plurality of host computers,

wherein a datapoint comprises;

a datakey which provides information to allow the datapoint to be properly routed in accordance with a type of analysis to be performed on the data, wherein the type of analysis to be performed on the data is based on the received index specification,

a data value which provides the data to be processed, and

a data interval which provides the time interval associated with the datapoint, and

wherein the analysis corresponds to at least one of a detection of shifts in purchasing patterns, a detection of purchasing sales levels, an evaluation of the effectiveness of promotions, real-time performance statistics for analysis of website traffic, real-time website performance statistics for analysis of purchasing trends, and historical website performance statistics for evaluation of customer experiences.

2. A method as defined in claim 1, further comprising communicating the user-specified index parameter to a plurality of host computers.

3. A method as defined in claim 2, further comprising receiving the data at the plurality of host computers from a plurality of data source computers.

4. A method as defined in claim 1, wherein the information concerning the user-specified index parameter is stored in the form of a calculation descriptor in the calculation table.

5. A method as defined in claim 4, further comprising maintaining at least partial consistency between copies of the calculation table stored at the plurality of host computers, and wherein the user-specified index parameter is communicated to the plurality of host computers to maintain consistency of the copies of the calculation table.

6. A method as defined in claim 5, wherein the data is time-series data, and wherein the plurality of host computers are configured to analyze the time-series data from the plurality of data source computers in accordance with the calculation descriptors.

7. A method as defined in claim 4, wherein the index comprises a plurality of indices and the calculation descriptor comprises a plurality of calculation descriptors, and wherein the plurality of indices comprises indices that are indexed along dimensions that are different from each other and that are user-specified using the calculation descriptors.

8. A method as defined in claim 1, wherein the index comprises a plurality of layers of indices including a first layer and a second layer.

9. A method as defined in claim 8, wherein the first layer comprises an index of second layer indices, and wherein the first layer is organized along a first dimension.

10. A method as defined in claim 9, wherein the user-specified index parameter comprises a plurality of user-specified index parameters, and wherein the second layer indices are indexed along dimensions that are different from each other and are user-specified using the plurality of user-specified index parameters.

11. A method as defined in claim 1, further comprising storing the index to facilitate subsequent retrieval of the data stored in the data repository using the index.

12. A method as defined in claim 1, wherein the user is a computer.

13. A method as defined in claim 11, wherein the index is generated in substantially real-time as the time-series data is received from the data source computers.

14. A method as defined in claim 13, wherein the number of data source computers is in excess of one thousand.

15. A method as defined in claim 13, wherein the number of data source computers is in excess of five thousand.

16. A method as defined in claim 13, wherein the index is lossy and is created using a Bloom filter.

17. A system for collecting and analyzing time-series data from a plurality of data source computers external to the system, wherein the data is received in the form of data messages that will be converted into datapoints by the system, comprising:

a data repository;

a plurality of host computers in communication with the data repository;

a calculation table comprising a plurality of calculation descriptors received from a plurality of user computers, the calculation table being accessible by the plurality of host computers, the calculation descriptors describing at least one of desired data analysis datapoints and desired data index datapoints, wherein a datapoint comprises;

a datakey which provides information to allow the datapoint to be properly routed based depending on analysis to be performed on the datapoint, wherein the analysis to be performed on the datapoint is based at least one of a product identifier, a session identifier, and a visitor identifier specified by a user;

a data value which provides the data to be processed, and

a data interval which provides the time interval associated with the datapoint;

a plurality of computer-implemented partitions associated with the plurality of host computers, the plurality of partitions being configured to (i) analyze the time-series data from the plurality of data source computers to produce the desired data analysis datapoints in accordance with the calculation descriptors specifying the desired data analysis datapoints, and (ii) generate the desired data index datapoints in accordance with the calculation descriptors specifying the desired data index datapoints.

18. A system as defined in claim 17, wherein the plurality of partitions comprise a plurality of processes, the plurality of processes being instantiated responsive at least in part to the addition of the calculation descriptors in the calculation table.

19. A system as defined in claim 18, wherein the calculation table comprises a plurality of copies, the copies of the calculation table being maintained at the plurality of hosts.

20. A system as defined in claim 19, wherein consistency is maintained between the copies of the calculation table maintained at the plurality of hosts using a communication protocol.

21. A system as defined in claim 20, wherein the protocol is a gossip protocol.

22. A system as defined in claim 20, further comprising a plurality of nodes implemented on the plurality of hosts, the plurality of nodes owning respective ones of the plurality of partitions, and the plurality of partitions comprising a plurality of processes created based on contents of the calculation table and configured to produce the desired analysis of the data.

23. A system as defined in claim 22, wherein the plurality of computer-implemented nodes are configured to receive data messages from the plurality of data source computers.

24. A system as defined in claim 23, wherein the plurality of partitions are configured (i) to permit each of the data messages to be received at an arbitrary one of the plurality of nodes, and (ii) to forward the received data messages to one or more of the plurality of nodes which own respective ones of the plurality of partitions responsible for participating in producing one of the desired analysis of the data.

25. A system as defined in claim 17, wherein the calculation table is configured to be updated by the plurality of user computers during operation of the system to cause new desired analysis of the data to be produced, including being configured to receive new output descriptors to cause new desired index datapoints to be produced.

26. A system as defined in claim 17, wherein the index datapoints create an index.

27. A system as defined in claim 26, wherein the index comprises a plurality of layers, and wherein the first layer comprises an index organized along a first dimension.

28. A system as defined in claim 27, wherein the second layer comprises a plurality of indexes constructed based on the index datapoints.

29. A system as defined in claim 28, wherein the plurality of indexes are organized along dimensions that are different from each other and are user-specified using the calculation descriptors in the calculation table.

30. A system as defined in claim 29, wherein the data repository stores datapoints including (i) datapoints generated as a result of the analysis of the time-series data from the data source computers in accordance with the calculation descriptors, and (ii) the index datapoints.

31. A system as defined in claim 17, wherein the plurality of data source computers each provide information to users of a network service, and wherein the data is time-series data relating to the information provided to the users of the network service.

32. A system as defined in claim 31, wherein the network service is a website, and wherein the plurality of data source computers each publish web pages to visitors of a web site, and the time-series data relating to the contents of the web pages.

33. A system as defined in claim 32,

wherein the plurality of partitions are configured to generate the index datapoints for time-series data received from each of the plurality of data source computers;

and

wherein the plurality of partitions are configured to generate the index datapoints at a rate which is not less than the rate at which new time-series data is received from the plurality of data source computers.

34. A system as defined in claim 33, wherein the number of data source computers is in excess of one thousand.

35. A system as defined in claim 33, wherein the number of data source computers is in excess of five thousand.

36. A method of collecting, analyzing and indexing time-series data received from a plurality of data source computers, comprising:

receiving the time-series data in the form of data messages at a plurality of partitions, the plurality of partitions being implemented on a plurality of data collection and analysis computers, each of the plurality of partitions comprising a plurality of processes to distribute workload across nodes, wherein routing to an appropriate partition is as indicated in a calculation table;

analyzing the data messages at the plurality of processes to generate datapoints, the datapoints comprising user output datapoints and index datapoints, wherein a datapoint comprises;

a datakey which provides information to allow the datapoint to be properly routed at least partially depending on a type of processing to be performed on the datapoint, wherein the type of processing to be performed on the datapoint is defined by at least one of a product identifier, a session identifier, and a visitor identifier specified by a user;

a data value which provides the data to be processed, and

a data interval which provides the time interval associated with the datapoint,

the user output datapoints and the index datapoints both being generated in response to calculation descriptors received from a plurality of user computers;

storing the index datapoints in a data repository; and

providing the user output datapoints to the plurality of user computers.

37. A method as defined in claim 36, wherein the plurality of data collection and analysis computers in combination receive the time-series data from each of the plurality of data source computers, and wherein the number of data source computers is in excess of five thousand.

38. A method as defined in claim 36,

wherein the index datapoints comprise real-time index datapoints and historical datapoints, the real-time index datapoints being generated based on the substantially real-time analysis;

wherein, after the time-series data is received at the plurality of partitions, the time-series data is stored in a data repository; and

wherein analyzing the time-series data to generate the index datapoints further comprises performing a historical analysis of the time-series data contained in the data repository to generate the historical index datapoints.

39. A method as defined in claim 38, wherein the historical index datapoints are generated in response to calculation descriptors received from the plurality of user computers after the time-series data that is indexed is received.

40. A method as defined in claim 39, further comprising assembling the index datapoints to create an index, wherein the index comprises a plurality of layers, wherein the first layer comprises an index organized along a first dimension, wherein the second layer comprises a plurality of indexes constructed based on the index datapoints, wherein the plurality of indexes are organized along dimensions that are different from each other and are user-specified using the calculation descriptors received from a plurality of user computers.

41. A method as defined in claim 40, wherein the second layer comprises both the real-time index datapoints and the historical index datapoints.

42. A method as defined in claim 40, wherein the first dimension is time.

43. A computer-implemented data collection and analysis method comprising:

receiving an index specification comprising a user-specified index parameter at a host computer, the host computer being one of a plurality of host computers;

storing information concerning the user-specified index parameter in a calculation table, the calculation table comprising a plurality of calculation descriptors inserted by a plurality of user computers, the calculation descriptors describing desired at least one of data analysis datapoints and data index datapoints of the plurality of host computers, wherein a datapoint comprises;

a datakey which provides information to allow the datapoint to be properly routed at least partially depending on a type of processing to be performed on the datapoint, wherein the type of processing to be performed on the datapoint is defined by the index specification received,

a data value which provides the data to be processed, and

a data interval which provides the time interval associated with the datapoint,

and the information concerning the user-specified index parameter being stored in the form of a calculation descriptor in the calculation table;

communicating the index specification to remaining ones of the plurality of host computers;

receiving time-series data in the form of data messages at the plurality of host computers from a plurality of data source computers;

analyzing the data messages at the plurality of host computers in accordance with the calculation descriptors to produce the desired at least one of data analysis datapoints and data index datapoints;

storing the time-series data in a data repository;

indexing the time-series data to create an index of the time-series data stored in the data repository, the data being indexed along a dimension of the data specified by the user-specified index parameter, the index being created substantially in real time as the data is received from the plurality of data source computers; and

storing the index to permit subsequent retrieval of the data stored in the data repository using the index.

44. A method as defined in claim 43, wherein the number of data source computers is in excess of one thousand.

45. A method as defined in claim 43, wherein the number of data source computers is in excess of five thousand.

46. A method as defined in claim 43, wherein the plurality of data source computers each provide information to users of a network service, and wherein the time-series data relates to the information provided to the users of the network service.

47. A method as defined in claim 46, wherein the network service is a website, wherein the information provided to the users is provided on web pages, and wherein the time-series data relates to contents of web pages provided to the users.

48. A non-transitory machine-readable storage media whose contents direct a computing system to:

receive an index specification comprising a user-specified index parameter;

store data in a data repository;

index the data to create an index of the data stored in the data repository, the data being indexed along at least one dimension of the data specified by the user-specified index parameter; and

store information concerning the user-specified index parameter in a calculation table, the calculation table comprising calculation descriptors received from a plurality of host computers, the calculation descriptors describing desired at least one indexing datapoints on the plurality of host computers, wherein an indexing datapoint comprises;

a datakey which provides information to allow the datapoint to be properly routed at least partially depending on a type of processing to be performed on the datapoint, wherein the type of processing to be performed on the datapoint is based on the index specification received,

a data value which provides the data to be processed, and

a data interval which provides the time interval associated with the datapoint.

49. A machine-readable storage media as defined in claim 48, wherein the contents further direct the computing system to store the index to facilitate subsequent retrieval of the data stored in the data repository using the index.

50. A machine-readable storage media as defined in claim 48, wherein the contents further direct the computing system to communicate the user-specified index parameter to a plurality of nodes implemented on a plurality of host computers.

51. A machine-readable storage media as defined in claim 50, wherein the contents further direct the computing system to receive the data at the plurality of host computers from a plurality of data source computers.

52. A machine-readable storage media as defined in claim 51, wherein the information concerning the user-specified index parameter is stored in the form of a calculation descriptor in the calculation table.

53. A machine-readable storage media as defined in claim 52, wherein the contents further direct the computing system to at least partially maintain consistency between copies of the calculation table stored at the plurality of host computers, and wherein the user-specified index parameter is communicated to the plurality of host computers to maintain consistency of the copies of the calculation table.

54. A machine-readable storage media as defined in claim 53, wherein the data is time-series data, and wherein the plurality of host computers are configured to analyze the time-series data from the plurality of data source computers in accordance with the calculation descriptors.

55. A machine-readable storage media as defined in claim 52, wherein the index comprises a plurality of indices and the calculation descriptor comprises a plurality of calculation descriptors, and wherein the plurality of indices comprise indices that are organized along dimensions that are different from each other and that are user-specified using the calculation descriptors.

56. A machine-readable storage media as defined in claim 48, wherein the index comprises a plurality of layers of indices including a first layer and a second layer.

57. A machine-readable storage media as defined in claim 56, wherein the first layer comprises an index of second layer indices, and wherein the first layer is organized along a first dimension.

58. A machine-readable storage media as defined in claim 57, wherein the user-specified index parameter comprises a plurality of user-specified index parameters, and wherein the second layer indices are organized along dimensions that are different from each other and are user-specified using the plurality of user-specified index parameters.

59. A machine-readable storage media as defined in claim 48, wherein the data is received from a plurality of data source computers, wherein the plurality of data source computers each provide information to users of a network service, and wherein the data is time-series data relating to the information provided to the users of the network service.

60. A machine-readable storage media as defined in claim 59, wherein the index is generated in substantially real-time as the time-series data is received from the data source computers.

61. A machine-readable storage media as defined in claim 60, wherein the number of data source computers is in excess of one thousand.

62. A machine-readable storage media as defined in claim 60, wherein the number of data source computers is in excess of five thousand.

63. A machine-readable storage media as defined in claim 60, wherein the network service is a website, wherein the information provided to the users is provided on web pages, and wherein the time-series data relates to contents of web pages provided to the users.