76 posts tagged with "system design"

Motivation & Assumptions​

• PB-level Blob storage
• Traditional NFS based desgin (Each image stored as a file) has metadata bottleneck: large metadata size severely limits the metadata hit ratio.
  • Explain more about the metadata overhead

For the Photos application most of this metadata, such as permissions, is unused and thereby wastes storage capacity. Yet the more significant cost is that the file’s metadata must be read from disk into memory in order to find the file itself. While insignificant on a small scale, multiplied over billions of photos and petabytes of data, accessing metadata is the throughput bottleneck.

Solution​

Eliminates the metadata overhead by aggregating hundreds of thousands of images in a single haystack store file.

Architecture​

Data Layout​

index file (for quick memory load) + haystack store file containing needles.

index file layout

haystack store file

CRUD Operations​

• Create: write to store file and then ==async== write index file, because index is not critical
• Read: read(offset, key, alternate_key, cookie, data_size)
• Update: Append only. If the app meets duplicate keys, then it can choose one with largest offset to update.
• Delete: soft delete by marking the deleted bit in the flag field. Hard delete is executed by the compact operation.

Usecases​

Upload

Download

Disclaimer: All things below are collected from public sources or purely original. No Uber-confidential stuff here.

Requirements​

• ride hailing service targeting the transportation markets around the world
• realtime dispatch in massive scale
• backend design

Architecture​

Why micro services?​

==Conway's law== says structures of software systems are copies of the organization structures.

Combining monolithic ==codebase== and micro services can bring benefits from both sides.

Dispatch Service​

• consistent hashing sharded by geohash
• data is transient, in memory, and thus there is no need to replicate. (CAP: AP over CP)
• single-threaded or locked matching in a shard to prevent double dispatching

Payment Service​

==The key is to have an async design==, because payment systems usually have a very long latency for ACID transactions across multiple systems.

• leverage event queues
• payment gateway w/ Braintree, PayPal, Card.io, Alipay, etc.
• logging intensively to track everything
• APIs with idempotency, exponential backoff, and random jitter

UserProfile Service and Trip Service​

• low latency with caching
• UserProfile Service has the challenge to serve users in increasing types (driver, rider, restaurant owner, eater, etc) and user schemas in different regions and countries.

Push Notification Service​

• Apple Push Notifications Service (not quite reliable)
• Google Cloud Messaging Service GCM （it can detect the deliverability) or
• SMS service is usually more reliable

Requirements​

1. Data size: Data size of values is too large to be held in memory, and we should leverage the external storage for them. However, we can still keep the data keys in memory.
2. Single-host solution. No distributed design.
3. Optimize for write.

Solution​

• In-memory hashmap index + index hint file + data files
• Append-only for write optimization. Have only one active data file for write. And compact active data to the older data file(s) for read.

Components​

1. In-memory HashMap<Key, <FildId, ValueOffset, ValueSize, Timestamp>>

2. Data file layout

|crc|timestamp|key_size|value_size|key|value|
...

3. (index) hint file that the in-memory hashmap can recover from

Operations​

Delete: get the location by the in-memory hashmap, if it exists, then go to the location on the disk to set the value to a magic number.

Get: get the location by the in-memory hashmap, and then go to the location on the disk for the value.

Put: append to the active data file and update the in-memory hash map.

Periodical compaction strategies

• Copy latest entries: In-memory hashmap is always up-to-date. Stop and copy into new files. Time complexity is O(n) n is the number of valid entries.

  • Pros: Efficient for lots of entries out-dated or deleted.
  • Cons: Consume storage if little entries are out-dated. May double the space. (can be resolved by having a secondary node do the compression work with GET/POST periodically. E.g., Hadoop secondary namenode).

• Scan and move: foreach entry, if it is up-to-date, move to the tail of the validated section. Time complexity is O(n) n is the number of all the entries.

  • Pros:
    • shrink the size
    • no extra storage space needed
  • Cons:
    • Complex and need to sync hashmap and storage with transactions. May hurt performance.

Following up questions

• How to detect records that can be compacted?
  • Use timestamp.
