System and method for storing data streams in a distributed environment

• US 8,738,572 B2
• Filed: 08/31/2012
• Issued: 05/27/2014
• Est. Priority Date: 05/24/2011
• Status: Active Grant

• Current Assignee: Mr. Yusefzadeh, Bahram Nmn
• Original Assignee: Red Lambda Incorporated
• Inventors: Bird, Robert, Leko, Adam, Whitlock, Matthew
• Primary Examiner(s): To, Baoquoc N
• Application Number:
  US13/600,853
• Publication Number:
  US 20120330909A1
• Time in Patent Office: 634 Days
• Field of Search: 707/636
• US Class Current: 707/636
• CPC Class Codes: G06F 16/24568 Data stream processing; Continuous queries

Abstract

Systems, methods and computer readable medium for storing data elements transmitted via data streams received from distributed devices connected via a network. The received data elements may be stored in block stores on the distributed devices. The stored data elements may be allocated to data blocks of a block store that have assigned block identifiers and further allocated to events of the data blocks. Stream schema of the received plurality of data streams may have the same stream schema, and indices may be generated indices based on the order of the event allocated data elements. Stream schema of the received data streams may comprise a list of token names. Token names may be assigned to the event allocated data elements. Indices may be generated for the event allocated data elements based on the stream schema.

Claims

1. A method for storing data, comprising the steps of:

receiving a plurality of data-streams, wherein the plurality of data-streams are received from a plurality of distributed devices, wherein the plurality of distributed devices are connected via a network, wherein each one of the plurality of data-streams comprise data-elements;

storing the data-elements of the received plurality of data-streams, wherein the stored data-elements are stored in a plurality of block-stores, wherein the block-stores are stored on the plurality of distributed devices;

assigning block-identifiers to data-blocks of the block-stores, wherein the block-identifiers are globally unique, wherein each of the block-stores comprises one or more data-blocks, wherein the data-blocks of a single data-stream are logically grouped, and wherein each of the data-blocks are read and written as a single unit;

allocating the stored data-elements to the data-blocks of the block-stores, wherein each data-block comprises the stored data-elements of only one of the received plurality of data-streams;

further allocating the block-allocated data-elements to events of the data-blocks, wherein each of the data-blocks comprise one or more events, wherein each of the events comprises the block-allocated data-elements of the corresponding data-block; and

,

generating indices for the event-allocated data-elements.

2. The method of claim 1, wherein each of the received plurality of data-streams has a stream-schema, wherein each of the stream-schema is the same, and wherein the indices are generated based on the order of the event-allocated data-elements in each event.

3. The method of claim 1, further comprising the steps of:

reading stream-schema of the received plurality of data-streams, wherein each stream-schema comprises a list of token-names for the event-allocated data-elements of the corresponding data-stream, wherein the event-allocated data-elements of the corresponding data-stream are capable of being identified based on the list of token-names; and

,

assigning to each of the event-allocated data-elements a token-name, wherein the token-names are human-readable names,

wherein the indices are generated based on the stream-schema.

4. The method of claim 1, wherein the generated indices have index-pages and index-sets based on stream-schema of the received plurality of data-streams.

5. The method of claim 1, further comprising the steps of:

executing at least one query of the event-allocated data-elements based on at least one of the generated indices; and

,

retrieving the event-allocated data-elements based on results from at least one of the executed query.

6. The method of claim 1, wherein each of the plurality of block-stores stores the stored data-elements of two or more of the received plurality of data-streams.

7. The method of claim 3, wherein each stream-schema further comprises an index-definition for configuring generated indices for each corresponding data-stream, wherein the generated indices have index-pages and index-sets based on index-definitions of the corresponding stream-schema.

8. The method of claim 7, wherein each of the index-pages further comprises data-block references based on the block-identifiers for the corresponding data-blocks.

9. The method of claim 7, wherein the index-page further comprises page-data summarized-information of the data-blocks of the corresponding data-stream, and wherein the index-set further comprises root-data summarized-information of the index-pages of the corresponding index-definition.

10. The method of claim 1, wherein the data-elements have a common structure.

11. The method of claim 10, wherein the common structure of the data-elements is an unrestricted-structure.

12. The method of claim 10, wherein each stream-schema corresponds to a common structure of each of the events of the corresponding data-stream.

13. The method of claim 1, wherein the data-elements have different structures, wherein each stream-schema corresponds to varying structures of the events of the corresponding data-streams.

14. The method of claim 1, wherein the stored data-elements comprise compound-data.

15. The method of claim 14, wherein the compound-data is selected from a group consisting of lists and objects.

16. The method of claim 1, wherein the block-stores support concurrent readers and writers.

