System requirements

Functional:

• Search nearby POI locations
• User update POI (restaurants etc) details (Open hours, menu etc). User submit rating, reviews. 
• Users browse POI details. 

Non-Functional:

• Scalable. 
• Support high throughput API request of search POI. 
• Support high throughput of user browsing POI details. 
• User updating and submit rating is relatively small. 
• Search by text name???
• Available.  
• No single-point of failure. Reliability. 
• Minimal downtime. 
• Search latency is low. 
• Consistency. 
• Eventual. Not strong consistency. Newly added restaurants could be shown, but could have some delay. 

Capacity estimation

QPS

• 500M locations in total. DAU 100M. 
• Read and Search
• 5 searches per day. 500M / 100K = 5000 QPS. Peak 50K QPS. 
• 20 browses per day. 20 * 100M / 100K = 20K QPS. Peak 200K QPS. 
• Update / Write
• 1 update per day. 100M / 100Ks = 1K RPS to update. 10K RPS peak. 
• Growth rate 15%

Capacity

• For each location, UUID 16, lon+lat=16, rating 8, description 200 bytes, open/close, details 1Kb
• 1.5KB per location. 1.5KB * 500M = 750 GB. 
• Growth rate 10%. 

API design

• GET /search?user_location=[lon,lat]&radius=X&optional_filter={type, rating, status/open/close}&max=Y&order=asc/desc
• POST/GET/DELETE /poi?field=X&value=Y
• POST/GET/DELETE /review?poi=23idkfi&field=X&value=Y (rating, reviews)

Database design

Places Details DB [location_UUID, Geohash, lon, lat, name, type, rating, descriptions].

Index should be built on Geohash

• SQL vs NoSQL. 
• No complex query needed. 
• High performance of read & write. Growth rate, we need good horizontal scalability. 
• So NoSQL is a better choice. Like MongoDB. 
• The NoSQL will distribute the data across different shards. So each shard could support a portion of 200K QPS Read and 10K RPS Write. MongoDB has its own replication as well for high availability. 

High-level design

API Gateway / Reverse Proxy

Web server

Location services

Search services

Places Details DB

Index Service

LRU Cache with TTL

Index Building Service

Request flows

Explain how the request flows from end to end in your high level design. Also you could draw a sequence diagram using the diagramming tool to enhance your explanation...

Detailed component design

API Gateway / Reverse Proxy

• TLS termination. Decrypt the request following HTTPS. Encrypt the response.
• Directly return 503 error code, if Front-end service is slow, or down or completely unavailable.
• Load balancing. 

Web servers 

• Auth. 
• Rate limiting. 
• Request validation and early return. 
• Re-route the requests to proper services. 
• Collect metrics for monitoring & alerts. 
• Cache (TODO). 

Location Services. (do you want to dive into both? Or focus on one of these)

• Provide CRUD operations of places details. POST/GET/DELETE /poi?id=sdfildf&field=X&value=Y
• Flow. 
• Services Read DB. allocate 200 machines. 1K QPS, if one single machine is overwhelmed. 
• Services Write to DB. 
• Bottleneck is on the DB side. 
• 200K QPS read. 10K RPS write. 
• DB [UUID, lon, lat, name, type, rating, descriptions, etc]. 
• SQL vs NoSQL. 
• No complex query needed. 
• High performance of read & write. Growth rate, we need good horizontal scalability. 
• NoSQL is a better choice. Like MongoDB. 
• The NoSQL will distribute the data across different shards. So each shard could support a portion of 200K QPS R and 10K RPS W. MongoDB has its own replication as well for high availability. 
• Cache. 
• Good use case of cache. Popular area (NYC, SF) will have lots of queries. 
• Location UUID => {details}
• Write through / cache-aside trade-off. 
• Cache-aside. return cached value, if it’s there. Otherwise read from DB and update cache. Update db only + TTL. 
• Write operation, updating db only. Fast
• Stale value. 
• Cache frequently reads values. 
• Write through. return cached value, if it’s there. Otherwise read from DB and update cache. Update db + cache. + TTL. 
• Each writing operation involves mem and DB updates. 
• Many items written are not being queried. We could have a TTL for these values. 
• No data in-consistency issue.

Geohash

Project the earth into 2D grid, and then split evenly in both vertical and horizontal following the meridian and equator. Recursively we split each sub-grid and also encode as Geohash. 12-levels of geohashes. 

• From Index services: (Lon, Lat, Radius) => List of Location_UUID. 

Steps:

