The Architecture Twitter Uses to Deal with 150M Active Users

http://highscalability.com/blog/2013/7/8/the-architecture-twitter-uses-to-deal-with-150m-active-users.html

by aespinoza • 13 years ago

316 162 13 years ago

highscalability.com

bpicolo• 13 years ago

They have a lot of RAM. Dang.

buro9• 13 years ago

Not as much as I would have thought.

2TB can be one server: http://www.supermicro.co.uk/products/system/5U/5086/SYS-5086...

An expensive server for sure... but it's just one server.

vidarh6• 13 years ago

In case people have a hard time finding prices for this, my vendor gives ~57k GBP ($85k) as the lower end price for configurations with that SuperMicro cabinet, 64 cores (8x 8-core CPU's) and 2TB of RAM. Dropping to 1TB lets you pick from a lot of cheaper cabinets and so the price drops by quite a bit more than half for the basic configurations.

papsosouid14• 13 years ago

Even the low end of single servers goes up to a couple TB of RAM. "Enterprise" hardware has been there for ages.

binarycrusader2• 13 years ago

That's really not a lot, especially when you consider that it's distributed among a cluster.

A single SPARC M5-32 can have up to 32TB of memory:

http://www.oracle.com/us/products/servers-storage/servers/sp...

druiid• 13 years ago

As others have said they don't actually have a lot of RAM dedicated to Redis. You can put 1.5TB of memory in many 'inexpensive' Dell servers (single machines, not clusters). So basically a cabinet of machines you could have over 30TB of memory available to you. Basically some of the design choices Twitter has made seem tailored to their (previous? Not sure if they have their own hardware now) choices to run on 'cloud' services in the past.

With good hardware and a bit of a budget you can easily scale to crazy numbers of processor cores and memory. That isn't to say the software side of the solution is going to be any easier to solve.

NelsonMinar• 13 years ago

This article cites its source as a talk by Twitter VP Raffi Krikorian. 38 minutes of video, audio, and slides are at http://www.infoq.com/presentations/Twitter-Timeline-Scalabil...

aespinoza• 13 years ago

Thank you for sharing it.... I was wondering about the video.

jmstout• 13 years ago

I also enjoyed this presentation: http://www.infoq.com/presentations/Timelines-Twitter

It goes into more depth about how they handle timelines.

webwanderings2• 13 years ago

Twitter knows a lot about you from who you follow and what links you click on.

No kidding. But we don't care as we live in the glass house of a celebrity culture.

PS: downvote the quoted text if you must. My point is not obvious.

mjolk• 13 years ago

If you expect others to misunderstand you, perhaps try to better represent your thoughts?

StavrosK• 13 years ago

You realize it's public data, right? When I tweet something, I don't expect it to be private. HN knows a lot about me too.

webwanderings2• 13 years ago

Yup, I mentioned glass houses.

Myrmornis3• 13 years ago

The meaning of your metaphor certainly seems obvious; explicit even. Are you referring to a layer of more profound non-obviousness?

brown9-24• 13 years ago

Here is a similar talk that Jeremy Cloud gave at QCon NY a few weeks ago: http://www.infoq.com/presentations/twitter-soa

Jeremy Cloud discusses SOA at Twitter, approaches taken for maintaining high levels of concurrency, and briefly touches on some functional design patterns used to manage code complexity.

mistertrotsky• 13 years ago

I would absolutely trust someone named Jeremy Cloud on this subject.

papsosouid14• 13 years ago

I really question the current trend of creating big, complex, fragile architectures to "be able to scale". These numbers are a great example of why, the entire thing could run on a single server, in a very straight forward setup. When you are creating a cluster for scalability, and it has less CPU, RAM and IO than a single server, what are you gaining? They are only doing 6k writes a second for crying out loud.

pdog• 13 years ago

How do you expect to serve 150 million concurrent users on a single server?

papsosouid14• 13 years ago

They are not serving 150 million concurrent users. They have 150 million active users. As in, people who do in fact use twitter at all, as opposed to the millions of dead accounts nobody touches. They are not all being served at once.

colinhowe2• 13 years ago

That's probably 6k writes to a "normal" DB. The fanout is handled by Redis which I doubt is included in the 6k writes. Not if you want to fan out to 30 million followers in under 5 seconds.

papsosouid14• 13 years ago

The "fanout" or "wasteful duplication of a single message 30 million times" is only required because they are using tiny underpowered hardware to begin with. The approach that they claim can't possibly work actually does work. You just can't do it on a $2000 "server".

spullara2• 13 years ago

I hope this comment is a troll and you don't actually believe that you could run Twitter on a single server. If you are interested in why this is true I suggest you get a job at a company that has to serve consumer scale web traffic.

papsosouid14• 13 years ago

The fact that you think "consumer scale web traffic" is some magical thing is exactly what I am talking about. Have you ever heard of TPC? They have done benchmarks of database driven systems for a long time. TPC-C measures performance of a particular write query, while maintaining the set ratio of other active queries. The top non-clustered result right now does 142,000 new orders per second. Yes, a single server can handle 300k reads and 6k writes per second.

http://www.tpc.org/tpcc/results/tpcc_perf_results.asp?result...

LewisJEllis• 13 years ago

It's not as simple as 6k writes - it's 6k write requests. Most of these 'write requests' are tweets which must be written to the timelines of hundreds or thousands (or in some cases millions!) of followers.

If you try to run it all on a single server, and Obama and Justin Bieber tweet at the same time, you suddenly have a backlog of 75 million writes to catch up on. Now imagine what happens if Bieber starts a conversation thread with any of the other 50 users with 10m+ followers.

vidarh6• 13 years ago

Agreed that this is an issue, but it's a smaller one than you make out: Consider that there's likely a marked hockey stick going on here. Now keep the very most recent tweets of the very "worst" in terms of followers in a hot cache, and mix their most recent tweets into the timelines of their followers as needed until they're committed everywhere. Then collapse writes, and you reduce the "conversation" problem.

It's still not cheap to handle, but consider that fan-out on write is essentially an optimization. You don't need that optimization to be "pure" in that you don't need to depend on the writes being completed in a timely fashion - there are any numbers of tradeoffs you can make between read and write performance.

nostrademons3• 13 years ago

They create big, complex, fragile architectures because they started with simple, off-the-shelf architectures that completely fell over at scale.

I dunno how long you've been on HN, but around 2007-2008 there were a bunch of HighScalability articles about Twitter's architecture. Back then it was a pretty standard Rails app where when a Tweet came in, it would do an insert into a (replicated) MySQL database, then at read time it would look up your followers (which I think was cached in memcached) and issue a SELECT for each of their recent tweets (possibly also with some caching). Twitter was down about half the time with the Fail Whale, and there was continuous armchair architects about "Why can't they just do this simple solution and fix it?" The simple solution most often proposed was write-time fanout, basically what this article describes.

Do the math on what a single-server Twitter would require. 150M active users * 800 tweets saved/user * 300 bytes for a tweet = 36T of tweet data. Then you have 300K QPS for timelines, and let's estimate the average user follows 100 people. Say that you represent a user as a pointer to their tweet queue. So when a pageview comes in, you do 100 random-access reads. It's 100 ns per read, you're doing 300K * 100 = 30M reads, and so already you're falling behind by a factor of 3:1. And that's without any computation spent on business logic, generating HTML, sending SMSes, pushing to the firehose, archiving tweets, preventing DOSses, logging, mixing in sponsored tweets, or any of the other activities that Twitter does.

(BTW, estimation interview questions like "How many gas stations are in the U.S?" are routinely mocked on HN, but this comment is a great example why they're important. I just spent 15 minutes taking some numbers from an article and then making reasonable-but-generous estimates of numbers I don't know, to show that a proposed architectural solution won't work. That's opposed to maybe 15 man-months building it. That sort of problem shows up all the time in actual software engineering.)

papsosouid14• 13 years ago

They create big, complex, fragile architectures because they started with simple, off-the-shelf architectures that completely fell over at scale.

No, they fell over at "shit we're hitting the limits of our hardware, lets re-architect everything instead of buying bigger hardware". Rather than buy 1000 shitty $2000 servers, buy 2 good $1,000,000 servers. I know it is not fad-compliant, but it does in fact work.

contingencies• 13 years ago

You are right to be sceptical, and I think you're right about throughput: they mentioned that all tweets are ~22MB/sec. My hard disk writes 345MB/sec. A modern multicore processor should probably tear through that datastream.

However, you should realize that in a decent sized organization there are usually different teams working on different things that are evolving at different rates. You have to manage these people, systems and their changes over time. After awhile, efficiency drops in priority versus issues such as manageability and security... often due to separation of concerns requirements.

Beyond the build and maintenance issues, a serious challenge for real time services is high availability (HA). Tolerance for hardware/software/network/human failings must be built in. That also throws efficiency out the window. It's cheaper to treat cheap, easily replaceable machines as a service-oriented cluster than to build two high performance machines and properly test the full range of potential failover scenarios.

xb95• 13 years ago

I think that in a lot of ways it's because of the roots of the organizations in question. Instead of saying "yeah, we have money, we can go buy this solution and trust in the vendor to support us" (an objectively viable solution in many situations) we instead say "we've got $100,000 in seed money and one engineer plus some friends she's convinced to help out pro bono, that won't buy us a sideways glance from Oracle".

Once you've survived the first year or two and gained some traction, suddenly the decision to investigate other solutions becomes a break from tradition -- and no engineer wants to throw away their work. It's incredibly hard to do a full stop, look around, and reevaluate your needs to see if you should just go in a different direction. I'm not sure I've ever seen it happen, really; we always just bow to the organic growth and inertia that we've established.

There is also a strong tradition of Not Invented Here (NIH) at work in this ecosystem. If it wasn't written by one of your peers (other startup people) then it's just not very cool. Use a Microsoft or Oracle product? Hahaha, no. Open source or bust!

...

To be fair, though, I've never worked for a startup that could afford that kind of software, so I guess the point is moot. I'm not spending 20% of my startup's cash reserves on software licenses to Oracle, when I can instead use that $2m to hire a few people to build it for me and know it inside and out. Plus, then they're invested in my company.

Also, please don't think I'm denigrating open source software with this commentary. I just think that the kind of zealotry that precludes even considering all of the options is, in general, a bad business decision, but one we seem to make all of the time.

sk00byd00• 13 years ago

Your expensive vendors sit on security patches for weeks and my free ones shoot them out almost immediately. Why is that not a business concern?

I don't think it's zelotry, there are a lot of advantages to open source software you aren't going to get from Microsoft. If you are an edge case and you happen to stumble on that race condition bug are you going to have your engineers black box test and reverse engineer someone elses product illegally while they wait around for Microsoft support or have them look under the hood, patch the bug and move on?

What the big licenses fees get you is accountability, which is of course a huge thing, but what open source gives you is control.

I think if your business is software, then open source makes perfect business sense, especially if one of your assets is a team of competent engineers.

adient• 13 years ago

You're arguing for scaling up (vertical) instead of scaling out (horizontal). Both are valid approaches. Scaling out is preferred because your architecture is mode modular and you do not have to constantly buy bigger machines as your usage grows; you just add additional machines with similar capacity. The main problem with scaling up is that price and performance are not linearly related and eventually you will be limited by the performance available to one system. But it's a perfectly valid approach for certain scenarios.

numbsafari• 13 years ago

Because if you want to have reasonable availability and low latency you can't just buy two "big" machines.

Going with the incrementally sized approach gives you more flexibility to economically distribute the risk, in addition to the load.

thezilch5• 13 years ago

They may average 6K w/s, but history has shown higher peaks, and those peaks often include tweets from those with large 30e6+ followers -- large events naturally have celebrities chiming in. Having more CPU [cores], RAM, and IO is not going to solve for the need to shard your Redis, which is bound by a single core. For the fanout, that is anywhere from 90 to 90e3 cores at 2e6 Redis IOPs -- 1-6e3 Lady Gagas per second. Before pipelining, which you'd probably consider complex or fragile. All you've done is moved some networking parts around. Your code and topology hasn't really changed.

papsosouid14• 13 years ago

Having more CPU [cores], RAM, and IO is not going to solve for the need to shard your Redis, which is bound by a single core

But they chose redis in the first place as part of their messy, crummy pile of barely working junkware solution. You wouldn't use redis in the first place if you were building a solution that scales up.

brown9-24• 13 years ago

They are only doing 6k writes a second for crying out loud.

And 600k reads per second (as of when this article was written), which seems like an important thing to leave out.

papsosouid14• 13 years ago

300k reads, which is approaching trivial. You can do that with an array of desktop SSDs. But in reality, it would be almost all in RAM anyways making those reads incredibly cheap.

dclusin• 13 years ago

The property of distributed message passing systems that everyone likes is the arbitrary scalability. Subsystem performing slowly? Add more nodes! The cost of this arbitrary scalability, as you have noticed, is brittleness, single points of failure, and complexity that will drive even the most seasoned of engineers nuts.

In the end though, the users want their tweets and they don't care how it works. Complexity is why we make so much money, after all.

davidgerard2• 13 years ago

I do consider it a pathology when tiny services, or tiny apps in a corporate structure, act like they have the problems of Google.

You are not Google.

You do not have Google's problems.

You do not have scaling issues.

For you, N is small and will stay small.

Stop giving me this delusional resume-padding garbage to implement. For you, here, it is delusion and lies.

jacques_chester6• 13 years ago

The point here is that Twitter really is one of the cases where vertical scaling is not, on the balance of non-functional requirements to engineering overhead, the right decision. They really do need to pay the complexity piper.

norkakn• 13 years ago

I remember asking a question about scaling on SO and getting a response about how my data was small, because it could fit on an SSD (for that one small component)

I miss not having real scaling issues sometimes. They make my head hurt.

mwcampbell• 13 years ago

Maybe Twitter would indeed be better off using a smaller number of more beefy servers. But at the software level, I wonder if the architecture would be very different.

Suppose you implemented something like Twitter as a single large application, running in a single process. Maybe run the RDBMS in its own process. The app server stores a lot of stuff in its own in-process memory, instead of using Redis or the like. Now, what happens when you have to deploy a new version of the code? If I'm not mistaken, the app would immediately lose everything it had stored in memory for super-fast access. Whereas if you use Redis, all that data is still in RAM, where it can be accessed just as fast as before the app was restarted.

wmf• 13 years ago

You'd use failover to do a rolling upgrade.

wprl• 13 years ago

Two thoughts: scaling down can be just as important as scaling up, and cluster nodes that need to talk to each other are probably less common than stateless cluster nodes that need to talk to the client.

lonelybug• 13 years ago

Something is not easy to say simple or complex by just looking at the shape or counting the number of them.

One big powerful server's architect and circuit design is not that simple as what it looks like, in the other hand, 2000 standard servers are not that complex as what they are.

The way of how you think is the key. You can think that the 2000 servers is a big computer cluster, but I prefer to think each one of them is a simple replaceable black box.

I can show you some scenarios at the following to explain why 2000 servers make more sense than one big machine,

Firstly, When I want to upgrade the system to deal with a suddently increased load I don't need to call vendor to arrange an onsite upgrade service, I can do it by connecting more servers right away, and when the load decreased, then I can disconnect some servers. This makes the maintenance job is more flexible and more convenient.

The second one is when I estimate the performance of the whole system, I can focus on estimating the performance for one server first, and then add them on, even there are some variables need to be involved into the calculation, it is still more clear and simpler than estimating the performance by looking at the vendor provided system specification.

The last scenario is that you cannot separate one big machine into different geolocations to handle the access all over the world, but seperate 2000 servers is much easier and doable. Moreover, different geolocation deployment can provide a genric 24*7 online service as some nature disasters happens.

When we engineer things, we should split one big(complex) thing into multiple small parts and keep it with a simplest structure or function, and later, we connect these small parts together.

English is not my first language, so if somewhere sounds wired, please excuse me.

cwt137• 13 years ago

This near hour long video is a deep look into Twitter's backend. Specially into the Firehose feature, Flock, etc. They go into detail on how they use Redis and even show one of the actual data structures they store in Redis. A must see video for anyone into high scalability.

http://www.infoq.com/presentations/Real-Time-Delivery-Twitte...

aroman• 13 years ago

Twitter no longer wants to be a web app. Twitter wants to be a set of APIs that power mobile clients worldwide, acting as one of the largest real-time event busses on the planet.

Wait, then why are they actively destroying their third-party app ecosystem...?

grbalaffa• 13 years ago

Yeah, somebody hasn't gotten the memo it seems.

blazingfrog2• 13 years ago

The piece did not say "third party". I'm assuming these clients are the official Twitter mobile apps.

mey• 13 years ago

The two are not the same. It sounds like they are becoming API driven to support multiple interfaces and to put clear boundaries between systems.

zanny• 13 years ago

Sounds like html. Or qt. Not 140 characters.

Yhippa• 13 years ago

Probably their mobile clients. After I read the article perhaps Twitter felt that having control of the clients reduces variability in stress to their system. Like bad actors accessing the firehose suboptimally.

mikemoka• 13 years ago

I am playing the armchair architect and my question will be probably wrong in infinite ways,but I might learn something, what is the reason why the service has to write a tweet on two million timelines, wouldn't it be cheaper if they let the client build the page on its own via restful apis?

rythie4• 13 years ago

Requesting the timelines of each of the people you follow is slow process for the client, meaning that the client has to make 100s or even 1000s or requests.

Also, much of the data is thrown away because it's replies to people you don't follow or too old. It's hard for clients to reconstruct the timeline that way. Also, it would vastly increase the number of HTTP requests and data Twitter has to ship out.

Also, it's basically impossible to make a system like that real time because you have to check 1000 feeds to see if anything is new.

I used to write a multi-network client, basically the combined home timeline request is the only feasible method for a client to use.

brown9-24• 13 years ago

They are making a tradeoff between cost of reading your timeline and cost of a write at tweet-time to optimize the former.

logic2• 13 years ago

In addition to what others have mentioned, consider passive endpoints, such as SMS and push notifications.

beagle33• 13 years ago

But they don't all have to fit within the same framework - indeed, don't fit into the same framework.

e.g., if you can't send the SMS or push right now, you have to do a retry 5 minutes later. If you can't update a followers queue (if that's what you do), you have to display a fail whale....

harryh2• 13 years ago

That is actually a very interesting question and it turns out that whether it's better to fanout on write or on read depends on a few different things. There's a very widely read paper on the subject you might enjoy:

http://research.yahoo.net/node/3203

sumzup• 13 years ago

The link doesn't seem to work (error: "Unable to connect to database server"); do you have an alternate link?

Myrmornis3• 13 years ago

Title? Link doesn't work.

beagle33• 13 years ago

Mostly because they chose the wrong backend technology (which they have been doing repeatedly since their early days).

The right way to solve twitter would be to have 140-byte tweets sorted by a <userid,time> 64-bit key, with a few more attributes (all falls into 256-bytes neatly), shard them across servers and keep everything recent in memory.

Logging into a server would fetch the list of following to the front end server, broadcast the request to all tweet servers, wait 50ms or so for responses, merge, sort and HTML format them.

The front end servers would not need any memory or disk (could be an army of $500 servers behind a load balance, or a few beefy ones). The backend servers would have to have some beefy CPU and memory, but still ultra commodity (256 bytes/tween means 1GB=4M tweets, so one 64GB server=256M tweets). Shard for latency, redundancy, etc. Also, special case the Gagas/Kutchers of this world by giving them their own server, and/or have them broadcast to and cache their tweets in the front end servers (Spend 256MB memory on tweet cache in the front end servers, and you get 1M cached tweets - which would cover all of the popular people and then some).

Network broadcast was invented for a reason.

rythie4• 13 years ago

My understanding is that Facebook does it that way. I think for Twitter it creates two problems though:

- It's hard to make that realtime, because you'd need to some sort of broadcast every second or at least every few seconds.

- It's hard to follow a lot of people (you could be getting a large number of replies), so there would need be a follow limit.

Facebook has a follow limit and people don't really expect Facebook to be realtime - the central bit is not really.

Also, there is no one right way to do these things, in my view.

davidw• 13 years ago

Your home timeline sits in a Redis cluster and has a maximum of 800 entries.

Wow, that's pretty cool. Congrats to antirez - it must be a nice feeling knowing that your software powers such a big system!

nasalgoat5• 13 years ago

Redis is probably the most useful tool powering the internet after nginx. It really is an amazing piece of engineering.

smandou2• 13 years ago

You forgot the JVM.

herge2• 13 years ago

Memcache?

wdewind4• 13 years ago

I have not used nginx, and have never heard of it spoken about in such glowing terms. Can you maybe compare/contrast it with apache?

antirez• 13 years ago

Thank you David, I'm very happy indeed that many companies are using Redis to get work done, this is basically one of the biggest reasons why after 4 years I'm not giving up...

As you know I'm usually not the kind of guy focused to the task at hand for more than a limited timeframe, and then I've the temptation to switch to something else, but this time I'm finding the right motivations in the big user base.

ampersandy• 13 years ago

No one else seems to have noticed that these details are out of date. Twitter has publicly stated that they currently have "well over 200M active users". The stats are also misleading in that -- I'm pretty sure -- 300k reads and 6k writes per second are only referencing tweets. Flock, Twitter's graph database, handles more than 20k writes and 100k reads per second on its own (peak numbers available from the two year old Readme on Github).

https://blog.twitter.com/2013/celebrating-twitter7

smandou2• 13 years ago

Well... It seems that hisghscalability doesn't scale... Error from squarespace on that page.

kushti• 13 years ago

Scala is awesome language to implement scalable things, like Twitter services

swah• 13 years ago

Why?

pesenti• 13 years ago

Does anybody know how that compares to Facebook? I believe they do not use write fanout but instead rely on the search federation model (the model claimed not to have worked for Twitter).

sheldoan3• 13 years ago

Facebook uses a fan-out-on-read approach

http://www.quora.com/Facebook-Engineering/Did-Facebook-devel...

apaprocki4• 13 years ago

Since tweets are very much like financial ticks (fixed size, tiny), we can put the numbers in perspective. Let's compare Twitter vs say, OPRA (North American options), a single "exchange" feed.

Twitter: 300K QPS, firehose 22 MB/sec, 400 million tweets/day

OPRA: 12.73M MPS, firehose 3.36 GB/sec, 26.4 billion messages/day

http://opradata.com/specs/2012_2013_Traffic_Projections.pdf

edit: Also worth noting, the diff between OPRA 1/1/13 & 7/1/13 is ~256 MB/sec, 2.0 billion messages/day. So in just 6 months the existing firehose increased roughly 10x Twitter's entire output and roughly 5x'd the number of messages/day.

squarecog2• 13 years ago

It's not the data coming in that makes things hard. It's the fanout of messages into their subscribers' timelines.

ak39• 13 years ago

I've read before that Twitter uses NoSQL (or non-relational databases) and that each tweet requires a copy to be pushed to each follower every time someone tweets (same for updates or deletes). I am of a very much traditional RDBMS mindset though I am willing to bend a bit :) ... but this idea of fanning out tweets (copying tweets) to each and every follower just sounds grossly inefficient. The amount of writes must be exponential.

The way I see it, here are the WRITES they'll ever need:

1:User -> n:Users (Followers) (Temporal links with effective dates for follow history)

1:User -> n:Tweets (Temporal with tweet datetime) - this includes Retweets.

The rest are just READS (pulls instead of fanning out the same data).

What am I missing here folks? :-) Educate me please.

thezilch5• 13 years ago

Tweets are fanned out to more than a single feed and, in the "most important" cases, millions of feeds -- 31 million feeds for Lady Gaga. Your comparing 300K reads, which might not even include the sub-queries, to 12.7M writes. A single tweet from a single user would trump the writes; it's unclear whether it could be done in the same one-second window.

Twitter already does 30B Timeline deliveries a day, compared to the 26.4B of OPRA. Again, there is no telling what can be implied by a "Timeline delivery;" does it include pulling user and other secondary and tertiary objects? An HTML renderer? The doesn't say much for the capabilities and focuses on what they do.

There's also little comparison to be made on what powers OPRA. If Twitter can simply add nodes to their architecture, are they doing it wrong?

apaprocki4• 13 years ago

Securities work a similar same way -- When you're looking at your portfolio you're only directly monitoring the individual fields for the securities in your portfolio, not the entire firehose. The OPRA feed is an equivalent to the raw Twitter firehose, only much fatter. Once you integrate it into the last-mile display to the users, you're doing all the same things. This is typically done with multicast topic-style subscriptions by ticker / type of data. It makes sense that if Lady Gaga writes a tweet you only want it to be "1 timeline delivery" (multicast) as opposed to "31 million timeline deliveries", which is making your infrastructure do a lot more work. Granted, you're kind of limited by what the browser can do in this regard, so you're kind of stuck with the socket model.

dchichkov• 13 years ago

300K QPS is pretty impressive. If you'd want to sustain that rate on a single modern machine you'll have to fit all your processing into about thirty LLC misses per request.

apaprocki4• 13 years ago

I admit the QPS/MPS thing doesn't really fit. I mainly wanted to illustrate the data rates of the raw feeds. Twitter has one of the largest raw feeds of the web world, but it's useful to know how that compares to other non-web feeds.

jacques_chester6• 13 years ago

Figuratively, Twitter have switched from doing a design rooted in Databases 101 (OLTP) to a design more rooted in Databases 102 (OLAP).

That is, they moved processing from query time to write time. And that's a perfectly legitimate strategy; it's the basis of data warehousing.

OLTP is about write-time speed. It's great for stuff like credit card transactions, where you only really care about the tally at the end of the banking day. The dominant mode of operation is writing.

OLAP is about read-time speed. You do a bunch of upfront processing to turn your data into something that can be queried quickly and flexibly.

One thing that's problematic about the teaching of database technology is that the read/write balance isn't explicitly taught. Your head is filled with chapter and verse of relational algebra, which is essential for good OLTP design. But the problems of querying large datasets is usually left for a different course, if it's treated at all.

joshuaellinger3• 13 years ago

Maybe at a high level, but OLAP involves precomputing aggregations and bit indexes. It's a pretty different beast.

OLAP is very rarely under real-time constraints and, when it is, it tends to push the heavy lifting out to OLTP.

jacques_chester6• 13 years ago

Oh, I'm handwaving, not talking about the underlying details of star schemata, clever column representations and whatnot.

But I think the analogy is still correct.

Every such system has two functional requirements:

1. Store data.

2. Query data.

And every system has the same non-functional requirement:

1. Storage (write) should be as fast as possible.

2. Queries (read) should be as fast as possible.

However, per an observation I made a while back, complexity in the problem domain is a conserved value.

Insofar as your data requires processing to be useful, that complexity cannot be made to go away. You can only decide where the complexity cost is paid.

You can pay it at write time and amortise that across reads. You can pay it at read time and excuse writes. Or you can pay it in the middle with some sort of ETL pipeline or processing queue.

But you must always pay it. The experiences of data warehousing made that bitterly clear.

So really, the job of a software architect is to take the business requirements as a non-functional requirement (an -ility) and then pick the architecture that fits that NFR. That includes dropping other nice-to-have non-functionals.

Twitter's non-functional requirement is that they want end-to-end latency to be 5 seconds or less, under conditions of very low write:read ratio. This suggests paying the complexity cost up front and amortising it over reads. And that's what they've done.

squarecog2• 13 years ago

That's an interesting thought, but rather than focusing on write vs read balance, I would claim that OLAP and OLTP are most distinguished by the nature of queries they need to support.

OLAP is characterized by fairly low-volume aggregation queries that touch very large volumes of data ("what fraction of tweets in English come from non-english speaking countries?"). Sure, the ingest is large, but it tends to be batch and has fairly loose real-time (in the EE sense) requirements. What makes OLAP hard is the sheer volume of data from which an aggregation must be calculated.

OLTP is characterized by very selective queries with much tighter real-time bounds ("what are the last 20 things that the 100 people I follow said?"). The overall size of the dataset might even be the same, but each individual query needs fast access to a tiny fraction of the dataset. In many applications, this is accompanied by very high QPS and in Twitter's case, extremely high write volume.

jacques_chester6• 13 years ago

Thanks, I hadn't mentally partitioned OLAP/OLTP that way. I still think that the read/write distinction more correctly forms the dividing line.

That's because the classical OLTP approach is to use relational databases. Relational databases do well in writes because in a properly normalised DB there is one, and only one, place for each datum to go. There is no fan out and potentially no coordination. A place for everything and everything in its place. That reduces the amount of write traffic required to support complex models of the problem domain by a lot.

But of course the relational view of a problem domain doesn't really look like the humanistic view of a problem domain. And building the humanistic view of the problem domain usually means denormalising and breaking the things that made OLTP useful. Enter ETL pipes and OLAP systems.

From there it's very humanistic and it requires no understanding of the relational model. Tools can easily turn the dimensional tables into dropdown filter lists that look like a spreadsheet -- or even run inside a spreadsheet.

jebblue2• 13 years ago

> it can take up to 5 minutes for a tweet to flow from Lady Gaga’s fingers to her 31 million followers

Why not break the load up among a farm of servers? 5 minutes to deliver a single message? It's too bad multicast can't be made to work for this use case.

At least analyze to see if there's a pattern of geographic concentration of her followers and optimize for where their datacenters are.

Use peer to peer, let the clients help distribute the messages.

joshuaellinger3• 13 years ago

The surprise for me is that the core component is Redis.

My first guess would have been custom C code. Yeah, you have to do everything yourself. Yeah, it would be hard to write. But you'd control every little bit of it.

Obviously, I must not fully understand the problem and what Redis buys.

Sam Puralla (if you are reading) -- do you know why didn't Twitter go with a full custom system at its heart?

Josh

nikcub• 13 years ago

Think of Redis as a nicer interface to data structures that otherwise would have been written in custom C code.

If you read the redis manifesto, it describes itself in point 1 as a DSL for abstract data types:

http://oldblog.antirez.com/post/redis-manifesto.html

bitbckt2• 13 years ago

I'm probably a better person to answer this than Sam - I'm a former lead on this project - so I'll take a swing:

We chose Redis because it gave us the specific, incremental improvements over our existing memcached-based system that we required, without requiring us to write (yet another) component. There was enough to do, and this choice has turned out to be good enough, I think.

As the project progressed though, we treated Redis much like we did own it. We altered the wire protocol, changed the eviction strategy, and reduced protocol parsing overhead, for example. Much of that work has long since made it upstream.

[edit] grammar

joshuaellinger3• 13 years ago

Sounds like good engineering choices. I think of Twitter as having unlimited resources but, of course, that can't be true.

Two follow-on questions:

1. Did your changes make it back into open source or were they only relevant to Twitter? When you say upstream to you mean on Redis or earlier in the Twitter pipeline?

2. How much is Redis on the critical path? Is it 90% of the processing work in the large fanout cases?

_pmf_• 13 years ago

The surprise for me is that the core component is Redis.
My first guess would have been custom C code.

I'm pretty sure they use a heavily tailored Redis, so both is true.

tschellenbach• 13 years ago

Their setup is very similar to what we use at Fashiolista. (though of course we only have millions and not hundreds of millions of users). We've open sourced our approach and you can see an early example here: https://github.com/tschellenbach/Feedly/

The Architecture Twitter Uses to Deal with 150M Active Users

Related Stories