Large-scale cluster management at Google with Borg(blog.acolyer.org)
blog.acolyer.org
Large-scale cluster management at Google with Borg
http://blog.acolyer.org/2015/05/07/large-scale-cluster-management-at-google-with-borg/
32 comments
Try Mesos/Marathon in addition to Kubernetes. Been running Kubernetes on Mesos and it's just great.
Swarm isn't shabby either.
Can anyone throw some light on what google is doing with tens of thousands of machines in these clusters? Are they hosting crawlers on them?
They host pretty much everything -- Maps, Gmail, Search, etc. Highly suggest reading the Wired link below as it talks about this in detail [1], but if you want a more technical look, check out the Youtube video [2], or the Borg paper [3]. One really interesting development, is that Mesos + Marathon + Chronos + Docker, puts a very similar system into the hands of your average IT team today [4].
[1] http://www.wired.com/2013/03/google-borg-twitter-mesos/
[2] https://www.youtube.com/watch?v=0ZFMlO98Jkc
[3] https://research.google.com/pubs/archive/43438.pdf
[4] https://www.youtube.com/watch?v=hZNGST2vIds
[1] http://www.wired.com/2013/03/google-borg-twitter-mesos/
[2] https://www.youtube.com/watch?v=0ZFMlO98Jkc
[3] https://research.google.com/pubs/archive/43438.pdf
[4] https://www.youtube.com/watch?v=hZNGST2vIds
- crawlers
- websearch
- google brain
- video transcoding for youtube
- appengine (think snapchat)
- gmail
Almost everything Google does runs on Borg.
- websearch
- google brain
- video transcoding for youtube
- appengine (think snapchat)
- gmail
Almost everything Google does runs on Borg.
When I ran Exacycle http://googleresearch.blogspot.com/2013/12/groundbreaking-si... the worker pool was just a low-priority (priority 0 in the paper) job like everything else, soaking up idle cycles. The command and control jobs were just regular borg jobs that managed the worker pools by assigning tasks to them.
So whatever else was going on, protein folding and drug docking and telescope simulation was soaking up whatever leftover cycles there were.
So whatever else was going on, protein folding and drug docking and telescope simulation was soaking up whatever leftover cycles there were.
Can anyone explain how MapReduce is built on top of Borg, when Borg doesn't support scheduling with data locality? I'm having a hard time reconciling that. One of the principal use cases of the MapReduce framework is reducing data movement...
From the MapReduce paper: "The MapReduce master takes the location information of the input files into account and attempts to schedule a map task on a machine that contains a replica of the corresponding input data."
One of two possible designs immediately come to mind: (1) The MapReduce master interacts with Borg directly on behalf of the programmer and places a requirement on the task to run on or near any node that has a replica of the data required or (2) for each Borglet, there is an accompanying MapReduce agent running to accept RPCs from the MapReduce master.
One of two possible designs immediately come to mind: (1) The MapReduce master interacts with Borg directly on behalf of the programmer and places a requirement on the task to run on or near any node that has a replica of the data required or (2) for each Borglet, there is an accompanying MapReduce agent running to accept RPCs from the MapReduce master.
It's the former.
You seem to be suggesting that using local i/o to read inputs would be preferable to using remote i/o, but I don't think that's a supportable statement in a datacenter with a high-performance network.
But it's a fact that original MapReduce/Borg used data locality (and networks weren't as fast 15 years ago).
Yes but time marches on. The amount of compute power available on a commodity server node has grown far faster than the available disk io time.
What makes you think the data is on disk?
No, seriously.
Picture if all of your data is in RAM, somewhere. If you can move your program to that machine quickly, then that's a huge win.
No, seriously.
Picture if all of your data is in RAM, somewhere. If you can move your program to that machine quickly, then that's a huge win.
It just doesn't seem very likely to me. It seems a lot more likely that the data would be spread around on thousands of individual hosts, one megabyte at a time. The only processes by which I imagine the entire contents of one shard of a mapreduce ending up in memory of a single host are 1) a miracle, 2) the previous stage of the mapreduce wrote it there, in which case it is effectively remote access before the fact, and 3) the data lives there all the time for some other reason, in which case you don't have a mapreduce, you have a coprocessor.
I agree about the one megabyte at a time. I disagree that that implies it's likely evicted from RAM.
For one, if I'm running a MapReduce, it's either likely on the same data someone else is investigating... Or I'm likely to have to re-run my MapReduce a few times, calculating exactly what I'm looking for.
So, I may be cheating by saying "Only the first time is the data cold. Every other time, the data is likely to be hot in RAM."
Because you'd be right to say, "Well, duh. I'm talking about the FIRST time. And since I'm talking about the first time, moving the data across the network wouldn't be that bad. After that, yes, of course it will still likely be hot, just as you describe."
My only real response is, I think it depends on how much data, versus how much code you're talking about. And can you maybe even broadcast the code to the right nodes, to even more effectively use the network.
"in which case you don't have a mapreduce, you have a coprocessor."
...a coprocessor which uses the MapReduce API. Yes, I think that is exactly what you have. And it works whether the data is local or remote, cold or hot. One unified API for all of those things, and it's probably optimized to work local and hot, because I'm betting more than half of all MapReduces actually work out to be local and hot.
For one, if I'm running a MapReduce, it's either likely on the same data someone else is investigating... Or I'm likely to have to re-run my MapReduce a few times, calculating exactly what I'm looking for.
So, I may be cheating by saying "Only the first time is the data cold. Every other time, the data is likely to be hot in RAM."
Because you'd be right to say, "Well, duh. I'm talking about the FIRST time. And since I'm talking about the first time, moving the data across the network wouldn't be that bad. After that, yes, of course it will still likely be hot, just as you describe."
My only real response is, I think it depends on how much data, versus how much code you're talking about. And can you maybe even broadcast the code to the right nodes, to even more effectively use the network.
"in which case you don't have a mapreduce, you have a coprocessor."
...a coprocessor which uses the MapReduce API. Yes, I think that is exactly what you have. And it works whether the data is local or remote, cold or hot. One unified API for all of those things, and it's probably optimized to work local and hot, because I'm betting more than half of all MapReduces actually work out to be local and hot.
Even with high-performance networks in place, you absolutely must budget your bandwidth between management communications, low-latency service communications, and batch job communications, amongst possibly even others. Having data locality is still a really, really big priority.
(Furthermore, building out separate networks is extremely cost-prohibitive once you've expanded out to a total footprint in the low-1e6 ballpark.)
Think about it this way: In recent history, we've added L3 cache to commodity processors because keeping data on-chip is still faster, in 2015, than going off-chip to RAM. The same analog holds when trying to fetch from local disk versus remote disk.
Brendan Gregg actually provides a really good time-stretched view of various operations in his book on performance tuning. Go give that book a read; there's some _really_ good stuff there.
(Furthermore, building out separate networks is extremely cost-prohibitive once you've expanded out to a total footprint in the low-1e6 ballpark.)
Think about it this way: In recent history, we've added L3 cache to commodity processors because keeping data on-chip is still faster, in 2015, than going off-chip to RAM. The same analog holds when trying to fetch from local disk versus remote disk.
Brendan Gregg actually provides a really good time-stretched view of various operations in his book on performance tuning. Go give that book a read; there's some _really_ good stuff there.
The MR master handles it.
Mapreduce workers can request shards (work) that have data local to the host it is running on. For large mapreduces this basically works because both the data and the tasks are in most of the cluster, and for small mapreduces it doesn't matter.
Mapreduce workers can request shards (work) that have data local to the host it is running on. For large mapreduces this basically works because both the data and the tasks are in most of the cluster, and for small mapreduces it doesn't matter.
to get an idea look at how Hadoop or spark run on Mesos. You push the data locality into tasks, executors and their caches.
Does Kubernetes use chroot jails like Borg, or has it advanced to LXC containers? Nowadays LXC isolation seems strictly preferable.
Borg also uses cgroups, just like LXC.
From the paper: "We use a Linux chroot jail as the primary security isolation mechanism between multiple tasks on the same machine… all Borg tasks run inside a Linux cgroup-based resource container."
From the paper: "We use a Linux chroot jail as the primary security isolation mechanism between multiple tasks on the same machine… all Borg tasks run inside a Linux cgroup-based resource container."
Remember that Borg is an old thing. At the time they probably started out with chroot jails, because cgroups weren't ready yet. But Kubernetes, Mesos use docker and cgroups(Mesos).
Cgroups have been around for a long time. Wikipedia says 8 years.
I think was in RHEL 6.2 possibly even 6.0
I think was in RHEL 6.2 possibly even 6.0
cgroups were largely written by members of the Borg team.
[deleted]
Kubernetes uses Docker (or now rkt) to do the actual containing. The focus is on scheduling and managing lots of containers.
You may run O(10) containers on a single machine. When you run O(10k)+ containers in a cluster you need new tools to manage things.
You may run O(10) containers on a single machine. When you run O(10k)+ containers in a cluster you need new tools to manage things.
LXC uses chroot + other things. chroot is essentially the foundation of any containerization tool.
Not really- chroot provides no isolation, namespaces do.
choot is the visible portion (changes root.. ie changes /mnt/blah/bleh to /) but not "the foundation"
Things can in fact work isolated without chroot, too. Its just convenient.
Things can in fact work isolated without chroot, too. Its just convenient.
Note that chroot will provide containerization, but without isolation (processes can always fork out of their chroot jails). You still need things like pid namespaces etc.
[deleted]
Does anyone else feel like calling your massive infrastructure software "Borg" might be tempting fate? I mean, I'm not superstitious, but I would be weary of calling my satellite network "Skynet."
Skynet, like the UK military satellite comms network?
https://en.m.wikipedia.org/wiki/Skynet_(satellites)
https://en.m.wikipedia.org/wiki/Skynet_(satellites)
Exciting times!