17. The method of claim 1, wherein the data-elements that are received from a distributed device are stored in a targeted block-store only when the distributed device has write-access to a file-system utilized by the targeted block-store.

18. The method of claim 3, further comprising the steps of:

receiving a request to write data of a data-stream from a distributed device;

reading the stream-schema of the received data-stream;

allocating local-memory to buffer each of the events of the corresponding data-stream; and

caching metadata of the stream-schema.

19. The method of claim 3, further comprising the step of:

caching metadata of the stream-schema of the received plurality of data-streams, wherein each stream-schema comprises metadata for the corresponding data-stream.

20. The method of claim 1, further comprising the steps of:

filling a buffer until the buffer reaches a maximum block-size provided by the block-size information for the data-blocks of the corresponding data-stream, wherein each stream-schema comprises block-size information for the data-blocks of the corresponding data-stream, wherein the block-size information provides a maximum block-size for data-blocks within the corresponding data-stream; and

flushing the buffer into a file-system utilized by the targeted block-store, wherein the buffer is capable of being reused for the data-elements of the received plurality of data-streams.

21. The method of claim 1, further comprising the step of:

allocating more than one buffer to a single data-stream, wherein concurrent-writers are utilized in the same distributed device.

22. The method of claim 1, comprising the step of:

performing a transaction on the event-allocated data-elements, wherein the transaction is one transaction selected from the group consisting of querying, retrieving, accessing, updating, and deleting.

23. The method of claim 7, further comprising the steps of:

executing a query of the event-allocated data-elements based on a Boolean-expression; and

,

retrieving a result set of event-allocated data-elements based on results of the executed query.

24. The method of claim 23, wherein the Boolean-expression is selected from a group consisting of a product-of-sums expression and a sum-of-products expression.

25. The method of claim 23, wherein the step of executing a query comprises the steps of:

identifying the index-definitions which reference token-names that match token-name constraints of a Boolean-expression of a query for the event-allocated data-elements, wherein the index-definitions reference a subset of the list of token-names of the corresponding stream-schema, wherein the referenced token-names of the index-definitions are compared to the token-name constraints of the Boolean-expression;

determining a candidate set of index-pages based on the identified index-definitions, wherein the index-sets of the generated indices that match the configuration of the corresponding identified index-definitions are compared to index-set constraints of the Boolean-expression;

determining a candidate set of data-blocks based on the candidate set of index-pages;

wherein the candidate set of index-pages are compared to index-page constraints of the Boolean-expression; and

,

determining the result set of event-allocated data-elements based on the candidate set of data-blocks, wherein the candidate set of data-blocks are compared to data-blocks constraints of the Boolean-expression,

wherein the result set of event-allocated data-elements comprise a maximal set of event-allocated data-elements that satisfy the Boolean-expression of the executed query.

26. The method of claim 25, further comprising the step of:

only if the referenced token-names of the index-definitions fail to match the token-name constraints of the Boolean-expression, setting a candidate set of data-blocks to comprise all data-blocks in the corresponding data-stream.

27. The method of claim 25, further comprising the steps of:

evaluating the candidate set of data-blocks based on the Boolean-expression; and

,

ignoring false-positives returned from the executed query, wherein event-allocated data-elements that fail to satisfy the Boolean-expression are removed from the result set of event-allocated data-elements.

28. The method of claim 25, further comprising the steps of:

updating at least one of the candidate set of data-blocks, wherein the index-page referencing the at least one of the candidate set of data-blocks is updated, wherein the index-set referencing the updated index-page is updated, wherein at least one of the data-elements of the result set of event-allocated data-elements is updated, wherein the update for the data-element is selected from a group consisting of a deletion of an event-allocated data-element, an insertion of an event-allocated data-element, or a modification of an event-allocated data-element.

29. The method of claim 5, further comprising the step of:

using the results from the at least one query as input to a streaming-query, wherein the streaming-query provides join and windowing operations for event-allocated data-elements of the received plurality of data-streams.

30. The method of claim 1, further comprising the step of:

removing data-blocks based on a file-deletion operation and updated indices that reference that data block.

31. The method of claim 30, wherein the file-deletion operation is a batched operation.

32. The method of claim 30, wherein the file-deletion operation is performed on a subsequent index-update operation.

33. The method of claim 1, wherein at least one of the block-stores requires a file-system that supports concurrent updates to directory-structures, wherein the at least one of the block-stores requires that writes to data-blocks are atomic, wherein the data-blocks of the at least one of block-stores are consistent for each query performed after new data-blocks are written.

34. The method of claim 33, wherein a locking service is utilized by the file-system.

35. A method of claim 34, wherein the locking service is utilized except when the writes to the data-blocks are atomic and writers are notified when a read-update-write transaction fails due to a concurrent write.

