Improve topN algorithm

[leventov]

Description

@leerho proposed how the topN algorithm can be improved here: #7161 (comment)

Motivation

Avoid potentially significant errors, see #1134 and #1135.

[leerho]

Objective

• Improve the current internal heuristic algorithm for "TopN" with streaming algorithms including DataSketches Frequent Items Sketch (FIS).

Motivation

• Resources: The current implementation of "TopN" is likely using more compute resources than is necessary. This can be addressed by the use of several algorithms, of which the FIS is one.

Where the FIS can be applied:

• Error vs. Size: The FIS has user configureable parameter that allows tradeoff of the sketch size and its resulting accuracy (or what we call its error threshold, or epsilon).
• Error Clarity: The FIS has well-defined error bounds that can be applied to each item in the result set.
• Query Flexibility: At query time the user can choose two different query modes NO_FALSE_POSITIVES (more strict) or NO_FALSE_NEGATIVES (less strict). The FIS will not return items below the noise floor. So you know that the items it does return are, in fact, significant.
• Streaming: The FIS is a streaming algorithm, just pass the input data into the sketch once. No need to presort the input stream. This will save processing cycles and time.
• Mergeable: The FIS is mergeable. Each node would sketch its own data independently, then the sketches from each node are passed to a central "broker" node where they are merged into a final sketch, which can then be queried. This could reduce the amount of data passing between the nodes and the broker, but it will depend on how the sketch is configured. There is no loss of accuracy due to the merging process.
• Data Independent: The FIS error properties are independent of how the input data is partitioned into separate streams (the nodes) and how the values and their weights are distributed within the streams themselves. The FIS can be somewhat input data order sensitive, in that the weights and order of the items in the result set may vary slightly from run to run, but they will always be within the error specification of the chosen sketch configuration.

Making it Happen!

We on the sketches team are not knowledgeable about Druid internals. We certainly can advise and consult on the proper use of the sketches, but it must be someone on the Druid team to integrate the sketch library into Druid.

[gianm]

It sounds like the behavior of FIS is different enough from topN that 'replace' is not in the cards. But it does sound like something that would be really nice to expose in Druid somehow. Maybe as a new query type? Can the 'weight' of an item be any Comparable object? (This would allow it to be compatible with any Druid aggregator.) Or is it limited to ints?

[leerho]

@gianm
I think we may have some confusion here. FIS uses Identity for the object and Comparable for the weight parameter. And the weights must be numeric positive longs. (The Quantiles sketch depends on Comparable).

Imagine you are Apple and have a stream of song title objects, each with a price as purchased. Different instances of the same title may have a different price due to store discounts or whatever.

The task is to identify the largest revenue song titles, by title and total revenue.

The input stream is a stream of {Object songTitle, long purchasePrice} (Price in cents)
You create the sketch:

ItemsSketch<SongTitle> fisSketch = new ItemSketch<SongTitle>(1024);

Note: the sketch size is configurable, depending on how accurate you want it to be.

You update the sketch with

while (remainingItems > 0) {
  SongTitle songTitle = next();
  fisSketch.update(nextItem, (long)(songTitle.getPrice()));
} 

Note that the sketch does the item aggregation for you, and no need to sort the input.
The purchasePrice is the "weight". When duplicate songTitles appear in the stream, their purchasePrice is accumulated in the sketch.

When all the sketches at each of the nodes are complete, you send the sketches to the Broker where they are all merged together.

The final merged sketch is then queried:

ItemSketch.Row<T>[] rows = getFrequentItems(ErrorType.NO_FALSE_POSITIVES);

Each row of the resulting array of rows contains:

• A reference to the frequent Item
• An Estimate of the total weight (total revenue) for that item
• The upper bound of what the true total Weight is
• The lower bound of what the true total Weight is.
• plus a few other convenience methods.

The array can be sorted by the accumulated weights, so they are in approximate rank order. The sketch will return all items that have a weight above the threshold.

This is effectively a superset of the typical "TopN" heuristic as it does a lot more than the current heuristic, and it provides accuracy guarantees that the existing heuristic cannot.

Fundamentally the Druid "TopN" heuristic is data sensitive and it can fail without warning where this will not.

So I am not clear where you think this would not replace the current functionality? What is the functionality that you think is missing?

It should certainly be worth a try :)

Lee.

[gianm]

The current topN algorithm can be used for any query shaped like SELECT x, AGG(y) FROM tbl ORDER BY AGG(y) DESC LIMIT N when useApproximateTopN is enabled (when disabled, groupBy is used instead, which is exact).

It sounds like FIS is limited in a couple ways that prevents it from being used for all queries of that form:

• "AGG" must be "SUM" and y must all be positive. (topN will accept any aggregation function and any y)
• "LIMIT" will not necessarily be adhered to. (topN will always return LIMIT results, or the number of distinct values of x, whichever is lower)

Now, you might argue that this level of flexibility is bad, because it involves throwing caution to the wind, and sometimes falls flat on its face. And you would have a good point. But, nevertheless, the flexibility is there, and is even useful sometimes.

Maybe it makes sense for the FIS to be exposed via its own native query type ("frequentItems" instead of "topN"). If we do that, I'm wondering how it makes sense to expose it through SQL. Maybe a context flag like useFrequentItemsForTopN that would plan a SQL query like SELECT x, AGG(y) FROM tbl ORDER BY AGG(y) DESC LIMIT N as a "frequentItems" query if "AGG" is "SUM", and would be documented to relax "LIMIT" such that fewer than N items might be returned. I'm not sure how to satisfy the "y must all be positive" requirement; maybe we need to track metadata on the brokers about min/max values for columns, and only use the frequentItems query if y is known to all be positive.

