Cache Based Efficient Access Scheduling for Super Scaled Stream Processing Systems

• US 20170242889A1
• Filed: 02/24/2016
• Published: 08/24/2017
• Est. Priority Date: 02/24/2016
• Status: Active Application

• Current Assignee: Salesforce.com, Inc.
• Original Assignee: Salesforce.com, Inc.
• Inventors: ZHAO, Yingwu
• Granted Patent: US 10,262,032 B2
• Time in Patent Office: Days
• CPC Class Codes: G06F 16/24537 of operators
• CPC Class Codes: G06F 16/24539 using cached or materialised query results
• CPC Class Codes: G06F 16/24554 Unary operations; Data partitioning operations
• CPC Class Codes: G06F 16/24568 Data stream processing; Continuous queries
• CPC Class Codes: G06F 16/283 Multi-dimensional databases or data warehouses, e.g. MOLAP or ROLAP
• CPC Class Codes: G06F 16/951 Indexing; Web crawling techniques

Abstract

The technology disclosed relates to discovering a previously unknown attribute of stream processing systems according to which client offsets or client subscription queries for a streaming data store rapidly converge to a dynamic tip of a data stream that includes the most recent messages or events. In particular, it relates to grouping clients into bins to reduce a number of queries to the streaming data store by several orders of magnitude when servicing tens, hundreds, thousands or millions of clients. The bin count is further reduced by coalescing bins that have overlapping offsets. It also relates to establishing separate caches only for the current tips of data streams and serving the bins from the caches instead of the backend data store using group queries. Further, the caches are periodically updated to include the most recent messages or events appended to the dynamic tips of the data streams.

Claims

1. A method of servicing a plurality of client bins making multi-dimensional queries against a streaming data store, the method including:

receiving a multitude of queries from a plurality of client bins, the queries including a topic-partition pair and start offsets that reference events in an unbounded event stream, the start offsets mixing historical offsets and current offsets;

establishing separate caches for active, unique topic-partition pairs;

periodically determining query increments that request data not already present in the separate caches and grouping the query increments across topic-partition pairs into a query group;

submitting the query group to a streaming data store that provides access to one or more unbounded data streams;

receiving stream data responsive to the query group, including larger data blocks responsive to historical offsets and smaller data blocks responsive to current offsets, and updating the separate caches for the topic-partition pairs; and

simultaneously servicing the multitude of queries from the separate caches.

2. The method of claim 1, further including:

periodically expiring part or all of the separate caches, at least when parts of data in a particular separate cache reach an expiry date-time and when the particular separate cache reaches an inactivity threshold.

3. The method of claim 1, wherein a plurality of the queries include start offsets that reference current offsets of the unbounded data stream comprising a tip of most recent events.

4. The method of claim 3, wherein the data requested by the query increments include most recent events from the tip.

5. The method of claim 1, wherein the current offsets maintained by the separate caches for respective topic-partition pairs have varying start and end offsets.

6. The method of claim 1, wherein the query increments are determined at 300 m/s intervals.

7. The method of claim 1, wherein the query increments are determined at 200 m/s intervals.

8. The method of claim 1, wherein the historical offsets of the unbounded data stream are periodically persisted in a stable storage.

9. The method of claim 1, wherein the client bins include millions of clients.

10. The method of claim 1, further including:

detecting one or more cache misses caused by a multitude of queries that include inactive topic-partition pairs for which there are no established caches;

handling the cache misses by;

grouping the multitude of queries across the inactive topic-partition pairs into a query group;

submitting the query group to a streaming data store that provides access to one or more unbounded data streams;

receiving stream data responsive to the query group, including large data blocks; and

simultaneously servicing the multitude of queries using the large data blocks.

11. The method of claim 10, further including:

establishing new separate caches for each of the inactive topic-partition pairs, thereby preventing subsequent cache misses.

12. The method of claim 1, further including:

