Problem: top k Views / leaderboard Main API: getTopK(startTime, endTime, k) If we only needed 1 leaderboard for all time (it'd be easy ): 1. every event goes in a log-based msg broker 2. We pull this data in a stream processing framework - partition this on the ID of the event 3. Each node has a k-sized minHeap. 4. At intervals, we just batch & merge all these different nodes together But doing this batch computation - we lose flexibility of real-time data. Naive solution: Use time-series DB. But here too, we would have to do all the work in the read path - and this is what we want to avoid. We can relax one requirement - approximate count vs fixed count> The perfect way to do it would mean having all COUNTS data on 1 Server - impossible to keep it in memory though. (11:50) This is where Count-min-sketch comes in 15:50 - Diagram of lamda architecture
One suggestion that you can use some drawing tools to draw the diagrams and explain it with your cursor at the time you are speaking. It is easier for newbies to follow what you are trying to explain.
We can now use a more accurate probabilistic data structure TopK heavy keeper inbuilt in Redis. For each user activity message determine the time bucket(fixed window) from the timestamp. The time bucket will be the key and value will be the TOPK data structure. We can keep adding the element in the topk data structure using redis inbuilt command. Also, if we want to increment the count of the element by some random value i.e score redis topk provides it out of the box. Additionally you can set the ttl for the key greater than the time window used for determining the time bucket
First of all thanks for the video. In real life, I will add a lot of nodes with time series database (or maybe one very big) and do all these calculations on db. Such databases use a lot of optimisation (and also probabilistic data structures))) and give a lot of others features.
Going through all these videos, and WOW, I'm impressed! For this video, what does "event" mean to you? My understanding is that we are keeping track of various leaderboards, each with their own "event" metric (score, clicks, views, etc.)
Hey Jordan, I have a systems design interview coming up soon and your videos have been a massive help. I just wanted to get your thoughts on how you decide which database to choose for a particular feature. Naturally structured data / data requiring ACID properties would be a relational database like MySQL. But how do you pick between something like MongoDB and Cassandra say? Cassandra seems to have a ton of advantages, is the only real use case for MongoDB when your data has a really wide range of fields and hence document storage is better than wide column?
Yeah I think chosing mongo for a systems design interview should be relatively rare, and you really only oughtta do it when you would prefer having no schema but also want single leader replication, unlike Cassandra which uses leaderless replication. And the only reason you may want single leader replication is to basically avoid conflicts that occur in Cassandra which are solved by last write wins, which means some writes are overwritten
I worry how should we build top k. count min sketch allows to get a counter for the current value in the stream, but for all previous values we have only a hash and a counter. ok, if we try to keep both value and counter for top K (or maybe double K for a more precise merge later on), then we need to have: - find and increase counter by 1 when we meet a heavy hitter (hash map will do the job) - find the smallest counter in top K heap (priority queue will do the job) the problem is to make +1 and update a value in the heap. it doesn’t allow any search. ok, we can just keep track of the smallest seen value in top K. but what if a new value have to be added to top K? then we need to drop the previous smallest value and find a new smallest value. I cannot see a good way for it. My best solution I can come up with: - count min sketch for counters on all values - bi-linked list sorted by counter containing pairs - hash map: value -> pointer to linked list element for top K when a new value comes: - update count min sketch and get a counter - if the value is in the hash map, get the pointer to the bi-linked list elem, update counter there. move the bi-linked list elem up the list to keep it sorted - if the value is not in top k hash map, compare it with the last element of the bi-linked list. If the new counter is bigger, get the value from the last element and remove it from hash map. drop the last element. Add a new element to the bi-linked list, add a value with the pointer to the new element to the hash map
Unrelated - Do you have a video in your series about "Sharded counters"? They are very high write counters (like for viral RUclips videos) which are impossible to manage with a regular database because of high concurrency on the same row and column. So the solution is to make copies of the same counter for the same document ID and accumulate the counts from all counters as the actual count. It gets challenging to do this dynamically when a video unexpectedly goes viral or when the view increase starts to waver.
In your final diagram, you saw count-min-sketch map is for short time windows but the one below it (using regular hash maps) is for larger time windows. It should be the opposite, right? The point of using a count-min-sketch map is to conserve space. If you didn't need to, we can just use the more accurate plain hash maps.
Yeah I think the assumption is that even in the span of a minute you're still using too much memory (so on a typically server which would be holding the memory this is infeasible), but then if latency of results isn't the issue we can perform these calculations on disk e.g. a Flink consumer.
Thanks for the awesome videos! 2 naive questions from me: 1. me not 100% understand that why you need count-min-sketch, it could do the work, but in my understanding, count-min is to use resolve hash collision, could we just use max heap, can dump every 0::00::00, or can we just use map-reduce for 1m count? 2. how do you handle delayed events?
It's to limit memory usage. Delayed events wouldn't be picked up in the approximate case, but depending on how long you wait to perform your exact solution you could maybe pick them up
The point of count min sketch is to be a memory optimization. Could we use Flink? Definitely - but then we have to store a crap ton of events (each partition needs to keep its own top 10 locally on its range of the keys) and then eventually we'd need to merge those together. If we don't want to store as much data, or we don't want to have to aggregate, count min sketch allows us to buffer everything that we need on one node and approximate.
One more question: In the final count min sketch solution, how is it possible that one server is processing all events while in the first solution we had to partition the events across nodes because of load? Thanks
The operations for count min sketch are not computationally intensive because there's a limited amount of memory being used. I think I misspoke originally: you can't count all the operations on one server, not necessarily due to CPU limitations, but moreso due to storage limitations in memory. On the other hand, the amount of memory used in count min sketch is bounded.
In the fast solution, the count-min sketch servers would still need to keep track of all the eventIds they have seen in the time-window and then every minute get the top k and write it to a cache or something. If that's the case, then what do we save by using a count-min sketch, we can just keep a hashtable to store freq map and dump it every 1 min or basically a small time window? Maybe I am not understanding this correctly, can you tell me what the count-min server will actually write to a database?
The count min sketch server is just for ad hoc querying where you want a recent leaderboard, and you can reset it once per day for example (remember this is just an approximation it will be wrong - for an exact answer you can ultimately query something like a TSDB and get the answer
@@jordanhasnolife5163 I think what they were asking is, count-min-sketch is a probabilistic data structure with collisions, so once you build it, while you have accumulated approx frequencies, you've lost the actual event IDs - you can't reverse hash and get the actual event IDs from the count-min-sketch map. So if you've going to have to keep a mapping to the actual event IDs anyway (which can get very large if the time window is large enough) what are you really saving? I think the solution would be not keep all the event IDs for the time window, but just the top K as per a min heap. At any point, you have a min heap of top K event IDs. When you're dealing with the current event ID, you hash it and increase the frequency. But as soon as that's done, see if the top of the min heap has a lower frequency than this, and if so, evict it and put the new one instead. (You might also have to keep a hash set to keep track of whether the current event ID is already inside the heap somewhere. ) With this approach, you end up keeping at most k top event IDs in addition to the fixed size count-min-sketch map.
@@OKJazzBro You can loop through the list of eventIds (which are kept on a database) and figure out the relative frequencies that way, I don't believe you need a reverse mapping - these are good points though. I'll be remaking all these videos eventually for good reason and will put more thought into detailed explanations.
@@jordanhasnolife5163 Looping through the list of eventIDs kinda defeats the purpose of creating the count-min-sketch because the list of eventIDs can be large.
Hi @divijdude , hope i am not late to the party. We only need to keep track k items, so min-heap with k elements and replace head of value if count of current processed eventId is greater. So we are not keeping huge hashmap to track count because of this algo. I hope this clarify your doubt.
I don't understand how the final solution with the slow and fast path work over a long time interval. Say if somebody wants the leaderboard for (now - 3h) to (now - 1h) mins, how will we do it? If the slow path aggregates results every hour, then we can merge the 2 windows but that's not going to be accurate, right? For eg. say k = 2 and the first hour has A : 11, B : 10 but the first hour had C (not in the top 2) with freq 8 and the second hour has A: 6, D: 5 and C with freq: 3 Now if we merge these two, our system will declare A, B as the top 2 but actually it should have been A, C because C has a freq of 11 while B only has 10?
That's correct - sometimes we'll be wrong, it's an approximation. In the slow path, we can actually compute the exact answer if we needed it (e.g we assemble all the results for one hour windows and combine them)
@@jordanhasnolife5163 In the slow path, aren't we just storing the top k? If that's all we keep track of then we can't assemble right? If we want to keep the freq of each event for all 1 hour window, it won't be scalable at all, right? Is there a way for us to be completely precise w.r.t assembling the slow part windows ?
why is 1 billion updates per day too much for a server? Can you give an average figure that can be handled by a CPU core or something similar just to get an idea of how big 1 billion is for a server? Thanks
Truthfully 1 billion operations may not even be, but consider every Google search in the world - there's no way those could all be run on 1 server, there are too many per second and the server wouldn't be able to support that type of load.
- apache logs -> map/reduce -> topN
- redis (top-N module)
- Flink for real-time
Problem: top k Views / leaderboard
Main API: getTopK(startTime, endTime, k)
If we only needed 1 leaderboard for all time (it'd be easy ):
1. every event goes in a log-based msg broker
2. We pull this data in a stream processing framework - partition this on the ID of the event
3. Each node has a k-sized minHeap.
4. At intervals, we just batch & merge all these different nodes together
But doing this batch computation - we lose flexibility of real-time data.
Naive solution: Use time-series DB.
But here too, we would have to do all the work in the read path - and this is what we want to avoid.
We can relax one requirement - approximate count vs fixed count>
The perfect way to do it would mean having all COUNTS data on 1 Server - impossible to keep it in memory though.
(11:50) This is where Count-min-sketch comes in
15:50 - Diagram of lamda architecture
This legend be uploading high quality content faster than I can watch it
That's like my relationship with Lana Rhodes
One suggestion that you can use some drawing tools to draw the diagrams and explain it with your cursor at the time you are speaking. It is easier for newbies to follow what you are trying to explain.
We can now use a more accurate probabilistic data structure TopK heavy keeper inbuilt in Redis. For each user activity message determine the time bucket(fixed window) from the timestamp. The time bucket will be the key and value will be the TOPK data structure. We can keep adding the element in the topk data structure using redis inbuilt command. Also, if we want to increment the count of the element by some random value i.e score redis topk provides it out of the box.
Additionally you can set the ttl for the key greater than the time window used for determining the time bucket
Interesting, I'll have to look into how that data structure works!
First of all thanks for the video. In real life, I will add a lot of nodes with time series database (or maybe one very big) and do all these calculations on db. Such databases use a lot of optimisation (and also probabilistic data structures))) and give a lot of others features.
Nice!
I liked the video before watching it, as I already know it's quality
Going through all these videos, and WOW, I'm impressed! For this video, what does "event" mean to you? My understanding is that we are keeping track of various leaderboards, each with their own "event" metric (score, clicks, views, etc.)
Yep that's correct! Thanks for the kind words :)
Hey man, love your videos. Thank you for doing this
Glad you enjoy :)
Hey Jordan, I have a systems design interview coming up soon and your videos have been a massive help. I just wanted to get your thoughts on how you decide which database to choose for a particular feature. Naturally structured data / data requiring ACID properties would be a relational database like MySQL. But how do you pick between something like MongoDB and Cassandra say? Cassandra seems to have a ton of advantages, is the only real use case for MongoDB when your data has a really wide range of fields and hence document storage is better than wide column?
Yeah I think chosing mongo for a systems design interview should be relatively rare, and you really only oughtta do it when you would prefer having no schema but also want single leader replication, unlike Cassandra which uses leaderless replication. And the only reason you may want single leader replication is to basically avoid conflicts that occur in Cassandra which are solved by last write wins, which means some writes are overwritten
@@jordanhasnolife5163 Makes sense, thanks so much! Keep up the great work on the channel, love the humour and I'm glad it's blowing up :D
@@jordanhasnolife5163 And MongoDB use B+ tree, which makes reads faster.
I worry how should we build top k. count min sketch allows to get a counter for the current value in the stream, but for all previous values we have only a hash and a counter.
ok, if we try to keep both value and counter for top K (or maybe double K for a more precise merge later on), then we need to have:
- find and increase counter by 1 when we meet a heavy hitter (hash map will do the job)
- find the smallest counter in top K heap (priority queue will do the job)
the problem is to make +1 and update a value in the heap. it doesn’t allow any search.
ok, we can just keep track of the smallest seen value in top K. but what if a new value have to be added to top K? then we need to drop the previous smallest value and find a new smallest value. I cannot see a good way for it.
My best solution I can come up with:
- count min sketch for counters on all values
- bi-linked list sorted by counter containing pairs
- hash map: value -> pointer to linked list element for top K
when a new value comes:
- update count min sketch and get a counter
- if the value is in the hash map, get the pointer to the bi-linked list elem, update counter there. move the bi-linked list elem up the list to keep it sorted
- if the value is not in top k hash map, compare it with the last element of the bi-linked list. If the new counter is bigger, get the value from the last element and remove it from hash map. drop the last element. Add a new element to the bi-linked list, add a value with the pointer to the new element to the hash map
Unrelated - Do you have a video in your series about "Sharded counters"? They are very high write counters (like for viral RUclips videos) which are impossible to manage with a regular database because of high concurrency on the same row and column. So the solution is to make copies of the same counter for the same document ID and accumulate the counts from all counters as the actual count. It gets challenging to do this dynamically when a video unexpectedly goes viral or when the view increase starts to waver.
yes this is a grow only CRDT :). I have a CRDT video on my channel if you look for it
In your final diagram, you saw count-min-sketch map is for short time windows but the one below it (using regular hash maps) is for larger time windows. It should be the opposite, right? The point of using a count-min-sketch map is to conserve space. If you didn't need to, we can just use the more accurate plain hash maps.
Yeah I think the assumption is that even in the span of a minute you're still using too much memory (so on a typically server which would be holding the memory this is infeasible), but then if latency of results isn't the issue we can perform these calculations on disk e.g. a Flink consumer.
Thanks for the awesome videos!
2 naive questions from me:
1. me not 100% understand that why you need count-min-sketch, it could do the work, but in my understanding, count-min is to use resolve hash collision, could we just use max heap, can dump every 0::00::00, or can we just use map-reduce for 1m count?
2. how do you handle delayed events?
It's to limit memory usage. Delayed events wouldn't be picked up in the approximate case, but depending on how long you wait to perform your exact solution you could maybe pick them up
Why not use Flink for real-time event processing? How does it compare to the count-min sketch?
Thinking the same
The point of count min sketch is to be a memory optimization. Could we use Flink? Definitely - but then we have to store a crap ton of events (each partition needs to keep its own top 10 locally on its range of the keys) and then eventually we'd need to merge those together. If we don't want to store as much data, or we don't want to have to aggregate, count min sketch allows us to buffer everything that we need on one node and approximate.
One more question: In the final count min sketch solution, how is it possible that one server is processing all events while in the first solution we had to partition the events across nodes because of load? Thanks
The operations for count min sketch are not computationally intensive because there's a limited amount of memory being used. I think I misspoke originally: you can't count all the operations on one server, not necessarily due to CPU limitations, but moreso due to storage limitations in memory. On the other hand, the amount of memory used in count min sketch is bounded.
Right on time!
If there are some drawing diagrams in the first 15 min, that would be better.
Yup, I'm remaking all of these videos. Give me about 2-3 months and then they'll all be done haha
Cool bro!@@jordanhasnolife5163
17:10 wouldnt the Kafka Streams API work here? so as to keep it homogenous within the Kafka ecosystem
Don't see why not
In the fast solution, the count-min sketch servers would still need to keep track of all the eventIds they have seen in the time-window and then every minute get the top k and write it to a cache or something. If that's the case, then what do we save by using a count-min sketch, we can just keep a hashtable to store freq map and dump it every 1 min or basically a small time window? Maybe I am not understanding this correctly, can you tell me what the count-min server will actually write to a database?
The count min sketch server is just for ad hoc querying where you want a recent leaderboard, and you can reset it once per day for example (remember this is just an approximation it will be wrong - for an exact answer you can ultimately query something like a TSDB and get the answer
@@jordanhasnolife5163 I think what they were asking is, count-min-sketch is a probabilistic data structure with collisions, so once you build it, while you have accumulated approx frequencies, you've lost the actual event IDs - you can't reverse hash and get the actual event IDs from the count-min-sketch map. So if you've going to have to keep a mapping to the actual event IDs anyway (which can get very large if the time window is large enough) what are you really saving?
I think the solution would be not keep all the event IDs for the time window, but just the top K as per a min heap. At any point, you have a min heap of top K event IDs. When you're dealing with the current event ID, you hash it and increase the frequency. But as soon as that's done, see if the top of the min heap has a lower frequency than this, and if so, evict it and put the new one instead. (You might also have to keep a hash set to keep track of whether the current event ID is already inside the heap somewhere. )
With this approach, you end up keeping at most k top event IDs in addition to the fixed size count-min-sketch map.
@@OKJazzBro You can loop through the list of eventIds (which are kept on a database) and figure out the relative frequencies that way, I don't believe you need a reverse mapping - these are good points though. I'll be remaking all these videos eventually for good reason and will put more thought into detailed explanations.
@@jordanhasnolife5163 Looping through the list of eventIDs kinda defeats the purpose of creating the count-min-sketch because the list of eventIDs can be large.
Hi @divijdude , hope i am not late to the party. We only need to keep track k items, so min-heap with k elements and replace head of value if count of current processed eventId is greater. So we are not keeping huge hashmap to track count because of this algo. I hope this clarify your doubt.
I'm a 3% lady. huzzah
Haha welcome to the channel! All jokes aside glad you're here :)
how is batch job + cout-min-sketch different from redis sorted set? if we only use redis sorted set isnt that enough?
If you use a sorted set you need to have linear memory usage relative to number of events
@@jordanhasnolife5163 is a sorted set not just event_id: count/score ?
I don't understand how the final solution with the slow and fast path work over a long time interval. Say if somebody wants the leaderboard for (now - 3h) to (now - 1h) mins, how will we do it? If the slow path aggregates results every hour, then we can merge the 2 windows but that's not going to be accurate, right?
For eg. say k = 2 and the first hour has A : 11, B : 10 but the first hour had C (not in the top 2) with freq 8
and the second hour has A: 6, D: 5 and C with freq: 3
Now if we merge these two, our system will declare A, B as the top 2 but actually it should have been A, C because C has a freq of 11 while B only has 10?
That's correct - sometimes we'll be wrong, it's an approximation. In the slow path, we can actually compute the exact answer if we needed it (e.g we assemble all the results for one hour windows and combine them)
@@jordanhasnolife5163 In the slow path, aren't we just storing the top k? If that's all we keep track of then we can't assemble right? If we want to keep the freq of each event for all 1 hour window, it won't be scalable at all, right? Is there a way for us to be completely precise w.r.t assembling the slow part windows ?
why is 1 billion updates per day too much for a server? Can you give an average figure that can be handled by a CPU core or something similar just to get an idea of how big 1 billion is for a server? Thanks
Truthfully 1 billion operations may not even be, but consider every Google search in the world - there's no way those could all be run on 1 server, there are too many per second and the server wouldn't be able to support that type of load.
6:20 --- You can't calculate top K on each node then aggregate to get the top K globally. That will not give accurate results at all.
I disagree. If all the events are partitioned such that they always go to the same node you can get the top k from all of the local top k's.
@@jordanhasnolife5163 Ah yes, I totally missed that from your solution.
how about OpenSearch and rollup jobs?
I'll have to look into these
Thanks
1st one..😀
ur quality is so horrible that ur quantity (respect 🙏) is getting wasted
😢
why are you always look high af?
What do you mean "look"
Improve your voice recording