[dylwylie]

I'm not sure how widely it's used but TopN's also supports ascending order. Curious if the sketch supports returning the inverse "most infrequent" items too?

[dylwylie]

Also curious on people's thoughts on user's being explicit in their queries that can return approximate results versus optimising queries based on context/configuration.

SELECT AGG(y), x FROM tbl where x in (SELECT FREQ_I_SKETCH(x, y, 100) from tbl)
versus trying to optimise a order by limit.

[gianm]

Also curious on people's thoughts on user's being explicit in their queries that can return approximate results versus optimising queries based on context/configuration.

That's probably better, given that FIS is doing a very particular thing, and SQL users might appreciate being able to trigger its activation explicitly.

[leerho]

This is very helpful and I think I see where we are talking past each other. There appear to be several different use cases for the concept of "TopN". The starting point for these use cases is after the selection and other possible SQL manipulation, where you end up with a set of distributed lists of items where the task is to efficiently merge together these lists and obtain the "TopN" of these items. It is the next operations that follow that distinguish the several use cases.

• Case 1: The items are Comparable with respect to some property y, which does not have to be numeric. It could be a string. If y is numeric, it could be either positive or negative, it doesn't matter. The other significant property is that these items would not be aggregated together, e.g. it would not be meaningful to do an operation like ItemC = ItemA + ItemB. To finish, the following operations would be performed: (Assume N = 100). Note that the "order" here could be either direction so that one could also obtain the "BottomN".
  -- Order the items in each node .
  -- Select top 100 from the list in each node (Note: we don't need more than N from each node)
  -- Merge the lists together in the Broker
  -- Order the items in the Broker, choose the top 100.

• Case 2: The items have the Identity property such that one can check if ItemA == ItemB, and they have a weight (or importance) property that is independent of the identity property. In other words, ItemA == ItemB and possibly ItemA.w1 != ItemB.w2. This weight is numeric, positive and obviously comparable. These objects can be "merged" together when the identity is satisfied as in ItemC.w3 = ItemA.w1 + ItemB.w2 where w3 = w1 + w2. (Assume finalN = 100). The current algorithm would be the following
  -- Order the items in each node .
  -- Select top (nodeN=1000 from the list in each node (Note: we must select nodeN >> finalN)
  -- Merge the lists together in the Broker
  -- Order the items in the Broker, merging identical items, choose the top 100.
  The sketch equivalent of this process would be
  -- Feed the items in each node into a FIS.
  -- Each node sends its sketch to the Broker.
  -- The Broker merges the sketches together into a single sketch
  -- The result "Most frequent Items" or "Heaviest Items" are returned from the sketch.

• Case 3: This is the same as Case 2, except that negative weight values would be allowed. This, unfortunately cannot be solved with our current sketch anyway. We will have to think about this.

I think this addresses all the use cases (this was kinda slapped together). But for both cases 1 and 2, improvements can be made in how the final steps are performed. For case 3 I would like to ask: "Is this a real case and how often does it occur?

[leerho]

I should point out that Case 1 above is also streamable and solvable with significantly less resources than you are using now. And I think we have a solution for Case 3 also, but that may require a bit more effort.

[peferron]

@leerho Going back to your song example, could FIS support getting the top song titles by number of unique listeners over the past week? The current topN can do that using HLL sketches as metric (weight). That's one example of the usefulness of accepting any Druid aggregation or post-aggregation as a metric. (Interestingly, this specific case could benefit from using a HLL Map sketch.)

Also, I don't think that the current topN algorithm orders all distinct items on each node (or segment) prior to selecting the top 1000, since that would be unnecessarily wasteful when the number of distinct items is >> 1000. Would be great if someone more familiar with the topN implementation could confirm.

[leerho]

getting the top song titles by number of unique listeners over the past week

@peferron Now that is very interesting! The current FIS doesn't handle this case, but I think it could be extended to do that, or to have a configurable aggregate function. I'll have to think about it.

I agree with you that having the ability to specify any Druid aggregation or post-aggregation as a metric is pretty powerful and I'm not trying to replace that functionality.

My principal concern is the way the final steps are performed to obtain the Top 100(for example) items:

• for each node sort all items in the node // cost: n log n //where n = total items in the node
• pass top 1000 items to the Broker //network cost
• The Broker sorts (numNodes(M) * 1000 items): cost: M * 1000 log (M*1000)
• The Broker chooses the top 100.

This heuristic approach is error prone, data sensitive and compute-wise costly. This, I think, can be improved, but not by substituting FIS!

I am not familiar with how the pre- and post-aggregation steps intertwine themselves with this process. But perhaps this heuristic could be accomplished with a Min-Heap? It would be faster, smaller, not data sensitive, and generally more robust. It would change the compute cost from a n log n to a n log k, which could be significant.

This (I realize now) has nothing to do with FIS. As I tried to state above, that Druid's concept of "TopN" consists of several very different types of operations. And the current FIS would only help in a subset of them.)

But where FIS could apply (many duplicates (with weights)) in the stream and you wish to return the aggregated heaviest weighted items (or the most frequent items). Now the the compute cost reduces to just n and makes it suitable for real-time queries. This type of query is fairly common and it might deserve a special query type of its own.

The idea of a compound FrequentItem<Item, Aggregator> sketch sounds very interesting, but we don't have it now, so it would have to be developed.

[leerho]

@peferron
By the way, we do have an HLL Map sketch, but it was designed with a different use-case in mind so doesn't quite fit what we need here.