36. The method of claim 7, furthering comprising the step of:

flushing pending data-blocks and corresponding index-pages and index-sets.

37. The method of claim 1, wherein at least one of the block-stores has unconstrained concurrency, wherein the unconstrained concurrency of the at least one of the block-stores is limited by file-system capabilities.

38. The method of claim 1, further comprising the steps of:

compressing the data-blocks, the index-pages and the index-sets;

flushing the compressed data-blocks, index-pages and index-sets; and

decompressing the flushed data-blocks, index-pages and index-sets.

39. The method of claim 1, wherein at least one of the plurality of distributed devices is a distributed storage device.

40. The method of claim 1, wherein at least one of the plurality of distributed devices is a sensor.

41. The method of claim 1, wherein at least one of the plurality of distributed devices is an infrastructural equipment.

42. The method of claim 1, wherein at least one of the indices is a hash-index.

43. The method of claim 1, wherein at least one of the indices is a range-index.

44. The method of claim 1, wherein at least one of the indices is an interval-index.

45. The method of claim 1, wherein at least one of the indices is a full-text index.

46. The method of claim 1, wherein at least one of the block-stores stores data based on an encoding standard.

47. The method of claim 46, wherein the encoding standard is an Abstract Syntax Notation One (ASN.1) encoding standard.

48. The method of claim 1, wherein at least one of the block-stores stores data based on a message-transfer syntax.

49. The method of claim 48, wherein the message-transfer syntax is a Distinguished Encoding Rules (DER) message-transfer syntax.

50. The method of claim 49, wherein the Distinguished Encoding Rules (DER) message-transfer syntax is based on a X.690 International Telecommunication Union (ITU)-T recommendation.

51. The method of claim 1, further comprising the steps of:

filling a buffer with data-elements of the received plurality of data-streams;

flushing the buffer into a file-system; and

,

writing the data-elements into the file-system.

52. A system for storing data, comprising:

a plurality of block-stores stored on a plurality of distributed devices, the plurality of distributed devices adapted to communicate via a network, the plurality of block-stores storing data-elements received via a plurality of data-streams from the plurality of distributed devices;

block-identifiers assigned to data-blocks of the block-stores, each of the block-stores comprising one or more data-blocks, the stored data-elements allocated to the one or more data-blocks of the block-stores, each data-block comprising the stored data-elements of only one of the received plurality of data-streams;

events of the one or more data-blocks, the block-allocated data-elements further allocated to the events of the data-blocks, each of the data-blocks comprise one or more events; and

,

indices generated for the event-allocated data-elements.

53. The system of claim 52, the data-blocks of a single data-stream are logically grouped.

54. The system of claim 52, each of the data-blocks are read and written as a single unit.

55. The system of claim 52, the block-identifiers being globally unique.

56. The system of claim 52, further comprising:

stream-schema of the received plurality of data-streams, each of the received plurality of data-streams having the same stream-schema, the generated indices based on the order of the event-allocated data-elements in each event.

57. The system of claim 52, further comprising:

stream-schema of the received plurality of data-streams, each of the received plurality of data-streams having a stream-schema, each stream-schema comprising a list of token-names for the event-allocated data-elements of the corresponding data-stream, the event-allocated data-elements of the corresponding data-stream being capable of being identified based on the list of token-names, wherein each stream-schema further comprises an index-definition for configuring generated indices for each corresponding data-stream; and

,

token-names assigned to each of the event-allocated data-elements, each one of the event-allocated data-elements has a token-name, the token-names being human-readable names, the generated indices based on the corresponding index-definition and the corresponding list of token-names.

58. A non-transitory computer readable medium having computer readable instructions stored thereon for execution by a processor, wherein the instructions on the non-transitory computer readable medium are adapted to:

receive a plurality of data-streams, wherein the plurality of data-streams are received from a plurality of distributed devices, wherein the plurality of distributed devices are connected via a network, wherein each one of the plurality of data-streams comprise data-elements;

store the data-elements of the received plurality of data-streams, wherein the stored data-elements are stored in a plurality of block-stores, wherein the block-stores are stored on the plurality of distributed devices;

assign block-identifiers to data-blocks of the block-stores, wherein the block-identifiers are globally unique, wherein each of the block-stores comprises one or more data-blocks, wherein the data-blocks of a single data-stream are logically grouped, and wherein each of the data-blocks are read and written as a single unit;

allocate the stored data-elements to the data-blocks of the block-stores, wherein each data-block comprises the stored data-elements of only one of the received plurality of data-streams;

further allocate the block-allocated data-elements to events of the data-blocks, wherein each of the data-blocks comprise one or more events, wherein each of the events comprises the block-allocated data-elements of the corresponding data-block; and

,

generate indices for the event-allocated data-elements.