Really a great way to do things... thank you so much for sharing this.
I can't really see any disadvantages to doing anything this way. You still obtain access to data in a relational sense, though where normally you compare columns of a table you'd now do joins. For example, you can get the unique users that have submitted a link by joining the index_link table to the entities table, then joining that to the index_user table.
The problem here seems to be that sharding would prevent this type of operation... so how do they get around this? It's possible they just don't need this data, but lets assume they do. I'm presuming they have some slave(s) munging non-realtime-needed data into whatever real relational tables they choose. But, if the problem because realtime, I'm at a loss as to how they'd do it.
They explained the sharding in the text - I glossed over it on first read-through too. When they say "join" in quotation marks, they don't actually mean a join in the MySQL sense. Rather, the initial call to user_id_index.get_all reads all entity_ids for that user into the Python code (they say it consults all shards for this, but isn't the index sharded on user_id, so all entities for a given user_id live on one shard?). The Python code then uses whatever shard function applies to the entities table to query its database backends, selecting the relevant entities. Then the Python code filters the returned records by the indexed field (in case the indices are out of date) and returns it.
As for disadvantages - well, it's denormalized, for starters. ("Normalization is for sissies", says Cal Henderson.) If an indexed field changes, you need to update it in both the index and the relevant entities. There're also a bunch of little inefficiencies, places where they traded performance for scalability. Imagine if you naively plugged this engine into an app with only 10 records: instead of a simple index search, it'd have to go to the index table, fetch the relevant entities, go to the entities table, fetch them, filter on indexed value, and then return them all. But then, if your database fits on one machine, you don't have the same sort of engineering challenges FriendFeed does.
When they say it's sharded on "user_id" what they mean is that's the field that decides which database the record is stored. It might go something like: if the user starts with 0-8, store in DB1, otherwise, store in DB2. This is up to their Datastore controller to decide how to hash based on the user_id and the number of databases.
Yeah. They actually explained that in the article too - the shard number = user_id % num_of_shards. So user 1 is on DB1, user 2 is on DB3, etc. If they have 10 shards, user 11 starts back on DB1 etc.
I don't understand how that scheme can work, since changing the number of shards changes the location of most users.
e.g. we have 4 shards, so user 5 is on shard1. If we go to 6 shards, user5 is now on shard5.
I guess it works with downtime to move the users, or another layer of indirection, where the newly created shards can "point back" to existing shards, but otherwise I don't see it.
My understanding of sharding is that it's best to have a global lookup table user->shard. That's not a huge amount of data, even for millions of users.
Anyone care to educate me as to how they get user%num_shards working in practice?
Ah, OK. So you start with enough shards "for ever" and not change the number?
Edit: no wait. If you can split a shard across multiple machines, what's the benefit of having more than 1 shard? Why not have 1 shard split across 1000 machines?
The above table wasn't too clear, but my understanding is that a shard exists on only a single machine, but that a single machine could host multiple shards.
OK, but # of shards is fixed "for ever" under the modulus scheme? You pick it once, when you first shard and then you're looking at downtime to adjust it?
In order to split across multiple dbs, you're looking at creating say 100/1000 dbs in our initial split (when you've got maybe 2-3 machines). And that number then caps the number of machines you can scale to without adding another layer (sharding-shards) or having downtime?
> You pick it once, when you first shard and then you're looking at downtime to adjust it?
Yes, but you say that like it's a bad thing, it's not. If you're ever forced to reshard, that means you grew beyond what you ever hoped... hurray, awesome, nice problem to have. The reality is however, 99% chance that'll never happen.
Like 1 in a 1000 people ever run into a scaling problem of this magnitude but if you read the blogs you'd come away thinking scaling issues that require sharding are common and everyone needs this stuff, but they aren't, and they don't.
You make a large number of virtual shards. Each virtual shard is mapped to some machine. When you add a new machine, you move some shards to the new real machine.
Presumably consistent hashing is also helpful here.
I don't see how this scheme can scale simply for the reason that there's no built in balancer. What's to stop shardN from becoming overwhelmed when all the power users end up there, while shardN-1 has no activity?
In practice it's going to be very rare that you'll overwhelm a single shard and adding an extra layer to point to where people actually are is quite simple and fast. You can use a simple cache key (read-through cache of course) that, if it exists means the user is on a specific shard, overriding the default algorithmic pick.
We don't do any joins in MySQL. We query the indexes in all of the shards in parallel to get a list of entity IDs and then query the entities tables in all of the shards in paralel to get the entity bodies as a second operation.
Comments
Really a great way to do things... thank you so much for sharing this.
I can't really see any disadvantages to doing anything this way. You still obtain access to data in a relational sense, though where normally you compare columns of a table you'd now do joins. For example, you can get the unique users that have submitted a link by joining the index_link table to the entities table, then joining that to the index_user table.
The problem here seems to be that sharding would prevent this type of operation... so how do they get around this? It's possible they just don't need this data, but lets assume they do. I'm presuming they have some slave(s) munging non-realtime-needed data into whatever real relational tables they choose. But, if the problem because realtime, I'm at a loss as to how they'd do it.
They explained the sharding in the text - I glossed over it on first read-through too. When they say "join" in quotation marks, they don't actually mean a join in the MySQL sense. Rather, the initial call to user_id_index.get_all reads all entity_ids for that user into the Python code (they say it consults all shards for this, but isn't the index sharded on user_id, so all entities for a given user_id live on one shard?). The Python code then uses whatever shard function applies to the entities table to query its database backends, selecting the relevant entities. Then the Python code filters the returned records by the indexed field (in case the indices are out of date) and returns it.
As for disadvantages - well, it's denormalized, for starters. ("Normalization is for sissies", says Cal Henderson.) If an indexed field changes, you need to update it in both the index and the relevant entities. There're also a bunch of little inefficiencies, places where they traded performance for scalability. Imagine if you naively plugged this engine into an app with only 10 records: instead of a simple index search, it'd have to go to the index table, fetch the relevant entities, go to the entities table, fetch them, filter on indexed value, and then return them all. But then, if your database fits on one machine, you don't have the same sort of engineering challenges FriendFeed does.
When they say it's sharded on "user_id" what they mean is that's the field that decides which database the record is stored. It might go something like: if the user starts with 0-8, store in DB1, otherwise, store in DB2. This is up to their Datastore controller to decide how to hash based on the user_id and the number of databases.
Yeah. They actually explained that in the article too - the shard number = user_id % num_of_shards. So user 1 is on DB1, user 2 is on DB3, etc. If they have 10 shards, user 11 starts back on DB1 etc.
I don't understand how that scheme can work, since changing the number of shards changes the location of most users.
e.g. we have 4 shards, so user 5 is on shard1. If we go to 6 shards, user5 is now on shard5.
I guess it works with downtime to move the users, or another layer of indirection, where the newly created shards can "point back" to existing shards, but otherwise I don't see it.
My understanding of sharding is that it's best to have a global lookup table user->shard. That's not a huge amount of data, even for millions of users.
Anyone care to educate me as to how they get user%num_shards working in practice?
You don't change the number of shards, you change the number of machines and re-balance the shards across them. Shard != machine.
And on and on...Ah, OK. So you start with enough shards "for ever" and not change the number?
Edit: no wait. If you can split a shard across multiple machines, what's the benefit of having more than 1 shard? Why not have 1 shard split across 1000 machines?
The above table wasn't too clear, but my understanding is that a shard exists on only a single machine, but that a single machine could host multiple shards.
Correct, you don't split shards across machines. Each machine hosts x number of shards.
OK, but # of shards is fixed "for ever" under the modulus scheme? You pick it once, when you first shard and then you're looking at downtime to adjust it?
In order to split across multiple dbs, you're looking at creating say 100/1000 dbs in our initial split (when you've got maybe 2-3 machines). And that number then caps the number of machines you can scale to without adding another layer (sharding-shards) or having downtime?
> You pick it once, when you first shard and then you're looking at downtime to adjust it?
Yes, but you say that like it's a bad thing, it's not. If you're ever forced to reshard, that means you grew beyond what you ever hoped... hurray, awesome, nice problem to have. The reality is however, 99% chance that'll never happen.
Like 1 in a 1000 people ever run into a scaling problem of this magnitude but if you read the blogs you'd come away thinking scaling issues that require sharding are common and everyone needs this stuff, but they aren't, and they don't.
Yes. In the simple modulo scheme, the number of shards is fixed. Unless you want to re-shard the entire thing. (AFAIK)
Instead of using normal hashing you can use consistent hashing so you don't need to move that many users. Here's a post explaining it: http://www.spiteful.com/2008/03/17/programmers-toolbox-part-...
You make a large number of virtual shards. Each virtual shard is mapped to some machine. When you add a new machine, you move some shards to the new real machine.
Presumably consistent hashing is also helpful here.
I don't see how this scheme can scale simply for the reason that there's no built in balancer. What's to stop shardN from becoming overwhelmed when all the power users end up there, while shardN-1 has no activity?
Edge-Case YAGNI. Have lots of shards, put multiple shards on each machine, shuffle the shards around as necessary. Rinse. Repeat.
flickr deals with that problem by manually moving power users if necessary. They have a tool that lets them do that. See: http://highscalability.com/how-flickr-handles-moving-you-ano...
In practice it's going to be very rare that you'll overwhelm a single shard and adding an extra layer to point to where people actually are is quite simple and fast. You can use a simple cache key (read-through cache of course) that, if it exists means the user is on a specific shard, overriding the default algorithmic pick.
Hopefully they simplified things for the sake of brevity and they're using consistent hashing or some other predictable algorithm in practice.
We don't do any joins in MySQL. We query the indexes in all of the shards in parallel to get a list of entity IDs and then query the entities tables in all of the shards in paralel to get the entity bodies as a second operation.
But would you, if MySQL could do hash joins?