Hi. What do you think is a single most important
component required for a successful system design interview? What should we focus on the most while preparing
for one? I bet this is knowledge. Knowledge of system design concepts and how
to combine them together. And let me prove you that by walking you through
all the stages of a typical system design interview. This is system design interview channel. And today we discuss how to count things at
a large scale. So, the interviewer asks us to design a system
that does counting. The problem may be articulated clearly, for
example we need to count views for videos on Youtube. Or likes on Instagram or Facebook. But more often than that, the problem will
be stated in a vague or a more general manner. For example, we may be asked to calculate
not a single metric, like number of views, but a set of metrics. For example we need to monitor performance
of different applications. Which means we need to count how many requests
go through the service, how many errors service produce, average response time. And of course nothing stops the interviewer
from stating the problem in a very generic form, like analyze data in real-time. What does data analysis mean? Who sends us data? Who uses results of this analysis? What real-time really means? These and many other questions need to be
clarified. And even if system design problem seems clear
to you, let me give you two big reasons why you still need to ask your interviewer questions. The first reason is really important for the
interviewer. The second reason is important for you. The interviewer wants to see how you deal
with ambiguity. Whether you can identify key pieces of the
system and define a scope of the problem. And why it is so important for the interviewer
- because she wants to understand how you will approach design problems in real life. System design interview problems are usually
open-ended. It is impossible to solve such problems within
45 or 60 minutes interview. We should be clear on what functional pieces
of the problem we are going to focus till the rest of the interview. And why requirements clarification is so important
for us, interviewees? Mainly because there may be many solutions
to the problem asked. And only when we fully understand what features
of the system we need to design, we can come up with a proper set of technologies and building
blocks. For example, let's take a problem of counting
Youtube video views. If you ask this problem a software engineer
with experience in SQL databases, he will explain you why MySQL database is a good fit. Engineer with profound experience in NoSQL
databases will explain how to count things at large scale using for example Apache Cassandra
database. We may use distributed cache to count stuff. Or apply stream processing techniques. Experts in cloud computing will solve this
problem using public cloud services only. Engineers focusing on batch processing will
solve this problem using for example Hadoop MapReduce. And views counting problem can indeed be solved
using all these different approaches. But these options are not equal. Each has its own pros and cons. We should pick those options that address
requirements. What we may want to ask about? My recommendation is to focus on the following
4 big categories: Users, where we want to understand who and
how will use the system. Scale, where we want to understand how our
system will handle a growing amount of data. Performance, where we want to know how fast
our system must be. And cost, where we need to understand budget
constraints. Let's see what specific questions in each
category we may ask. Here and below we imply views counting system
as an example. Who will use the system? Is this all Youtube viewers who will see the
total views count for a video? Or this is a per-hour statistics available
to a video owner only? Or may be this total count will be used by
some machine learning models to generate recommendations. How the system will be used? The data may be used by marketing department
only to generate some monthly reports. In other words data is retrieved not often. Or data is sent to recommendation service
in real-time. Meaning that data may be retrieved from the
system with a very high pace. Questions in this category help us understand
what data we store in the system. Questions in the Scale category should give
us a clue how much data is coming to the system and how much data is retrieved from the system. So, the questions we should ask:
How many read queries per second the system needs to process. How much data is queried per request. How many video views per second are processed
by the system. Should we deal with sudden traffic spikes
and how big they may be. The interviewer will help us define these
numbers. Or we may assume some reasonable values. Questions in the performance category should
help us quickly evaluate different design options. For example, if we can count views several
hours later than these views actually happened, both batch data processing and stream processing
design options can be considered. But if time matters and we need to count views
on the fly, or in other words in real-time, batch data processing is not an option. It is just too slow. Another good question to ask, is to clarify
how fast data must be retrieved from the system. And if interviewer tells us that response
time must be as small as possible, it's a hint that we must count views when we write
data and we should do minimal or no counting when we read data. In other words data must already be aggregated. And questions in the cost category should
help us evaluate technology stack. For example, if asked to minimize development
cost, we should be leaning towards well-regarded open-source frameworks. And if future maintenance cost is a primary
concern, we should consider public cloud services for our design. Here is a little secret you might already
know. During requirements clarification interviewer
starts to understand your level of expertise in systems design. As with coding interviews, if you do not ask
questions, this is a warning sign for the interviewer. And here is my advice: think along 4 categories
mentioned here, think about data - what data, how it gets in and gets out of the system,
and do not worry about time too much. You better spend additional 5 minutes clarifying
requirements and scope than find yourself solving a different or more complex problem
than the interviewer actually asked. The end goal of requirements clarification
discussions is to get us closer to defining both functional and non-functional requirements. When we say functional requirements, we mean
system behavior, or more specifically APIs - a set of operations the system will support. This is what our system will do. And when we say non-functional requirements,
we mean system qualities, such as fast, fault-tolerant, secure. This is basically how a system is supposed
to be. Let's define functional requirements first. And here is my practical advice. After figuring out what the system should
do, you write it down on the whiteboard. In a few sentences. For example, we write down that the system
has to count video view events. Count view events is the actual action the
system performs and video becomes the input parameter. And if we want the system to calculate not
just views, but a broader set of events, let's say likes and shares, we may generalize our
API a bit and introduce event type parameter. This parameter indicates type of the event
we process. We can go one step further and make the system
calculate not only count function, but other functions as well, like sum and average. By supporting sum function we can calculate
such metric as "total watch time" for a video. While average function can help us calculate
average view duration. And we can even further generalize the API
and say that the system will not just process events one by one, but a list of events as
a single batch. Where each event is an object that contains
information about a video, type of the event, time when event happened and so forth. Similar thought process can be applied for
data retrieval API. We first write down something like: the system
has to return video views count for a time period. Get views count becomes an action. While video identifier and start and end time
become input parameters. If we want to retrieve count not only for
video views, but for likes and dislikes for example, we can introduce event type parameter. And if we want our API return not just count
statistics, but also statistics for sum and average, we should specify function as a parameter
and name our API in a more generic way, like getStats. Following this simple technique will help
you identify names of APIs, input parameters and, if needed, make several iterations to
generalize APIs. Now, let's talk about non-functional requirements. Usually, the interviewer will not tell us
specific non-functional requirements. Most likely, she will challenge us with mentioning
big scale and high performance, as it is hard to achieve both at the same time. And we will need to find tradeoffs. In one of the previous videos I recommended
to focus on scalability, performance and availability as top priority requirements. Let's use them here. We need to design a system that should handle
tens of thousands of requests per second. We also want view count statistic to be returned
in a matter of few tens of milliseconds. And we want view count statistics to be shown
to users all the time. Even when there are network or hardware failures
in the system. Although these 3 will be our primary concern,
let's also talk about two other interesting requirements. Let's recall a CAP theorem and talk about
consistency a bit. And I would like us to touch a cost minimization
topic. And my advice is that you write down non-functional
requirements on the whiteboard as well. It will help you later while choosing among
several design options. We will talk more about this. Now, we have come to the next stage of a system
design interview: high-level architecture. And we will start with something really simple. We need a database to store data. We will have a web service that processes
incoming video view events and stores data in the database. To retrieve view counts from the database,
let's introduce another web service. Nothing scary so far, right? And it should not be. At this point we do not yet have a clear vision
of all pieces of the design. So, we just throw some very high-level components
to the whiteboard. But what do we do next? High chances the interviewer is an expert
in the field and knows the question very well. And she may start asking questions about any
component we outlined in the high-level architecture. But we may not feel comfortable discussing
any component just yet. Unless you are an expert in this field yourself,
the whole picture may not yet be clear to you. This is in our best interest as interviewees
to drive the conversation and move forward one step at a time. This is like assembling a puzzle. The interviewer can assemble the puzzle starting
from any color group. But we need to start with something simple
and construct the frame with the outside pieces first. So, what is this frame of outside pieces of
a system design puzzle? It's data. And more specifically we need to think what
data we want to store and how. Or using more professional terms we need to
define a data model. We have two options for what data we want
to store. We may store each individual video view event. Or we may calculate views on the fly and store
aggregated data. When we store individual events we capture
all attributes of the event: video identifier, timestamp, user related information such as
country, device type, operating system and so on. When we aggregate data we calculate a total
count per some time interval, let's say one minute and we lose details of each individual
event. There are pros and cons of each option. Individual events can be stored really fast. We just get the event and push it to the database. Later, when we retrieve data, we can slice
and dice data however we want. We can filter based on specific attributes,
aggregate based on some rules. And if there was a bug in some business report,
we can recalculate numbers from scratch. But there are drawbacks of this approach. We cannot read data quickly. We need to count each individual event when
total count is requested. This takes time. Another big disadvantage of this approach
- it may cost a lot of money to store all the raw events. Youtube generates billions of view events
per day. Raw events storage must be huge. On the other hand, reads become fast when
we aggregate data. We do not need to calculate each individual
event, we just retrieve total count value. Another nice property of the aggregated data,
we can use it for decision making in real-time. For example, we may send the total count value
to a recommendation service or trending service, for popular videos to be promoted to trends. Aggregated data approach has drawbacks as
well. First of all, we can only query data the way
it was aggregated. Ability to filter data or aggregate it differently
is very limited. This approach also requires us to implement
data aggregation pipeline. We need to somehow pre-aggregate data in memory
before storing it in the database. This is not an easy task and later you will
see why. Another big problem with this approach, it
is hard or even impossible to fix errors. Let's say we introduced a bug in the aggregation
logic. Then, how do we fix total counts after the
bug was fixed? So, which option should we chose? Store raw events or aggregate data in real-time? This is where we need interviewer to help
us make a decision. We should ask the interviewer about expected
data delay. Time between when event happened and when
it was processed. If it should be no more than several minutes
- we must aggregate data on the fly. If several hours is ok, then we can store
raw events and process them in the background. Former approach is also known as stream data
processing, while latter is called batch data processing. The interviewer will let us know what option
she is interested the most. But because I have no-one to ask...anybody...no...I
will pick both options. And by the way combining both approaches makes
a lot of sense for many systems out there. We will store raw events, and because there
are so many of them, we will store events for several days or weeks only. And then purge old data. And we will also calculate and store numbers
in real-time. So that statistics is available for users
right away. By storing both raw events and aggregated
data we get the best of both worlds: fast reads, ability to aggregate data differently
and re-calculate statistics if there were bugs or failures on a real-time path. But there is a price to pay for all this flexibility,
the system becomes more complex and expensive. Great topic to discuss with the interviewer. Further in this video we will mainly focus
on real-time aggregation option. As I find it more educational from the system
design perspective. Now let's talk about where we store the data. The interviewer wants to know specific database
name and why we make this choice. We should know (and do not worry if you do
not know, you will know it in details in several minutes) that both SQL and NoSQL databases
can scale and perform well. So, we may want to evaluate both options. And here is where we should recall non-functional
requirements. Remember that we wrote them down on the whiteboard
before, right? What are those? Scalability, performance and availability. So, we should evaluate databases against these
requirements. And let's add some more requirements along
the way. Feel free to use this list for other interview
questions and system design in general. Database solution we chose should scale well
for both reads and writes. It should be fast and guarantee high availability. We should be able to achieve required level
of data consistency. We should understand how to recover data,
achieve security, apply future data model changes. We need to pick hardware and evaluate cost
of the solution. First, let's see how SQL databases handle
these requirements. Things are simple when we can store all our
data on a single database machine. But when a single machine is not enough, we
need to introduce more machines and split data between them. This procedure is called sharding or horizontal
partitioning. Each shard stores a subset of all the data. And because we now have several machines,
services that talk to the database need to know how many machines exist and which one
to pick to store and retrieve data. We discussed before that we have Processing
service, that stores data in the database and Query service, that retrieves data from
the database. We could have made both these services to
call every database machine directly. A better option is to introduce a light proxy
server that knows about all database machines and routes traffic to the correct shard. Now, both Processing and Query services talk
to this cluster proxy only. They do not need to know about each and every
database machine anymore. But cluster proxy has to know. Moreover, proxy needs to know when some shard
dies or become unavailable due to network partition. And if new shard has been added to the database
cluster, proxy should become aware of it. How do we achieve this? We introduce a new component - Configuration
service. Configuration service maintains a health check
connection to all shards. So, it always knows what database machines
are available. So, cluster proxy calls a particular shard. And instead of calling database instance directly,
we can introduce one more proxy - shard proxy, that sits in front of a database. Shard proxy will help us in many different
ways: it can cache query results, monitor database instance health and publish metrics,
terminate queries that take too long to return data and many more. Great, this setup helps us address several
requirements we mentioned before. Like scalability and performance. But availability is not yet addressed. What if database shard died? How to make sure data is not lost? I believe you already know the answer to this
question. We need to replicate data. Let's call each existing shard a master shard
or a leader. And for every master shard we introduce a
copy of it, called read replica or a follower. We call it read replica because writes still
go through a master shard, but reads may go through both master shard and a replica. We also put some replicas to a data center
different from their master shard. So that if the whole data center goes down,
we still have a copy of data available. So, when store data request comes, based on
the information provided by Configuration service, cluster proxy sends data to a shard. And data is either synchronously or asynchronously
replicated to a corresponding read replica. And when retrieve data request comes, cluster
proxy may retrieve data either from a master shard or from a read replica. Ideas we have just discussed is what Youtube
is actually using. They have built a database solution to scale
and manage large clusters of MySQL instances. It is called Vitess. Great, we now know how to scale SQL databases. But this solution does not seem simple, right. We have all these proxies, configuration service,
leader and replica instances. May be we can simplify things a little bit. Let's take a look at what NoSQL databases
can offer us. And more specifically, Apache Cassandra database. In NoSQL world we also split data into chunks
- shards, also known as nodes. But instead of having leaders and followers
we say that each shard is equal. We no longer need configuration service to
monitor health of each shard. Instead, let's allow shards talk to each other
and exchange information about their state. To reduce network load, we do not need each
shard to talk to every other shard. Every second shard may exchange information
with a few other shards, no more than 3. Quickly enough state information about every
node propagates throughout the cluster. This procedure is called a gossip protocol. Ok, each node in the cluster knows about other
nodes. And this is a big deal. Remember that previously we used Cluster Proxy
component to route requests to a particular shard. As Cluster Proxy was the only one who knew
about all shards. But now every node knows about each other. So, clients of our database no longer need
to call a special component for routing requests. Clients may call any node in the cluster and
node itself will decide where to forward this request further. Let's elaborate on this. Processing service makes a call to store views
count for some video B. And let's say node 4 is selected to serve this request. We can use a simple round robin algorithm
to chose this initial node, or we may be smarter and chose a node that is "closest" to the
client in terms of network distance. Let's call this node 4 a coordinator node. The coordinator node needs to decide which
node stores data for the requested video. We can use consistent hashing algorithm to
pick the node. As you may see node 1 should store the data
for the video B. Coordinator node will make a call to the node 1 and wait for the response. Actually, nothing stops coordinator node to
call multiple nodes to replicate data, for example 3 nodes if we want 3 copies of data. Waiting for 3 responses from replicas may
be too slow, so we may consider the write to be successful as soon as only 2 replication
requests succeeded. This approach is called quorum writes. Similar to quorum writes, there is a quorum
reads approach. When Query service retrieves views count for
video B, coordinator node 4 will initiate several read requests in parallel. In theory, the coordinator node may get different
responses from replica nodes. Why? Because some node could have been unavailable
when write request happened. That node has stale data right now, other
2 nodes has up-to-date data. Read quorum defines a minimum number of nodes
that have to agree on the response. Cassandra uses version numbers to determine
staleness of data. And similar to SQL databases, we want to store
copies of data across several different data centers. For high availability. Do you remember where else on this channel
we saw practical application of a consistent hashing algorithm? Right, when we designed distributed cache. Please check that video if you want to know
more about consistent hashing. Another important concept to mention is consistency. Remember that when we defined non-functional
requirements we chose availability over consistency? It simply means we prefer to show stale data
than no data at all. Let me clarify this. Synchronous data replication is slow, we usually
replicate data asynchronously. In case of a leader-follower replication for
example, some read replicas may be behind their master. Which leads to a situation when different
users see different total count for a video. This inconsistency is temporary, over time
all writes will propagate to replicas. This effect is known as eventual consistency. Cassandra actually extends the concept of
eventual consistency by offering tunable consistency. Let's discuss this big topic separately and
in more details. After discussing what we store and where,
let's think how we can store the data. There is a big difference how we do data modeling
for SQL and NoSQL databases. When designing data models for relational
databases we usually start with defining nouns in the system. We then convert these nouns into tables and
use foreign keys to reference related data in these tables. Let's take a look at the example. We want to build a report that shows the following
three entities: information about video, number of total views per hour for last several hours
and information about the channel this video belongs to. We pass video identifier as input for this
report. In a relational database we would define the
following three tables: video info table that contains information about videos, video stats
table that contains total views accumulated for each hour and channel info table that
stores information about video channels. And to generate report mentioned above, we
run a JOIN query that retrieves data from all three tables. And important property of a relational database
- data is normalized. It simply means we minimize data duplication
across different tables. For example we store video names in the video
info table only. And we do not store video names in other tables. Because if some video name changes, we have
to change it in several places in the database. Which may lead to inconsistent data. So, normalization is good for relational databases. But NoSQL databases promote a different paradigm. They say that we no longer think in terms
of nouns, but in terms of queries that will be executed in the system we design. And denormalization is perfectly normal. Not something that we always have to do, but
something that we should not be afraid of. In Cassandra for example, report mentioned
above will be logically represented as shown in the table. We store everything required for the report
together. And instead of adding rows, as in a relational
database, we keep adding columns for every next hour. Great, we have covered the storage portion
of our design. And hopefully by now you have got the idea
that both SQL and NoSQL database can be used for our solution. What database would you chose? Please let me know in the comments. I also would like to clarify one thing. As you know there are 4 types of NoSQL databases:
column, document, key-value and graph. So far we have used Cassandra as a representative
of NoSQL databases. We chose Cassandra because it is fault-tolerant,
scalable (both read and write throughput increases linearly as new machines are added). It supports multi datacenter replication and
works well with time-series data. And of course other options are available
as well. For a typical system design interview we usually
do not need to know architectures of different databases. But we need to know advantages and disadvantages
of those and when to use what. And please do not think that all NoSQL database
have architectures similar to the one we discussed earlier. Cassandra is a wide column database that supports
asynchronous masterless replication. But other NoSQL databases have different architectures. For example MongoDB, a documented-oriented
database, uses leader-based replication. HBase, which is another column-oriented data
store, similar to Cassandra, has a master-based architecture as well. We will talk more about different database
design principles on this channel. As well as when to use each particular database
type. Ok, enough talking about databases, let's
move on to data processing. First, let's define what processing really
means. When Youtube users open some video, we want
total views count for this video to be displayed immediately. It means we need to calculate view counts
on the fly, in real-time. Also, when video owner opens statistics for
the video we want to show per hour counts. So, processing basically means we get a video
view event and increment several counters: total and per hour counters. You are in front of the whiteboard, interviewer
is looking at you, ready to capture your next ideas. Where to start? As usual start with requirements. Remember, we wrote them down on the whiteboard
before? We want the processing service to scale together
with increase in video views. We want to process events quickly. And we do not want to lose data either when
some processing service machine crashes or when database becomes unavailable. So, we literally ask ourselves how to make
events processing scalable, reliable and fast. And this is another example of why knowledge
and preparation are crucial for system design interviews. Because curiosity and study will equip you
with blueprints you can apply for various system designs. Even if you never solved such problems in
the past. For example you know already that when we
want to scale data processing we should think about partitioning. When we want to avoid data loss we replicate
data. And when speed matters we should keep things
in memory and minimize disk reads. Easy, right? But before diving into processing service
detailed design, let's make sure we all agree on some basics. The first question I have for you is whether
or not we should pre-aggregate data in the processing service. Let me clarify the question. We have two options how to update counters
in the database. In the first option the processing service
increments counter for every incoming event. Meaning that if three users opened the same
video A, the processing service simply increments total count in the database three times. Logic is simple, event comes and we increment
the counter. In the second option, we accumulate data in
the processing service memory for some period of time, lets say several seconds. And add accumulated value to the database
counter. For example three users opened some video
A, the processing service takes each event and increments in-memory counter. And every several seconds in-memory counter
value is sent to the database to calculate the final count. I claim that the second option is better (for
large scale systems of course). Someone who agrees with me, please help me
prove it. By sharing your arguments in the comments
section. Further in this video I assume that we aggregate
data in memory. With this assumption in mind, what do you
think is better: push or pull? Where push means that some other service sends
events synchronously to the processing service. While pull means that the processing service
pulls events from some temporary storage. Although the answer is that both options are
totally valid and we can make both work, pull option has more advantages, as it provides
a better fault-tolerance support and easier to scale. Let me explain. We have two users opening two different videos
A and B. Processing service updates in-memory counters, returns successful responses back
to clients and the machine crashes without sending this data to the database. Data is lost. The alternative to push approach is for the
processing service to pull events from a storage. Events generated by users are stored in that
temporary storage first. Processing service machine pulls events and
updates in-memory counters. And if machine crashes, we still have events
in the storage and can re-process them. And here we come to another important concept,
called checkpointing. We have a temporary storage. And as you may see I intentionally draw it
as a queue. Because when events arrive we put them into
that storage in order, one by one. Fixed order allows us to assign an offset
for each event in the storage. This offset indicates event position in the
sequence. Events are always consumed sequentially. Every time an event is read from the storage,
the current offset moves forwards. After we processed several events and successfully
stored them in the database, we write checkpoint to some persistent storage. If processing service machine fails, it will
be replaced with another one and this new machine will resume processing where the failed
machine left off. Very important concept in stream data processing. Another important concept is partitioning. We already discussed it when talked about
databases. Main idea remains the same when applied to
events processing. Instead of putting all events into a single
queue, let's have several queues. Each queue is independent from the others. Every queue physically lives on its own machine
and stores a subset of all events. For example we compute a hash based on video
identifier and use this hash number to pick a queue. As you may see partitioning allows us to parallelize
events processing. More events we get, more partitions we create. Equipped with these basic concepts, we now
ready to look deep inside the processing service. We discussed so far that processing service
reads events from partition one by one, counts events in memory, and flushes this counted
values to the database periodically. So, we need a component to read events. The consumer establishes and maintains TCP
connection with the partition to fetch data. We can think of it as an infinite loop that
polls data from the partition. When consumer reads event it deserializes
it. Meaning it converts byte array into the actual
object. Usually, consumer is a single threaded component. Meaning that we have a single thread that
reads events. We can implement multi-threaded access. When several threads read from the partition
in parallel. But this approach comes with a cost, checkpointing
becomes more complicated and it is hard to preserve order of events if needed. Consumer does one more important thing - helps
to eliminate duplicate events. If the same message was submitted to the partition
several times (and later you will see why this can happen), we need a mechanism to avoid
double counting. To achieve this we use a distributed cache
that stores unique event identifiers for let's say last 10 minutes. And if several identical messages arrived
within a 10 minutes interval, only one of them (the first one) will be processed. Event then comes to the component that does
in-memory counting. Let's call it aggregator. Think of it as a hash table that accumulates
data for some period of time. Periodically, we stop writing to the current
hash table and create a new one. A new hash table keeps accumulating incoming
data. While old hash table is no longer counting
any data and each counter from the old hash table is sent to the internal queue for further
processing. Why do we need this internal queue? Why can't we send data directly to the database. Glad you asked. Remember, we have a single thread that reads
events from the partition. But nothing stops us from processing these
events by multiple threads, to speed up processing. Especially if processing takes time. By sending data to the internal queue we decouple
consumption and processing. The best analogy I could think of is a security
check queue at the airport. We can have a single person that quickly checks
a passport and a ticket for every passenger, but we need several lines for carry-on bags
screening, as this process takes much more time. You may argue whether we should put internal
queue before Aggregator component. It is up to you. Both options are fine. Ok, we now ready to send pre-aggregated values
to the database. So, we need a component responsible for this. Database writer is either a single-threaded
or a multi-threaded component. Each thread takes a message from the internal
queue and stores pre-aggregated views count in the database. Single-threaded version makes checkpointing
easier. But multi-threaded version increases throughput. Think about it for a moment and let me know
in the comments which version you like better. No need to worry if data flow is not completely
clear to you right now, we will have a simple simulation a bit later. It should help to further clarify the whole
process. Meanwhile, I would like to point out two more
important features of the database writer. The first concept is called a dead letter
queue. The dead-letter queue is the queue to which
messages are sent if they cannot be routed to their correct destination. Why do you think we may need one? You are correct, to protect ourselves from
database performance or availability issues. If database becomes slow or we cannot reach
database due to network issues, we simply push messages to the dead letter queue. And there is a separate process that reads
messages from this queue and sends them to the database. This concept is widely used when you need
to preserve data in case of downstream services degradation. So, you may apply it in many system designs. Another viable option is to store undelivered
messages on a local disk of the processing service machine. The second concept is data enrichment. Remember how we store data in Cassandra? We store it the way data is queried, right? If we want for example to show video title
in the report, we need to store video title together with views count. The same is true for the channel name and
many other attributes that we may want to display. But all these attributes do not come to the
processing service with every video view event. Event contains minimum information, like video
identifier and timestamp. It does not need to contain video title or
channel name or video creation date. So, these information comes from somewhere
else, right? Some database. But the trick here is that this database lives
on the same machine as the processing service. All these additional attributes should be
retrieved from this database really quickly. Thus, having it on the same machine eliminates
a need for remote calls. Such databases are called embedded databases. LinkedIn for example uses this concept for
the "who viewed your profile" feature. When they show additional information about
people who viewed your profile. For example, how many viewers have recruiter
job title. One last concept I would like to mention is
state management. We keep counters in memory for some period
of time. Either in in-memory store or internal queue. And every time we keep anything in memory
we need to understand what to do when machine fails and this in-memory state is lost. But that is easy, you may say. We have events stored in the partition, let's
just re-create the state from the point where we failed. In other words we just re-process events one
more time. This is a good idea. And it will work well if we store data in-memory
for a relatively short period of time and state is small. Sometimes it may be hard to re-create the
state from raw events from scratch. The solution in this case is to periodically
save the entire in-memory data to a durable storage. And new machine just re-loads this state into
its memory when started. We have looked inside the processing service. We now ready to finalize the data ingestion
pipeline. From the moment events appear in our counting
system. We know already that we have a set of partitions
and processing service reads events from them, count data in memory for some short period
of time and stores total count in the database. Someone needs to distribute data across partitions,
right? Let's have a component called Partitioner
service. Let's also have a load balancer component
in front of our partitioner service. To evenly distribute events across partitioner
service machines. When user opens a video, request goes through
API Gateway, component that represents a single-entry point into a video content delivery system. API Gateway routes client requests to backend
services. Our counting system may be one of such backend
services. And one more important component to mention
is the partitioner service client. Please prepare yourself for a lot of useful
information on the next slide. Pause this video, grab a cup of coffee or
tea. Give yourself a rest for the next 15 minutes
and come back. You will learn a ton. I could not think of a better way then dumping
all this information on you in a single shot. Do not blame me, blame distributed systems
for all the complexities they bring. We talked about database and processing service
in details. Now let's cover remaining 3 components of
the data ingestion path: partitioner service client, load balancer and partitioner service. We will discuss many ideas these components
are built upon. And will start with some basics. When client makes a request to a server, server
processes the request and sends back a response. The client initiates the connection by using
sockets. When a client makes a request, the socket
that handles that connection on the server side is blocked. This happens within a single execution thread. So, the thread that handles that connection
is blocked as well. And when another client sends a request at
the same time, we need to create one more thread to process that request. This is how blocking systems work. They create one thread per connection. Modern multi-core machines can handle hundreds
of concurrent connections each. But let's say server starts to experience
a slow down and number of active connections and threads increases. When this happens, machines can go into a
death spiral and the whole cluster of machines may die. Remember we designed a rate limiter in one
of the previous videos. That is exactly why we need rate limiting
solutions, to help keep systems stable during traffic peeks. Alternative to blocking I/O is non-blocking
I/O. When we can use a single thread on the server
side to handle multiple concurrent connections. Server just queues the request and the actual
I/O is then processed at some later point. Piling up requests in the queue are far less
expensive than piling up threads. Non-blocking systems are more efficient and
as a result has higher throughput. You may be wondering that if non-blocking
systems are so great, why we still have so many blocking systems out there? Because everything has a price. And the price of non-blocking systems is increased
complexity of operations. Blocking systems are easy to debug. And this is a big deal. In blocking systems we have a thread per request
and we can easily track progress of the request by looking into the thread's stack. Exceptions pop up the stack and it is easy
to catch and handle them. We can use thread local variables in blocking
systems. All these familiar concepts either do not
work at all or work differently in the non-blocking world. Moving on to buffering and batching. There are thousands of video view events happening
on Youtube every second. To process all these requests, API Gateway
cluster has to be big in size. Thousands of machines. If we then pass each individual event to the
partitioner service, partitioner service cluster of machines has to be big as well. This is not efficient. We should somehow combine events together
and send several of them in a single request to the partitioner service. This is what batching is about. Instead of sending each event individually,
we first put events into a buffer. We then wait up to several seconds before
sending buffer's content or until batch fills up, whichever comes first. There are many benefits of batching: it increases
throughput, it helps to save on cost, request compression is more effective. But there are drawbacks as well. It introduces some complexity both on the
client and the server side. For example think of a scenario when partitioner
service processes a batch request and several events from the batch fail, while other succeed. Should we re-send the whole batch? Or only failed events? What do you think? The next concept is timeouts. Timeouts define how much time a client is
willing to wait for a response from a server. We have two types of timeouts: connection
timeout and request timeout. Connection timeout defines how much time a
client is willing to wait for a connection to establish. Usually this value is relatively small, tens
of milliseconds. Because we only try to establish a connection,
no heavy request processing is happening just yet. Request timeout happens when request processing
takes too much time, and a client is not willing to wait any longer. To choose a request timeout value we need
to analyze latency percentiles. For example we measure latency of 1% of the
slowest requests in the system. And set this value as a request timeout. It means that about 1% of requests in the
system will timeout. And what should we do with these failed requests? Let's retry them. May be we just hit a bad server machine with
the first request. And the second attempt may hit a different
server machine, increasing our chances to succeed. But we should be smart when retry. Because if all clients retry at the same time
or do it aggressively, we may create a so-called retry storm event and overload sever with
too many requests. To prevent this, we should use exponential
backoff and jitter algorithms. Exponential backoff algorithm increases the
waiting time between retries up to a maximum backoff time. We retry requests several times, but wait
a bit longer with every retry attempt. And jitter adds randomness to retry intervals
to spread out the load. If we do not add jitter, backoff algorithm
will retry requests at the same time. And jitter helps to separate retries. Even with exponential backoff and jitter we
may still be in danger of too many retries. For example when partitioner service is down
or degraded. And majority of requests are retried. The Circuit Breaker pattern stops a client
from repeatedly trying to execute an operation that's likely to fail. We simply calculate how many requests have
failed recently and if error threshold is exceeded we stop calling a downstream service. Some time later, limited number of requests
from the client are allowed to pass through and invoke the operation. If these requests are successful, it's assumed
that the fault that was previously causing the failure has been fixed. We allow all requests at this point and start
counting failed requests from scratch. The loop completes. The Circuit Breaker pattern also has drawbacks. For example, it makes the system more difficult
to test. And it may be hard to properly set error threshold
and timers. By the way, have you noticed that everything
we discussed so far has tradeoffs? None of these concepts is a silver bullet. This is true for almost all concepts in distributed
systems. We should always know and remember about tradeoffs. It is hard, I agree. More knowledge and experience you have, easier
it becomes. Let's now talk about load balancing. It is a big topic of course, we will consider
only selected concepts. As you know load balancers distribute data
traffic between multiple servers. There are two types of load balancers: hardware
and software. Hardware load balancers are network devices
we buy from known organizations. Theses are powerful machines with many CPU
cores, memory and they are optimized to handle very high throughput. Millions of requests per second. Software load balancer is only software that
we install on hardware we choose. We do not need big fancy machines, and many
software load balancers are open source. Load balancers provided by public clouds (for
example ELB from AWS) are examples of software load balancer type as well. Another gradation of load balancers is what
traffic they serve TCP or HTTP. It is a bit more complicated than that, but
I do not want to overload you with terminology too much. Let's keep things simple for now. TCP load balancers simply forward network
packets without inspecting the content of the packets. Think of it as if we established a single
end-to-end TCP connection between a client and a server. This allows TCP load balancers to be super
fast and handle millions of requests per second. HTTP load balancers, on contrast, terminate
the connection. Load balancer gets an HTTP request from a
client, establishes a connection to a server and sends request to this server. HTTP load balancer can look inside a message
and make a load‑balancing decision based on the content of the message. For example based on a cookie information
or a header. Load balancers may use several algorithms
to distribute the load. Round robin algorithm distributes requests
in order across the list of servers. Least connections algorithm sends requests
to the server with the lowest number of active connections. Least response time algorithm sends requests
to the server with the fastest response time. Hash-based algorithms distribute requests
based on a key we define, such as the client IP address or the request URL. Ok, you got it, there are many benefits of
using load balancers. I should stop selling them to you. Instead, let's return to our original system
design problem and address several very specific questions. Such as, how does our partitioner service
client know about load balancer? How does load balancer know about partitioner
service machines? And how does load balancer guarantee high
availability? Because it looks like a single point of failure,
right? Here is where we should recall DNS, Domain
Name System. DNS is like a phone book for the internet. It maintains a directory of domain names and
translate them to IP addresses. We register our partitioner service in DNS,
specify domain name, for example partitionerservice.domain.com and associate it with IP address of the load
balancer device. So, when clients hit domain name, requests
are forwarded to the load balancer device. For the load balancer to know about partitioner
service machines, we need to explicitly tell the load balancer the IP address of each machine. Both software and hardware load balancers
provides API to register and unregister servers. Load balancers need to know which server from
the registered list are healthy and which are unavailable at the moment. This way load balancers ensure that traffic
is routed to healthy servers only. Load balancer pings each server periodically
and if unhealthy server is identified, load balancer stops to send traffic to it. It will then resume routing traffic to that
server when it detects that the server is healthy again. As for high availability of load balancers,
they utilize a concept of primary and secondary nodes. The primary load balancer accepts connections
and serves requests, while the secondary load balancer monitors the primary. If, for any reason, the primary load balancer
is unable to accept connections, the secondary one takes over. Primary and secondary also live in different
data centers, in case one data center goes down. There are several other interesting topics
related to load balancing, and we should definitely cover those in a separate video. But now let's move on to the next two components
of our design, partitioner service and partitions. With partitioner service it is more or less
clear. It's a web service that gets requests from
clients, looks inside each request to retrieve individual video view events (because remember
we batch events on the client side), and routs each such event (we can also use the word
message) to some partition. But what partitions are? Partitions is also
a web service, that gets messages and stores them on disk in the form of the append-only
log file. So, we have a totally-ordered sequence of
messages ordered by time. This is not a single very large log file,
but a set of log files of the predefined size. Partitioner service has to use some rule,
partition strategy, that defines which partition gets what messages. A simple strategy is to calculate a hash function
based on some key, let's say video identifier and chose a machine based on this hash. This simple strategy does not work very well
with large scale. As it may lead to so called "hot partitions". For example when we have a very popular video
or set of videos and all view events for them go to the same partition. One approach to deal with hot partitions is
to include event time, for example in minutes, into partition key. All video events within the current minute
interval are forwarded to some partition. Next minute, all events go to a different
partition. Within one minute interval a single partition
gets a lot of data, but over several minutes data is spread more evenly among partitions. Another solution to hot partitions problem
is to split hot partition into two new partitions. To get an idea how this approach might work,
remember consistent hashing algorithm and how adding a new node to the consistent hashing
ring splits a range of keys into two new ranges. And if to push this idea of partition split
even further, we may explicitly allocate dedicated partitions for some popular video channels. All video view events from such channels go
to their allocated partitions. And view events from all other channels never
go to those partitions. These are the powerful techniques and there
is little information on the "hot partitions" topic on the internet. So, please remember them. To send messages to partitions, partitioner
service needs to know about every partition. This is where the concept of service discovery
comes on stage. In the world of microservices there are two
main service discovery patterns: server-side discovery and client-side discovery. We already looked at server-side discovery
when talked about load balancers. Clients know about load balancer, load balancer
knows about server-side instances. Easy. But we do not need a load balancer between
partitioner service and partitions. Partitioner service itself acts like a load
balancer by distributing events over partitions. This is a perfect match for the client-side
discovery pattern. With client-side discovery every server instance
registers itself in some common place, named service registry. Service registry is another highly available
web service, which can perform health checks to determine health of each registered instance. Clients then query service registry and obtain
a list of available servers. Example of such registry service is Zookeeper. In our case each partition registers itself
in Zookeeper, while every partitioner service instance queries Zookeeper for the list of
partitions. Do you remember where else we discussed client-side
service discovery pattern, although we did not name it there that way? When we talked about distributed cache design. When cache client needs to pick a cache shard
that stores the data. There we discussed several other options as
well. Please go and check that video. One more option for service discovery is similar
to what Cassandra does. Remember we mentioned before that Cassandra
nodes talk to each other? So, every node in the cluster knows about
other nodes. It means clients only need to contact one
node from the server cluster to figure out information about the whole cluster. Think about this. Next important concept is replication. We must not lose events when store them in
partitions. So, when event is persisted in a partition,
we need to replicate it. If this partition machine goes down, events
are not lost. There are three main approaches to replication:
single leader replication, multi leader replication and leaderless replication. Do you remember where we used single leader
replication? Correct, when we discussed how to scale a
SQL database. Great, and do you remember when we talked
about leaderless replication? When we discussed how Cassandra works, right? So far we did not talk about multi leader
replication. Let me make a separate deep dive into replication
topic. And only mention right now that multi leader
replication is mostly used to replicate between several data centers. So, which approach should we chose for partitions? Let's go with single leader replication. Each partition will have a leader and several
followers. We always write events and read them from
the leader only. While a leader stays alive, all followers
copy events from their leader. And if the leader dies, we choose a new leader
from its followers. The leader keeps track of its followers: checks
whether the followers are alive and whether any of the followers is too far behind. If a follower dies, gets stuck, or falls behind,
the leader will remove it from the list of its followers. Remember a concept of a quorum write in Cassandra? We consider a write to be successful, when
predefined number of replicas acknowledge the write. Similar concept applies to partitions. When partitioner service makes a call to a
partition, we may send response back as soon as leader partition persisted the message,
or only when message was replicated to a specified number of replicas. When we write to a leader only, we may still
lose data if leader goes down before replication really happened. When we wait for the replication to complete,
we increase durability of the system, but latency will increase. Plus, if required number of replicas is not
available at the moment, availability will suffer. Tradeoffs, as usual. Next topic is the last one on this slide,
I promise. Let's talk about message formats. We can use either textual or binary formats
for messages. Popular textual formats are XML, CSV, JSON. Popular binary formats are Thrift, Protocol
Buffers and Avro. What's great about textual formats - they
are human-readable. They are well-known, widely supported and
heavily used by many distributed systems. But for the large scale real-time processing
systems binary formats provide much more benefits. Messages in binary format are more compact
and faster to parse. And why is this? As mentioned before, messages contain several
attributes, such as video identifier, timestamp, user related information. When represented in JSON format, for example,
every message contains field names, which greatly increases total message size. Binary formats are smarter. Formats we mentioned before require a schema. And when schema is defined we no longer need
to keep field names. For example Apache Thrift and Protocol Buffers
use field tags instead of field names. Tags are just numbers and they act like aliases
for fields. Tags occupy less space when encoded. Schemas are crucial for binary formats. Message producers (or clients) need to know
the schema to serialize the data. Message consumers (processing service in our
case) require the schema to deserialize the message. So, schemas are usually stored in some shared
database where both producers and consumers can retrieve them. Important to mention that schemas may and
will change over time. We may want to add more attributes into messages
and use them later for counting or filtering. Apache Avro is a good choice for our counting
system. That was tough. Talking about all these concepts completed
our discussion of the data ingestion path. And all the most difficult stuff is behind
us. Do not worry if some mentioned before concepts
are not completely clear to you right now. Each of them deserves a video on its own. And we will talk more about them on this channel. Eventually, you will feel comfortable using
all of them, either on the interview or in your day to day job. Now let's take a look at data retrieval path. When users open a video on Youtube, we need
to show total views count for this video. To build a video web page, several web services
are called. A web service that retrieves information about
the video, a web service that retrieves comments, another one for recommendations. Among them there is our Query web service
that is responsible for video statistics. All these web services are typically hidden
behind an API Gateway service, a single-entry point. API Gateway routes client requests to backend
services. So, get total views count request comes to
the Query service. We can retrieve the total count number directly
from the database. Remember we discussed before how both SQL
and NoSQL databases scale for reads. But total views count scenario is probably
the simplest one. This is just a single value in the database
per video. The more interesting use case is when users
retrieve time-series data, which is a sequence of data points ordered in time. For example, when channel owner wants to see
statistics for her videos. As discussed before, we aggregate data in
the database per some time interval, let's say per hour. Every hour for every video. That is a lot of data, right? And it grows over time. Fortunately, this is not a new problem and
solution is known. Monitoring systems, for example, aggregate
data for every 1 minute interval or even 1 second. You can imaging how huge those data sets can
be. So, we cannot afford storing time series data
at this low granularity for a long period of time. The solution to this problem is to rollup
the data. For example, we store per minute count for
several days. After let's say one week, per minute data
is aggregated into per hour data. And we store per hour count for several months. Then we rollup counts even further and data
that is older than let's say 3 months, is stored with 1 day granularity. And the trick here is that we do not need
to store old data in the database. We keep data for the last several days in
the database, but the older data can be stored somewhere else, for example, object storage
like AWS S3. In the industry, you may also hear terms like
a hot storage and a cold storage. Hot storage represents frequently used data
that must be accessed fast. Cold storage doesn’t require fast access. It mostly represents archived and infrequently
accessed data. When request comes to the Query service, it
does so-called data federation, when it may need to call several storages to fulfill the
request. Most recent statistics is retrieved from the
database, while older statistics is retrieved from the Object Storage. Query service then stitches the data. And this is ideal use case for the cache. We should store query results in a distributed
cache. This helps to further improve performance
of queries and scale them. We covered both data ingestion and data retrieval. Not many things left. Let me show you the full picture and share
with you several other important topics. Three users opened some video A. And API Gateway
got 3 requests. Partitioner service client batches all three
events and sends them in a single request to the partitioner service. This request hits the load balancer first. And load balancer routes it to one of the
partitioner service machines. Partitioner service gets all three events
from the request and sends them to some partition. All three events end up in the same partition,
as we partition data based on the video identifier. Here is where processing service appears on
the stage. Partition consumer reads all three messages
from the partition one by one and sends them to the aggregator. Aggregator counts messages for a one minute
period and flushes calculated values to the internal queue at the end of that minute. Database writer picks count from the internal
queue and sends it to the database. In the database we store count per hour and
the total number of views for each video. So, we just add a one minute value to the
current hour count as well as the total count. Total count was 7 prior to this minute and
we add 3 for the current minute. And during data retrieval, when user opens
video A, API Gateway sends request to the Query service. Query service checks the cache. And if data is not found in the cache, or
cache value has expired, we call the database. Total count value is then stored in the cache
and Query service returns the total count back to the user. I hope that this simulation helped you further
understand the meaning of each component in the architecture. Feel free to post any questions you have in
the comments section. Another important aspect of an interview and
system design in general is a technology stack. When we design some system, we usually do
not need to reinvent the wheel. We rely on some well-regarded technologies. Either open source or commercial. Public cloud services. During the interview do not forget to discuss
these technologies. You may do this along the way or at the end
of the interview. So, let's see what technologies we may use
for our solution. Netty is a high-performance non-blocking IO
framework for developing network applications, both clients and servers. Frameworks such as Hystrix from Netflix and
Polly simplify implementation of many client-side concepts we discussed before: timeouts, retries,
circuit breaker pattern. Citrix Netscaler is probably the most famous
hardware load balancer. Among software load balancers NGINX is a very
popular choice. And if we run our counting system in the cloud,
for example Amazon cloud, then Elastic Load Balancer is a good pick. Instead of using our custom Partitioner service
and partitions, we could use Apache Kafka instead. Or Kafka's public cloud counterpart, like
Amazon Kinesis. To process events and aggregate them in memory
we can use stream-processing frameworks such as Apache Spark or Flink. Or cloud-based solutions, such as Kinesis
Data Analytics. We already talked about Apache Cassandra. Another popular choice for storing time-series
data is Apache HBase database. These are wide column databases. There are also databases optimized for handling
time series data, like InfluxDB. We also mentioned before that we may need
to store raw events to recalculate counts in case of any error or if customers need
to run ad-hoc queries. We can store raw events in Apache Hadoop or
in a cloud data warehouse, such as AWS Redshift. And when we roll up the data and need to archive
it, AWS S3 is a natural choice. I would like to also mention several other
important technologies we may use in our design. Vitess is a database solution for scaling
and managing large clusters of MySQL instances. Vitess has been serving all Youtube database
traffic since 2011. In several places of our design we rely on
a distributed cache: for message deduplication and to scale read data queries. Redis is a good option. For a dead-letter queue mechanism, when we
need to temporarily queue undelivered messages, we may use an open-source message-broker such
as RabbitMQ. Or public cloud alternative, such as Amazon
SQS. For data enrichment, when we store video and
channel related information locally on the machine and inject this information in real-time,
we may use RocksDB, a high performance embedded database for key-value data. To do leader election for partitions and to
manage service discovery, we may rely on Apache Zookeeper, which is a distributed configuration
service. For the service discovery piece we actually
have an alternative, Eureka web service from Netflix. To monitor each of our system design components
we may rely on monitoring solutions provided by public cloud services, such as AWS CloudWatch. Or use a popular stack of open source frameworks:
Elasticsearch, Logstash, Kibana. Or ELK for short. We discussed before that binary message format
is preferred for our system. Popular choices are Thrift, Protobuf and Avro. For Partitioner service to partition the data,
we should use a good hashing function, for example a MurmurHash. We are done with the detailed system design. And here is where our interviewer will start
challenging us with the choices we have made. There are several goals of this exercise. The interviewer wants to see that we know
and understand tradeoffs. Have you noticed that I usually use verbs
like may or can and rarely use must or have to? This is because we usually have several options
to choose from. When we design a system or a part of it in
real life, we usually bring several options to discuss with the team, right? And very important is not only know your options,
but be able to explain pros and cons of each one. We discussed many tradeoffs of individual
components throughout this video. Let's see what else the interviewer may want
to discuss with us. To identify bottlenecks in the system we need
to test it under a heavy load. This is what performance testing is about. There are several types of performance testing. We have load testing, when we measure behavior
of a system under a specific expected load. We have stress testing, when we test beyond
normal operational capacity, often to a breaking point. We have soak testing, when we test a system
with a typical production load for an extended period of time. With load testing we want to understand that
our system is indeed scalable and can handle a load we expect. For example, a two or three times increase
in traffic. With stress testing we want to identify a
breaking point in the system. Which component will start to suffer first. And what resource it will be: memory, CPU,
network, disk IO. And with soak testing we want to find leaks
in resources. For example, memory leaks. So, generating high load is the key. Tools like Apache JMeter can be used to generate
a desired load. Health monitoring. All the components of our system must be instrumented
with monitoring of their health. Metrics, dashboards and alerts should be our
friends all the time. Metric is a variable that we measure, like
error count or processing time. Dashboard provides a summary view of a service’s
core metrics. And alert is a notification sent to service
owners in a reaction to some issue happening in the service. Remember about the four golden signals of
monitoring, which are latency, traffic, errors, and saturation. Let's leave details for a separate video. Ok, we designed a system and deployed all
the components. We know it is running healthy and can handle
a high load. But how to make sure it counts things correctly? This becomes critical when we not just count
video views, but, for example, number of times some ad was played in a video. As we need to properly charge an ad owner
and pay money to a video owner. This problem is typically addressed by building
an audit system. There can be two flavors of audit systems. Let's call them weak and strong. Weak audit system is a continuosly running
end-to-ed test. When let's say once a minute we generate several
video view events in the system, call query service and validate that returned value equals
to the expected count. This simple test gives us a high confidence
that the system counts correctly. And it is easy to implement and maintain such
test. But unfortunately, this test is not 100% reliable. What if our system loses events in some rare
scenarios? And weak audit test may not identify this
issue for a long period of time. That is why we may need a better approach. Strong audit system calculates video views
using a completely different path then out main system. For example we store raw events in Hadoop
and use MapReduce to count events. And then compare results of both systems. Having two different systems doing almost
the same may seem like an overkill, right? You may be surprised but this is not so uncommon
in practice. Not such a long time ago it was quite a popular
idea. And it even has a name - Lambda Architecture. The key idea is to send events to a batch
system and a stream processing system in parallel. And stitch together the results from both
systems at query time. You can get a better understanding of this
idea if you watch the previous video on the channel, where we designed a system for finding
the top k most frequent items. Ideally, we should have a single system. Let me share with you advice from Jay Kreps,
who is one of the authors of Apache Kafka. We should use a batch processing framework
like MapReduce if we aren’t latency sensitive, and use a stream processing framework if we
are, but not to try to do both at the same time unless we absolutely must. And please note that out today's problem can
indeed be solved with MapReduce. But MapReduce-based system would have a much
higher latency. We already discussed the problem with popular
videos. I will just reiterate the key idea. We have to spread events coming for a popular
video across several partitions. Otherwise, a single consumer of a single "hot"
partition may not be able to keep up with the load. And will fall behind. Let's talk more about this. Imaging a situation when the processing service
cannot keep up with the load. Maybe because number of events is huge, maybe
because processing of a single event is complicated and time consuming. I will not dive too much into details, but
describe the main idea of the solution. We batch events and store them in the Object
Storage service, for example AWS S3. Every time we persist a batch of events, we
send a message to a message broker. For example SQS. Then we have a big cluster of machines, for
example EC2, that retrieve messages from SQS, read a corresponding batch of events from
S3 and process each event. This approach is a bit slower than stream
processing, but faster than batch processing. Everything is a tradeoff. Let's summarize what we have discussed. We start with requirements clarification. And more specifically, we need to define APIs,
what exactly our system is supposed to do. We then discuss non-functional requirements
with the interviewer and figure out what qualities of the system she is most interested in. We can now outline a high-level architecture
of the system. Draw some key components on the whiteboard. At the next stage we should dive deep into
several of those components. Our interviewer will help us understand what
components we should focus on. And the last important step is to discuss
bottlenecks and how to deal with them. And let me quickly remind you some specifics
we discussed for each of these steps. To define APIs, we discuss with the interviewer
what specific behaviors or functions of the system we need to design. We write down verbs characterizing these functions
and start thinking about input parameters and return values. We then can make several iterations to brush
up the APIs. After this step we should be clear on what
the scope of the design is. To define non-functional requirements, just
know what your options are. Open a list of non-functional requirements
on wiki and read the list. There are many of them. I recommend to focus on scalability, availability
and performance. Among other popular choices we have consistency,
durability, maintainability and cost. Try to pick not more than 3 qualities. To outline a high-level design, think about
how data gets into the system, how it gets out of the system and where data is stored
inside the system. Draw these components on the whiteboard. It is ok to be rather generic at this stage. Details will follow later. And although it is not easy, try to drive
the conversation. Our goal here is to get understanding of what
components to focus on next. And the interviewer will help us. While designing specific components, start
with data. How it is stored, transferred and processed. Here is where our knowledge and experience
becomes critical. By using fundamental concepts of system design
and by knowing how to combine these concepts together, we can make small incremental improvements. And apply relevant technologies along the
way. After technical details are discussed, we
can move to discussing other important aspects of the system. Listen carefully to the interviewer. She sees bottlenecks of our design and in
her questions there will be hints what those bottlenecks are. And what can really help us here is the knowledge
of different tradeoffs in system design. We just need to pick and apply a proper one. Today we covered a big topic. If you are still with me watching this video
you should be proud of yourself. Seriously. There were many system design concepts covered
in the video. And I hope you have a better understanding
right now why I consider knowledge the key to the successful system design interview. And system design in general. And although we talked about a specific problem
today, like video views counting, the same ideas can be applied to other problems, for
example counting likes, shares, reposts, ad impressions and clicks. The same ideas can be applied to designing
monitoring systems, when we count metrics. When we design a fraud prevention system we
need to count number of times each credit card was used recently. When we design recommendation service we may
use counts as input to machine learning models. When we design "what's trending" service,
we count all sorts of different reactions: views, re-tweets, comments, likes. And many other applications. There are plenty of other important system
design concepts we have not covered today. As it is practically impossible to do in a
single video. But we will have them covered in other videos
on this channel. Thank you for being with me and I will see
you next time. Bye.