• What if one hashmap cannot fit into a single machine’s memory?
  • Consistent hashing, chord DHT, query time complexity is O(logn) with the finger table, instead of O(1) here with a hashmap.

Design a system that can convert URLs provided by users into short URLs, allowing users to access their original URLs (hereinafter referred to as long URLs) using these short URLs. Describe how this system operates, including but not limited to the following questions: How are short URLs allocated? How is the mapping between short URLs and long URLs stored? How is the redirection service implemented? How is access data stored?

Assumptions: The initial problem description does not include these assumptions. An excellent candidate will ask about system scale when given a specific design.

• There are approximately tens of thousands of long URL domains.
• The traffic for new long URLs is about 10,000,000 per day (100 per second).
• The traffic for the redirection service using short URLs to access long URLs is about 10 billion per day (100,000 per second).
• Remind the candidate that these are average figures - during peak times, these numbers can be much higher (one type of peak time is time-related, such as when users return home from work, and another type is event-related, such as during the Spring Festival Gala).
• Recent data (e.g., today's data) should be collected in advance and should be available within five minutes when users want to view it.
• Historical data should be calculated daily.

Assumptions​

1 billion new URLs per day, 100 billion short URL accesses. The shorter the short URL, the better. Data presentation (real-time/daily/monthly/yearly).

URL Encoding​

http://blog.codinghorror.com/url-shortening-hashes-in-practice/

Method 1: md5 (128 bits, 16 hexadecimal digits, collisions, birthday paradox, 2^(n/2) = 2^64) Shorter? (64 bits, 8 hexadecimal digits, collisions 2^32), base 64.

• Advantages: Hashing is relatively simple and easy to scale horizontally.
• Disadvantages: Too long, how to handle expired URLs?

Method 2: Distributed ID generator. (Base 62: az, AZ, 0~9, 62 characters, 62^7), partitioning: each node contains some IDs.

• Advantages: Easier to eliminate expired URLs, shorter URLs.
• Disadvantages: Coordination between different partitions (e.g., ZooKeeper).

Key-Value (KV) Storage​

MySQL (10k requests per second, slow, no need for a relational database), key-value (100k requests per second, Redis, Memcached).

An excellent candidate will ask about the expected lifespan of short URLs and design a system that can automatically clean up expired short URLs.

Follow-Up​

Question: How to generate short URLs?

• A poor candidate might suggest using a single ID generator (single point of failure) or require coordination between ID generators for each ID generation. For example, using an auto-increment primary key in a database.
• An acceptable candidate might suggest using md5 or some UUID generators that can generate IDs independently on some nodes. These methods can generate non-colliding IDs in a distributed system, allowing for the production of a large number of short URLs.
• An excellent candidate will design a method using several ID generators, where each generator first reserves a block of ID sequences from a central coordinator (e.g., ZooKeeper), and these ID generators can allocate IDs from their ID sequences independently, cleaning up their ID sequences when necessary.

Question: How to store the mapping between long URLs and short URLs?

• A poor candidate might suggest using a single, non-distributed, non-relational database. It is merely a simple key-value database.
• An excellent candidate will suggest using a simple distributed storage system, such as MongoDB/HBase/Voldemort, etc.
• A more excellent candidate will ask about the expected usage cycle of short URLs and then design a system that can clean up expired short URLs.

Question: How to implement the redirection service?

• A poor candidate will design the system from scratch to solve problems that have already been solved.
• An excellent candidate will suggest using an existing HTTP server with a plugin to translate the short URL ID, look up this ID in the database, update access data, return a 303 status, and redirect to the long URL. Existing HTTP servers include Apache/Jetty/Netty/Tomcat, etc.

Question: How to store access data?

• A poor candidate will suggest writing to the database on every access.
• An excellent candidate will suggest having several different components handle this task: generating access stream data, collecting and organizing it, and writing it to a permanent database after a certain period.

Question: How to separate the different components of storing access data proposed by the excellent candidate?

• An excellent candidate will suggest using a low-latency information system to temporarily store access data and then hand the data over to the collection and organization component.
• The candidate may ask how often access data needs to be updated. If updated daily, a reasonable method would be to store it in HDFS and use MapReduce to compute the data. If near-real-time data is required, the collection and organization component must compute the necessary data.

Question: How to block access to restricted websites?

• An excellent candidate will suggest maintaining a blacklist of domains in the key-value database.
• A good candidate might propose some advanced technologies that can be used when the system scales significantly, such as bloom filters.

Failover: Failover is a backup operational mode used to enhance system stability and availability. When the primary component fails or is scheduled for downtime, the functions of system components (such as processors, servers, networks, or databases) are transferred to secondary system components.

Cold Backup: Cold backup refers to copying critical files to another location, using features or metrics/alerts to track failures. The system provides a new standby node in the event of a failure; however, cold backup is only suitable for stateless services. For backing up Oracle databases, cold backup is the fastest and safest method.

Hot Backup: This involves maintaining two active systems that share the same task roles, meaning the system operates normally while providing backup. The data between the two systems is nearly mirrored in real-time and contains the same information.

Warm Backup: This keeps two active systems, where the secondary system does not consume traffic unless a failure occurs.

Checkpoint (or similar to Redis snapshots): The system uses write-ahead logging (WAL) to record requests before processing tasks. The standby node recovers from the log during failover.

• Disadvantages
  • A large amount of log recovery can be time-consuming
  • Data loss since the last checkpoint
• User Cases: Storm, WhillWheel, Samza

Dual-host (or all-host) mode: This keeps two active systems behind a load balancer. The hosts operate in parallel, and data replication is bidirectional.

Cold Standby: Use heartbeat or metrics/alerts to track failure. Provision new standby nodes when a failure occurs. Only suitable for stateless services.

Hot Standby: Keep two active systems undertaking the same role. Data is mirrored in near real time, and both systems will have identical data.

Warm Standby: Keep two active systems but the secondary one does not take traffic unless the failure occurs.

Checkpointing (or like Redis snapshot): Use write-ahead log (WAL) to record requests before processing. Standby node recovers from the log during the failover.

• cons
  • time-consuming for large logs
  • lose data since the last checkpoint
• usercase: Storm, WhillWheel, Samza

Active-active (or all active): Keep two active systems behind a load balancer. Both of them take in parallel. Data replication is bi-directional.

Design a system to take user-provided URLs and transform them to a shortened URLs that redirect back to the original. Describe how the system works. How would you allocate the shorthand URLs? How would you store the shorthand to original URL mapping? How would you implement the redirect servers? How would you store the click stats?

Assumptions: I generally don't include these assumptions in the initial problem presentation. Good candidates will ask about scale when coming up with a design.

• Total number of unique domains registering redirect URLs is on the order of 10s of thousands
• New URL registrations are on the order of 10,000,000/day (100/sec)
• Redirect requests are on the order of 10B/day (100,000/sec)
• Remind candidates that those are average numbers - during peak traffic (either driven by time, such as 'as people come home from work' or by outside events, such as 'during the Superbowl') they may be much higher.
• Recent stats (within the current day) should be aggregated and available with a 5 minute lag time
• Long look-back stats can be computed daily
• Targeted use cases: The shortened URLs are to be copied-and-pasted only.
  • The URLs won't be typed in via a keyboard. Therefore, don't worry about distinguishing between 0 and o, l and 1, etc.
  • The URLs won't be spelled out verbally. Therefore, there's no need to make the shortened URLs phonetic.

Assumptions​

1B new URLs per day, 100B entries in total the shorter, the better show statics (real-time and daily/monthly/yearly)

Encode Url​

http://blog.codinghorror.com/url-shortening-hashes-in-practice/

Choice 1. md5(128 bit, 16 hex numbers, collision, birthday paradox, 2^(n/2) = 2^64) truncate? (64bit, 8 hex number, collision 2^32), Base64.

• Pros: hashing is simple and horizontally scalable.
• Cons: too long, how to purify expired URLs?

Choice 2. Distributed Seq Id Generator. (Base62: az, AZ, 0~9, 62 chars, 62^7), sharding: each node maintains a section of ids.

• Pros: easy to outdate expired entries, shorter
• Cons: coordination (zookeeper)

KV store​

MySQL(10k qps, slow, no relation), KV (100k qps, Redis, Memcached)

A great candidate will ask about the lifespan of the aliases and design a system that purges aliases past their expiration.

Followup​

Q: How will shortened URLs be generated?

• A poor candidate will propose a solution that uses a single id generator (single point of failure) or a solution that requires coordination among id generator servers on every request. For example, a single database server using an auto-increment primary key.
• An acceptable candidate will propose a solution using an md5 of the URL, or some form of UUID generator that can be done independently on any node. While this allows distributed generation of non- colliding IDs, it yields large "shortened" URLs
• A good candidate will design a solution that utilizes a cluster of id generators that reserve chunks of the id space from a central coordinator (e.g. ZooKeeper) and independently allocate IDs from their chunk, refreshing as necessary.

Q: How to store the mappings?

• A poor candidate will suggest a monolithic database. There are no relational aspects to this store. It is a pure key-value store.
• A good candidate will propose using any light-weight, distributed store. MongoDB/HBase/Voldemort/etc.
• A great candidate will ask about the lifespan of the aliases and design a system that ==purges aliases past their expiration==

Q: How to implement the redirect servers?

• A poor candidate will start designing something from scratch to solve an already solved problem
• A good candidate will propose using an off-the-shelf HTTP server with a plug-in that parses the shortened URL key, looks the alias up in the DB, updates click stats and returns a 303 back to the original URL. Apache/Jetty/Netty/tomcat/etc. are all fine.

Q: How are click stats stored?

• A poor candidate will suggest write-back to a data store on every click
• A good candidate will suggest some form of ==aggregation tier that accepts clickstream data, aggregates it, and writes back a persistent data store periodically==

Q: How will the aggregation tier be partitioned?

• A great candidate will suggest a low-latency messaging system to buffer the click data and transfer it to the aggregation tier.
• A candidate may ask how often the stats need to be updated. If daily, storing in HDFS and running map/reduce jobs to compute stats is a reasonable approach If near real-time, the aggregation logic should compute stats

Q: How to prevent visiting restricted sites?

• A good candidate can answer with maintaining a blacklist of hostnames in a KV store.
• A great candidate may propose some advanced scaling techniques like bloom filter.

Why lambda architecture?​

To solve three problems introduced by big data

1. Accuracy (好)
2. Latency (快)
3. Throughput (多)

e.g. problems with scaling a pageview service in a traditional way

1. You start with a traditional relational database.
2. Then adding a pub-sub queue.
3. Then scaling by horizontal partitioning or sharding
4. Fault-tolerance issues begin
5. Data corruption happens

The key point is that ==X-axis dimension alone of the AKF scale cube is not good enough. We should introduce Y-axis / functional decomposition as well. Lambda architecture tells us how to do it for a data system.==

What is lambda architecture?​

If we define a data system as

Query = function(all data)

Then a lambda architecture is

batch view = function(all data at the batching job's execution time)
realtime view = function(realtime view, new data)

query = function(batch view. realtime view)

==Lambda architecture = CQRS (batch layer + serving layer) + speed layer==

Why Use Lambda Architecture?​

To address the three issues brought by big data:

1. Accuracy (good)
2. Latency (fast)
3. Throughput (high)

For example: The problems of scaling web browsing data records in a traditional way:

1. First, use a traditional relational database.
2. Then, add a "publish/subscribe" model queue.
3. Next, scale through horizontal partitioning or sharding.
4. Fault tolerance issues begin to arise.
5. Data corruption phenomena start to appear.

The key issue is that in the AKF Scaling Cube, ==having only the X-axis for horizontal partitioning of one dimension is not enough; we also need to introduce the Y-axis for functional decomposition. The lambda architecture can guide us on how to scale a data system==.

What is Lambda Architecture?​

If we define a data system in the following form:

Query=function(all data)

Then a lambda architecture is:

batch view = function(all data at the batching job's execution time)
realtime view = function(realtime view, new data)

query = function(batch view, realtime view)

==Lambda architecture = Read/Write separation (Batch Processing Layer + Service Layer) + Real-time Processing Layer==

Relational database is the default choice for most storage use cases, by reason of ACID (atomicity, consistency, isolation, and durability). One tricky thing is "consistency" -- it means that any transaction will bring database from one valid state to another, which is different from Consistency in CAP theorem.

Schema Design and 3rd Normal Form (3NF)​

To reduce redundancy and improve consistency, people follow 3NF when designing database schemas:

• 1NF: tabular, each row-column intersection contains only one value
• 2NF: only the primary key determines all the attributes
• 3NF: only the candidate keys determine all the attributes (and non-prime attributes do not depend on each other)

Db Proxy​

What if we want to eliminate single point of failure? What if the dataset is too large for one single machine to hold? For MySQL, the answer is to use a DB proxy to distribute data, either by clustering or by sharding

Clustering is a decentralized solution. Everything is automatic. Data is distributed, moved, rebalanced automatically. Nodes gossip with each other, (though it may cause group isolation).

Sharding is a centralized solution. If we get rid of properties of clustering that we don't like, sharding is what we get. Data is distributed manually and does not move. Nodes are not aware of each other.

In a regular Internet service, the read:write ratio is about 100:1 to 1000:1. However, when reading from a hard disk, a database join operation is time-consuming, and 99% of the time is spent on disk seek. Not to mention a distributed join operation across networks.

To optimize the read performance, denormalization is introduced by adding redundant data or by grouping data. These four categories of NoSQL are here to help.

Key-value Store​

The abstraction of a KV store is a giant hashtable/hashmap/dictionary.

The main reason we want to use a key-value cache is to reduce latency for accessing active data. Achieve an O(1) read/write performance on a fast and expensive media (like memory or SSD), instead of a traditional O(logn) read/write on a slow and cheap media (typically hard drive).

There are three major factors to consider when we design the cache.

1. Pattern: How to cache? is it read-through/write-through/write-around/write-back/cache-aside?
2. Placement: Where to place the cache? client-side/distinct layer/server side?
3. Replacement: When to expire/replace the data? LRU/LFU/ARC?

Out-of-box choices: Redis/Memcache? Redis supports data persistence while Memcache does not. Riak, Berkeley DB, HamsterDB, Amazon Dynamo, Project Voldemort, etc.

Document Store​

The abstraction of a document store is like a KV store, but documents, like XML, JSON, BSON, and so on, are stored in the value part of the pair.

The main reason we want to use a document store is for flexibility and performance. Flexibility is achieved by the schemaless document, and performance is improved by breaking 3NF. Startup's business requirements are changing from time to time. Flexible schema empowers them to move fast.

Out-of-box choices: MongoDB, CouchDB, Terrastore, OrientDB, RavenDB, etc.

Column-oriented Store​

The abstraction of a column-oriented store is like a giant nested map: ColumnFamily<RowKey, Columns<Name, Value, Timestamp>>.

The main reason we want to use a column-oriented store is that it is distributed, highly-available, and optimized for write.

Out-of-box choices: Cassandra, HBase, Hypertable, Amazon SimpleDB, etc.

Graph Database​

As the name indicates, this database's abstraction is a graph. It allows us to store entities and the relationships between them.

If we use a relational database to store the graph, adding/removing relationships may involve schema changes and data movement, which is not the case when using a graph database. On the other hand, when we create tables in a relational database for the graph, we model based on the traversal we want; if the traversal changes, the data will have to change.

Out-of-box choices: Neo4J, Infinitegraph, OrientDB, FlockDB, etc.

A Bloom filter is a data structure used to detect whether an element is in a set in a time and space efficient way.

False positive matches are possible, but false negatives are not – in other words, a query returns either "possibly in set" or "definitely not in set". Elements can be added to the set, but not removed (though this can be addressed with a "counting" bloom filter); the more elements that are added to the set, the larger the probability of false positives.

Usecases

• Cassandra uses Bloom filters to determine whether an SSTable has data for a particular row.
• An HBase Bloom Filter is an efficient mechanism to test whether a StoreFile contains a specific row or row-col cell.
• A website's anti-fraud system can use bloom filters to reject banned users effectively.
• The Google Chrome web browser used to use a Bloom filter to identify malicious URLs.