• user_location=[lon,lat] => geohash. This is the center of the circle and radius is specified. 9fy939fjdckd
• Based on radius=X, find the right size of geohash bounding boxes in geohash covering the circle completely. Proper size is the key, if too large, we have too many locations to search and filter. If size is too small, lots of boxes to search. 
• The best practice is to find the bounding boxes, whose height & width is smallest, but larger than radius. Then we find all the neighboring boxes at the same Grid level. The eventual 9 boxes will cover the circle.  
• Level 5 => 9fy93. 
• Surrounding boxes of specified geohash. We get 9 geohashes. 9fy94, 9fy95
•  
• If we directly search in DB, Then we could do this, 
• Select * from POI where GeoHash LIKE (9fy934%), run this for all 9 boxes identified. Indexes on Geohash. B+ tree. 
• Range scanning of the whole indexes built on top of GeoHash. Prefix is 9fy93.  
• Then we need to filter the results by rating, type, open/close and return the results. 
• Optimizations #1. 
• Most likely 4~5 surrounding boxes will have no overlap with the circle. 
• 
  
• We could calculate by finding the closest point of boxes to center of the circle. Check if it’s less than radius. 
• This will filter out half of the boxes in most cases. 
• Optimization #2.
• Query are independent. Could initiate 9 queries in parallel. (5 in parallel)
• Optimization #3. Put all indexes in the memory, by partition. 
• If we only keep [location UUID, lon, lat, Geohash, type, rating] in memory, no other details. 
• 16bytes + 16 + 8 + 16bytes +  = 60B X 500M = 30GB. Total size 45 GB => 300GB. 
• 5000 QPS. 
• Both Index size and QPS could grow, so that one machine cannot handle. Could we distribute that among multiple nodes. The index is written on disk. But we could optimize and host it in memory. That will be very fast to look up.  
• Index services: (Lon, Lat) => Geohash => 9 Geohashes => List of Location_UUID and other info. 
• Then with the list of Location_UUID, we could query places DB to collect all the details, if the in-memory data is not sufficient. 
• Partition. 
• Each non-leaf node represents an area. All the children nodes  are close to each other. Could use geohash as partition key. Assign one node and its children into one machine. Eventually it’s possible that SF is in one node and NYC is in another. 
• Each node could be like 
•          u33dbfcyeg
•         /                                                      \
•       u33dbfcyegc                                u33dbfcyegc
•      /                         \                             /                         \
• u33dbfcyegc6     u33dbfcyegcb       u33dbfcyegcc   u33dbfcyegcf
•     /                                 /                            /                                 /
•   A                            [B, E,G,H]               [C, I,L,M]                   D
• This way we distribute the Geospatial index among different servers. 
• How to find the right servers? 
• We could have separated servers to route the request. It could be a B+ tree. 
• We need to find the nodes at the proper level as partition keys.
• One the routing machine, the leaf node will have the IP address of the subtree. 
• Why Grid level 4/5/6 is proper level to assign to separated node? 
• Should be low enough, each sub-tree doesn’t have very large number of nodes. Should be high enough, Can be finished within a limited number of machines. 
• Assign a non-leaf node at Grid level 4/5/6 to be on one machine. 
• [Most searches are at level 4/5/6, one partition could finish the search. No need to merge the results from multiple machines. This number also needs some fine tuning, such that each partition does not have lots of POI, not overwhelmed. Each partition represents a logic area for easier search]
• Issues. 
• Imbalanced. Uneven distribution. NYC has more density. 
• When we build the B+ tree, the non-leaf node could track the number of POI in the area geohash maps to. 
• If it’s larger than the threshold, we could use a finer level of granularity. For example, most of the areas are at level 4, but NYC is divided into sub-areas at level 6 on different machines. 
• Benefits
• Each search is on a smaller scale. 
• Parallel search. 
• Keep them in memory, b/c it’s smaller. Not on disk. 
• This helps achieve better availability. Even if one partition died, others partition could still work. 
• Optimization #4. 
• Replicas. We could have replication of each partition. This helps achieve high availability (no single point of failure). Contribute to better performance (throughput). 
• Reduce the traffic load on one machine, reduce the possibility of hot shards. 
• Also helps balance the traffic among different machines. Example, RF=4
• M1. G1-R1, G2-R3, G3-R2
• M2. G1-R2, G2-R4, G3-R3
• M3. G1-R3, G2-R1, G3-R4
• M4. G1-R4, G2-R2, G3-R1

Trade offs/Tech choices

Explain any trade offs you have made and why you made certain tech choices...

Failure scenarios/bottlenecks

Try to discuss as many failure scenarios/bottlenecks as possible.

Future improvements

What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?