Add bounded unique count aggregation #781
Conversation
jbrooks-stripe
force-pushed
the
jbrooks-bounded-unique-count
branch
6 times, most recently
from
685e0b9 to
3c368e8
Compare
pengyu-hou
left a comment
pengyu-hou
left a comment
There was a problem hiding this comment.
@jbrooks-stripe the change looks good. Could you please also update the operation in the groupby.py at https://github.com/airbnb/chronon/blob/main/api/py/ai/chronon/group_by.py#L56
Could you follow the same style with the HISTOGRAM and HISTOGRAM_K? Thanks!!
pengyu-hou
force-pushed
the
jbrooks-bounded-unique-count
branch
from
3c160e0 to
d5746da
Compare
| override def irType: DataType = ListType(StringType) | ||
|
|
||
| override def merge(ir1: util.Set[String], ir2: util.Set[String]): util.Set[String] = { | ||
| ir2.asScala.foreach(v => |
There was a problem hiding this comment.
| ir2.asScala.foreach(v => | |
| ir2.iterator().asScala.foreach(v => |
There was a problem hiding this comment.
otherwise it will create intermediate collections
nikhil-zlai
left a comment
nikhil-zlai
left a comment
There was a problem hiding this comment.
thanks for putting this together. have some minor perf related comments.
| override def irType: DataType = ListType(StringType) | ||
|
|
||
| override def merge(ir1: util.Set[String], ir2: util.Set[String]): util.Set[String] = { | ||
| ir2.asScala.foreach(v => |
There was a problem hiding this comment.
otherwise it will create intermediate collections
| } | ||
|
|
||
| private def md5Hex(bytes: Array[Byte]): String = | ||
| MessageDigest.getInstance("MD5").digest(bytes).map("%02x".format(_)).mkString |
There was a problem hiding this comment.
lets say i want to unique count a bunch of user / merchant ids (long values) - won't this be less efficient than simply keeping the set of longs?
There was a problem hiding this comment.
made the code change to keep the numeric type as is
|
|
||
| override def update(ir: util.Set[String], input: T): util.Set[String] = { | ||
| if (ir.size() >= k) { | ||
| return ir |
There was a problem hiding this comment.
memory optimization: we can use a sentinel set when k is reached.
| return ir | |
| if(ir == Constants.SentinelSet || ir.size() >= k) return Constants.SentinelSet |
There was a problem hiding this comment.
Hm.. I don't think we have sentinel set yet in OSS branch.
| ir1 | ||
| } | ||
|
|
||
| override def finalize(ir: util.Set[String]): Long = ir.size() |
There was a problem hiding this comment.
| override def finalize(ir: util.Set[String]): Long = ir.size() | |
| override def finalize(ir: util.Set[String]): Long = if(ir == Constants.SentinelSet) k else ir.size() |
| if (ir == BoundedUniqueCount.SentinelSet) { | ||
| val list = new util.ArrayList[Any]() | ||
| list.add(BoundedUniqueCount.SentinelMarker) | ||
| list | ||
| } else { | ||
| new util.ArrayList[Any](ir) |
There was a problem hiding this comment.
aren't both of these same - can we just do
| if (ir == BoundedUniqueCount.SentinelSet) { | |
| val list = new util.ArrayList[Any]() | |
| list.add(BoundedUniqueCount.SentinelMarker) | |
| list | |
| } else { | |
| new util.ArrayList[Any](ir) | |
| new util.ArrayList[Any](ir) |
There was a problem hiding this comment.
Hi @nikhil-zlai , this is used to differentiate the empty Sentinel Set and the actual empty set.
nikhil-zlai
left a comment
nikhil-zlai
left a comment
There was a problem hiding this comment.
This looks great! thanks pengyu!
Summary
Adds a BOUNDED_UNIQUE_COUNT aggregation. This will allow exact unique/distinct counts, but will cap at a given value to keep memory usage constant.
Why / Goal
We have use cases where we'd prefer an exact solution instead of the approx equivalents, but want to have protections in place so that memory doesn't become an issue.
Test Plan
Checklist
Reviewers