detecting one or more cache misses caused by a multitude of queries that reference expired start offsets for active, unique topic-partition pairs, wherein the expired start offsets are below respective start offsets of current offsets maintained by separate caches established for respective active, unique topic-partition pairs;

handling the cache misses by;

grouping the multitude of queries into partition queries based on respective active, unique topic-partition pairs;

submitting the partition queries to a streaming data store that provides access to one or more unbounded data streams;

receiving stream data responsive to the partition queries, including large data blocks; and

simultaneously servicing the multitude of queries using the large data blocks.

13. The method of claim 12, further including:

using a thread pool to concurrently process the partition queries.

14. The method of claim 12, further including:

for a particular partition query, identifying start offsets of clients bins that issued queries grouped in the particular partition query;

receiving stream data responsive to the particular partition query in one or more data batches determined based on the start offsets and a preconfigured batch size; and

simultaneously servicing the queries using the data batches.

15. The method of claim 12, wherein each of the partition queries identify a unique topic-partition pair and a set of start offsets of clients bins that issued queries grouped in respective partition queries.

16. The method of claim 1, wherein a cache size of each separate cache is preconfigured based at least on number of bytes and number of events.

17. A non-transitory computer readable storage medium impressed with computer program instructions to service a plurality of clients from an unbounded data stream, while allowing each client to select any available service offset in the data stream as a starting offset for streaming service to the client, the instructions, when executed on a processor, implement a method comprising:

receiving a multitude of queries from a plurality of client bins, the queries including a topic-partition pair and start offsets that reference events in an unbounded event stream, the start offsets mixing historical offsets and current offsets;

establishing separate caches for active, unique topic-partition pairs;

periodically determining query increments that request data not already present in the separate caches and grouping the query increments across topic-partition pairs into a query group;

submitting the query group to a streaming data store that provides access to one or more unbounded data streams;

receiving stream data responsive to the query group, including larger data blocks responsive to historical offsets and smaller data blocks responsive to current offsets, and updating the separate caches for the topic-partition pairs; and

simultaneously servicing the multitude of queries from the separate caches.

18. The non-transitory computer readable storage medium of claim 17, implementing the method comprising:

detecting one or more cache misses caused by a multitude of queries that include inactive topic-partition pairs for which there are no established caches;

handling the cache misses by;

grouping the multitude of queries across the inactive topic-partition pairs into a query group;

submitting the query group to a streaming data store that provides access to one or more unbounded data streams;

receiving stream data responsive to the query group, including large data blocks; and

simultaneously servicing the multitude of queries using the large data blocks.

19. The non-transitory computer readable storage medium of claim 17, implementing the method comprising:

detecting one or more cache misses caused by a multitude of queries that reference expired start offsets for active, unique topic-partition pairs, wherein the expired start offsets are below respective start offsets of current offsets maintained by separate caches established for respective active, unique topic-partition pairs;

handling the cache misses by;

grouping the multitude of queries into partition queries based on respective active, unique topic-partition pairs;

submitting the partition queries to a streaming data store that provides access to one or more unbounded data streams;

receiving stream data responsive to the partition queries, including large data blocks; and

simultaneously servicing the multitude of queries using the large data blocks.

20. A system including one or more processors coupled to memory, the memory loaded with computer instructions to service a plurality of clients from an unbounded data stream, while allowing each client to select any available service offset in the data stream as a starting offset for streaming service to the client, the instructions, when executed on the processors, implement actions comprising:

receiving a multitude of queries from a plurality of client bins, the queries including a topic-partition pair and start offsets that reference events in an unbounded event stream, the start offsets mixing historical offsets and current offsets;

establishing separate caches for active, unique topic-partition pairs;

periodically determining query increments that request data not already present in the separate caches and grouping the query increments across topic-partition pairs into a query group;

submitting the query group to a streaming data store that provides access to one or more unbounded data streams;

receiving stream data responsive to the query group, including larger data blocks responsive to historical offsets and smaller data blocks responsive to current offsets, and updating the separate caches for the topic-partition pairs; and

simultaneously servicing the multitude of queries from the separate caches.