"But you don’t have to install Hadoop at all, unless you decide to use HDFS as Spark’s distributed file system. If you don’t, you can choose one of the other DFS’s that it supports"
What other file systems does spark support? And how seamless is that integration? I know with spark, I can write a Spark SQL query that will go directly against data persisted in HDFS. Should I expect the same behavior if I swap out HDFS with some other DFS?
Spark doesn't depend on much of anything from the Hadoop ecosystem, including HDFS. It supports S3, as well as NFS or any other locally mountable filesystem right out of the box. The FAQ talks a bunch about Spark's weak ties to Hadoop.
http://spark.apache.org/faq.html
We typically do #2 at our company and it's been fine so far. The bigger issue isn't a single 1 TB data set. It's multiple large data sets as one then must handle the data shuffles during joins, etc. The ability to keep the RDD in memory through the operations still tends to beat normal long winded hadoop operations anyways...
Saying things like "Going beyond Hadoop" is very misleading. Virtually all of the Hadoop vendors out there, whether it be Cloudera, Hortonworks, MapR commercially support Spark as a computation framework for Hadoop, and some already have customers who are already using Spark on HDFS to power mission critical applications. People still pretend that Hadoop is some batch-orientated system with a distributed file system, but they couldn't be farther from the truth. Hadoop is a movement, an evolution of how data is to be analyzed in today's world.
The fact is, the vast majority of people who use spark (or will use spark), will rely on Hadoop for a lot of underlying technologies, such as, YARN (resource management) or HDFS (distributed filesystem / in-memory caching). Further, if you think Spark is somehow the "end all be all" computation framework, you're living in a fantasy world. The best part of Hadoop is that depending on your use case, you can bring a multitude of applications to your data, whether it's Spark, Tez, MapReduce, HBase, Impala, Drill, Presto, Tajo, Accumulo, ...the list goes on, and continues to evolve. Spark is in no way replacing Hadoop, it's only strengthening it.
You have a point, but it's also a fact that lots of people use "Hadoop" interchangeably with "MapReduce". And Spark can in fact replace the Hadoop infrastructure entirely, as it's not a component of that ecosystem. Just because the various Hadoop vendors also support Spark only validates the point that there's a need to "go beyond Hadoop".
Just because people use Hadoop and MapReduce interchangeably doesn't make it correct. I would love to hear how you think Spark can replace Hadoop, because that is an astonishingly inaccurate statement. Which part of Spark reliably distributes data? Which part of Spark handles enterprise level security? Which part of Spark can coordinate resources in multi-tenant environments? The answer is none of them, it relies on Hadoop for that.
um, you sound like a vendor. hadoop does mean map-reduce + hdfs as the common usage by the majority of devs + admins. Claiming hadoop is now some distribution of tools is fine, but that's simply not what the common usage is.
It remains to be seen if yarn will carry the day or no; my suspicion is that many people are essentially going to be running spark on hdfs. I don't see much use for yarn unless you need to balance hadoop and yarn, and weren't their claims that yarn was going to support eg mpi style computation that didn't pan out?
um, you sound like a vendor. hadoop does mean map-reduce + hdfs as the common usage by the majority of devs + admins. Claiming hadoop is now some distribution of tools is fine, but that's simply not what the common usage is.
I am a Hadoop developer, and I can tell you that Hadoop does not mean "map-reduce + hdfs". That's also not what people are installing when they install Cloudera's distribution of Hadoop, Hortonworks' distribution of Hadoop, or even Intel's distribution of Hadoop (which is being discontinued in favor of adopting Cloudera's). This is more old information from 2008, being replayed as current. YARN even lives in the Hadoop source code repository, it's hard to get more "Hadoop" than that. Spark has its own repo, but it uses many classes from Hadoop like InputFormat, etc.
It remains to be seen if yarn will carry the day or no; my suspicion is that many people are essentially going to be running spark on hdfs. I don't see much use for yarn unless you need to balance hadoop and yarn, and weren't their claims that yarn was going to support eg mpi style computation that didn't pan out?
Much confusion. Much sadness.
You run YARN (or its close competitor, Mesos) because you want to have multiple jobs going on in the same cluster at once. You need things like per-user queues, job control, reserving CPU and memory resources. The jobs going on at once may be multiple MapReduce jobs, or they may be multiple Spark jobs. Even Databricks, which employs many of the early Spark developers, doesn't ship a product that runs Spark in standalone mode. They run on Mesos.
You seem to be very knowledgeable about Hadoop. Could you tell us a bit about the technical reasons why its performance is so god-awfully underwhelming. Is it a wrong choice of algorithms, internal data structure, architecture, design. I have used Google's implementation of mapreduce and then current version of Hadoop. Hadoop's performance was a piece of crap in comparison, and yes the the bottleneck was shuffle. With Java, I would have expected a 20~30% hit not a slowdown by 4~6 times and using roughly that many times more memory. Weirdly enough, I have heard that Python streaming on Hadoop outperforms the Java Hadoop, but I have not tried this myself. At that time UIUC's sector would outperform Hadoop too, and that was an university project maintained by a grad student.
Finally has this changed with Hadoop 2.* And in case you have experienced Googles's mapreduce and Hadoop (which seems likely) I would like to know more.
Yahoo's hadoop has a variety of issues. I'd rather not say company names, but this comes from working at a company that had hadoop in production by 2007 and last I heard was running 10+pb datastores with 10k+ hadoop cores.
Yahoo's hadoop writes all output (from mappers and reducers) to local disks which are typically spinning rust. This is an anti-pattern: you run multiple mappers per box, typically 1.1-1.3 per core, and you have multiple jobs running simultaneously. You then perform a disk based merge sort, and then every reducer must connect to every mapper to get the data associated with its keys. The second you get a hot disk, whole jobs get choked because of the api invariants: all mappers complete before you can finish (or perhaps start) sort, all sorts complete before reducers, all reducers finish before the next mapper pass. Further, avoiding having every reducer have to have high throughput network to every mapper is key; the hadoop design makes it difficult to optimize over network topology like higher within-rack bandwidth.
There are far better designs implemented: never write to local disk from the mappers, instead writing to a global dfs then performing the sort in the dfs. I worked for a company that took hadoop 0.17 and made it roughly 10x more performant (and not on toy datasets: on petabyte scale datasets) with roughly 10 engineer-years with this design.
I think it's also fair to say that the quality of engineers on hadoop, particularly when they worked at yahoo, was... subpar. There was a ton of low hanging fruit.
Also, yahoo's workload wasn't, or so I've heard, highly cpu bound. They where using hadoop at least in part just for hdfs capacity. So if you aren't cpu constrained, a lot of optimizations don't matter. Better companies have to run at say 95%+ cpu or their cfo won't authorize more boxes.
You can also go after things like optimizing your compressors / decompressors, which are probably by far the hottest bits of code. Get icc, break out vtune, hire a really good optimizing dev to go after even small gains, etc.
Finally, the hdfs design is pretty shitty, on two fronts: the namenode gets slow (choking everyone), and the disk-based format is missing lots of things to enable optimizations like certain types of append, or indexing. Indexing is incredibly powerful for things like join, enabling a map-side join instead of having to read all your data once, sort by join keys, then reread in your reducers.
Another problem with hdfs design is capacity adding; hdfs has shitty rebalancing tools. The typical problem is this: hot data is new data. So if you add capacity to hdfs, your old capacity is say 80% full and your new capacity is 0% full so where does all new data go? To the new capacity. Which is quickly 100% saturated and choking your jobs. Properly rebalancing a dfs is a hard problem on it's own. Further, hdfs' use of triple replication (well, you can actually set the replication level) instead of better algorithms like erasure encoding does no favors to performance as well.
edit: by the way, never writing to local disk and instead going from mappers straight to dfs then performing a rough/fine grained sort in the dfs is a lot like google's design. You do trade network i/o for disk i/o, but that's often easier to scale and, as mentioned, easier to optimize over your network topology. Finally, you can do multi-pass sorts partially locally by running sorters on your disk nodes.
I already upvoted, but this needs more, thanks for the excellent read. Some of the inefficiencies you describe are pretty crazy ! If any of the optimizations you did are documented in blogs, would love to have a pointer.
@cbsmith Just in case you have links handy could you post them. Much appreciated.
These issues with Hadoop and HDFS in particular are pretty well documented even by people in the Hadoop project. There was also a bunch of blog posts posted at Yahoo shortly after Cutting's departure that looked at a bunch of other issues with the IO subsystem.
Much of this stuff has been heavily refactored, particularly with Hadoop 2.0, so the problems are at least much smaller than they used to be. Early Hadoop was not exactly a masterful work in terms of the code & algos. What was great about it was the community that was built around it.
> I'd rather not say company names, but this comes from working at a company that had hadoop in production by 2007 and last I heard was running 10+pb datastores with 10k+ hadoop cores.
> Another problem with hdfs design is capacity adding; hdfs has shitty rebalancing tools. The typical problem is this: hot data is new data. So if you add capacity to hdfs, your old capacity is say 80% full and your new capacity is 0% full so where does all new data go? To the new capacity. Which is quickly 100% saturated and choking your jobs. Properly rebalancing a dfs is a hard problem on it's own. Further, hdfs' use of triple replication (well, you can actually set the replication level) instead of better algorithms like erasure encoding does no favors to performance as well.
The tools for HDFS balancing are quite primitive, which is why a lot of people like using MapR or Cassandra, but the rebalancing tools built in to HDFS after a 0.18.x along with a few good practices would get you there... just not the most efficient way possible.
Messages like this make me sad. Hadoop hasn't been "Yahoo's Hadoop" since at least 2008. MapReduce is deprecated in favor of newer frameworks like Cloudera Impala and Apache Spark. These tools are years old. If you are still discussing 2008 software as if it were current, you're giving people a very false picture of what is out there.
Further, avoiding having every reducer have to have high throughput network to every mapper is key; the hadoop design makes it difficult to optimize over network topology like higher within-rack bandwidth.
Hadoop was designed around rack topologies where bandwidth within racks was higher than bandwidth between them. That is why it tries to place mappers near where the data is that the mappers will be reading. The general assumption is that mappers will be generating a much smaller data set than what they're reading. If that's not true, then yes, you may have locality issues. You can solve this problem by writing your own InputFormat.
There are far better designs implemented: never write to local disk from the mappers, instead writing to a global dfs then performing the sort in the dfs. I worked for a company that took hadoop 0.17 and made it roughly 10x more performant (and not on toy datasets: on petabyte scale datasets) with roughly 10 engineer-years with this design.
There have been a bunch of projects to optimize MapReduce, like Sailfish. But ultimately higher-level optimizations like avoiding spilling between different phases of a multi-stage job (like Spark implements) are more important. And Impala implements a ton of join optimizations, of course, as well as doing absolutely everything it can in memory.
Finally, the hdfs design is pretty shitty, on two fronts: the namenode gets slow (choking everyone), and the disk-based format is missing lots of things to enable optimizations like certain types of append, or indexing. Indexing is incredibly powerful for things like join, enabling a map-side join instead of having to read all your data once, sort by join keys, then reread in your reducers.
In practice, the HDFS NameNode only gets slow if you create lots of small files.
There are plenty of external indices you can put on top of HDFS. For example, you can run HBase. Or if you don't want to run another daemon, you can use Apache Parquet or Apache ORCFile... file formats which are inherently indexed.
Another problem with hdfs design is capacity adding; hdfs has shitty rebalancing tools. The typical problem is this: hot data is new data. So if you add capacity to hdfs, your old capacity is say 80% full and your new capacity is 0% full so where does all new data go? To the new capacity. Which is quickly 100% saturated and choking your jobs.
This is completely untrue. By default, HDFS doesn't make any attempt to load balance between disks or nodes at the moment; it will keep writing until the disk / node fills up. The sysadmin must run the balancer to move things around between full and less-full nodes. I don't think any version of HDFS ever implemented the behavior you talk about as the default.
hdfs' use of triple replication (well, you can actually set the replication level) instead of better algorithms like erasure encoding does no favors to performance as well.
Erasure encoding isn't better for all use cases, only for cold data. Sometime you want those replicas. At some point, I expect HDFS will get RAID support, as well.
To be honest, it's rare to see a cluster where lack of disk space is a problem. With 20 4 TB disk drives on each node, optimizing for reduced disk space is a bit foolish unless you profile and find that I/O is indeed the bottleneck.
It's really depressing to read anything about Hadoop on Hacker News. It seems like people talk about 2008 software and Hadoop 0.20 as though it were current. Nobody talks about Windows XP as though that were the latest version of Windows (which it was in 2006), but somehow talk...
In practice -- again, see my experience writing jobs processing pbs of data on 1000s of nodes -- hadoop experiences namenode issues. Hadoop was designed many ways, blah blah blah, but that's essentially not what the software is as shipped. And that's about it; I was going to respond point by point, but your post displays a general lack of knowledge of large-scale hadoop deployments, so it's basically a waste of time.
I work for a big Hadop vendor and I regularly diagnose performance issues with big clusters. Yes, sometimes even thousands of nodes. Most of the time, it comes down to an incorrect assumption or configuration. There are certainly real bugs and shortcomings, but the bottom line is that the systems we sell are orders of magnitude cheaper to build and operate than what they replace.
And that's about it; I was going to respond point by point, but your post displays a general lack of knowledge of large-scale hadoop deployments, so it's basically a waste of time.
You can't even respond to a single point? I'm sorry to hear that. I have been polite and respectful to you, can you pay me the same courtesy?
You are playing from the old playbook though. Heck, we have HA NameNode setups and name node clusters now, and some implementations of Hadoop don't even use HDFS. The Hadoop's tech stack has changed a lot over the years. It's much different now. It's worth noting that there are many examples now of clusters with five figures worth of nodes.
27 comments
[ 3.2 ms ] story [ 43.0 ms ] threadWhat other file systems does spark support? And how seamless is that integration? I know with spark, I can write a Spark SQL query that will go directly against data persisted in HDFS. Should I expect the same behavior if I swap out HDFS with some other DFS?
2) Scale out. Spark can easily handle hundreds of nodes in a single cluster. Aggregate RAM across all of them can be used.
3) Cache intermediate data sets and/or hot data sets, as opposed to the entire data set.
We typically do #2 at our company and it's been fine so far. The bigger issue isn't a single 1 TB data set. It's multiple large data sets as one then must handle the data shuffles during joins, etc. The ability to keep the RDD in memory through the operations still tends to beat normal long winded hadoop operations anyways...
The fact is, the vast majority of people who use spark (or will use spark), will rely on Hadoop for a lot of underlying technologies, such as, YARN (resource management) or HDFS (distributed filesystem / in-memory caching). Further, if you think Spark is somehow the "end all be all" computation framework, you're living in a fantasy world. The best part of Hadoop is that depending on your use case, you can bring a multitude of applications to your data, whether it's Spark, Tez, MapReduce, HBase, Impala, Drill, Presto, Tajo, Accumulo, ...the list goes on, and continues to evolve. Spark is in no way replacing Hadoop, it's only strengthening it.
It remains to be seen if yarn will carry the day or no; my suspicion is that many people are essentially going to be running spark on hdfs. I don't see much use for yarn unless you need to balance hadoop and yarn, and weren't their claims that yarn was going to support eg mpi style computation that didn't pan out?
I am a Hadoop developer, and I can tell you that Hadoop does not mean "map-reduce + hdfs". That's also not what people are installing when they install Cloudera's distribution of Hadoop, Hortonworks' distribution of Hadoop, or even Intel's distribution of Hadoop (which is being discontinued in favor of adopting Cloudera's). This is more old information from 2008, being replayed as current. YARN even lives in the Hadoop source code repository, it's hard to get more "Hadoop" than that. Spark has its own repo, but it uses many classes from Hadoop like InputFormat, etc.
It remains to be seen if yarn will carry the day or no; my suspicion is that many people are essentially going to be running spark on hdfs. I don't see much use for yarn unless you need to balance hadoop and yarn, and weren't their claims that yarn was going to support eg mpi style computation that didn't pan out?
Much confusion. Much sadness.
You run YARN (or its close competitor, Mesos) because you want to have multiple jobs going on in the same cluster at once. You need things like per-user queues, job control, reserving CPU and memory resources. The jobs going on at once may be multiple MapReduce jobs, or they may be multiple Spark jobs. Even Databricks, which employs many of the early Spark developers, doesn't ship a product that runs Spark in standalone mode. They run on Mesos.
Finally has this changed with Hadoop 2.* And in case you have experienced Googles's mapreduce and Hadoop (which seems likely) I would like to know more.
Yahoo's hadoop writes all output (from mappers and reducers) to local disks which are typically spinning rust. This is an anti-pattern: you run multiple mappers per box, typically 1.1-1.3 per core, and you have multiple jobs running simultaneously. You then perform a disk based merge sort, and then every reducer must connect to every mapper to get the data associated with its keys. The second you get a hot disk, whole jobs get choked because of the api invariants: all mappers complete before you can finish (or perhaps start) sort, all sorts complete before reducers, all reducers finish before the next mapper pass. Further, avoiding having every reducer have to have high throughput network to every mapper is key; the hadoop design makes it difficult to optimize over network topology like higher within-rack bandwidth.
There are far better designs implemented: never write to local disk from the mappers, instead writing to a global dfs then performing the sort in the dfs. I worked for a company that took hadoop 0.17 and made it roughly 10x more performant (and not on toy datasets: on petabyte scale datasets) with roughly 10 engineer-years with this design.
I think it's also fair to say that the quality of engineers on hadoop, particularly when they worked at yahoo, was... subpar. There was a ton of low hanging fruit.
Also, yahoo's workload wasn't, or so I've heard, highly cpu bound. They where using hadoop at least in part just for hdfs capacity. So if you aren't cpu constrained, a lot of optimizations don't matter. Better companies have to run at say 95%+ cpu or their cfo won't authorize more boxes.
You can also go after things like optimizing your compressors / decompressors, which are probably by far the hottest bits of code. Get icc, break out vtune, hire a really good optimizing dev to go after even small gains, etc.
Finally, the hdfs design is pretty shitty, on two fronts: the namenode gets slow (choking everyone), and the disk-based format is missing lots of things to enable optimizations like certain types of append, or indexing. Indexing is incredibly powerful for things like join, enabling a map-side join instead of having to read all your data once, sort by join keys, then reread in your reducers.
Another problem with hdfs design is capacity adding; hdfs has shitty rebalancing tools. The typical problem is this: hot data is new data. So if you add capacity to hdfs, your old capacity is say 80% full and your new capacity is 0% full so where does all new data go? To the new capacity. Which is quickly 100% saturated and choking your jobs. Properly rebalancing a dfs is a hard problem on it's own. Further, hdfs' use of triple replication (well, you can actually set the replication level) instead of better algorithms like erasure encoding does no favors to performance as well.
edit: by the way, never writing to local disk and instead going from mappers straight to dfs then performing a rough/fine grained sort in the dfs is a lot like google's design. You do trade network i/o for disk i/o, but that's often easier to scale and, as mentioned, easier to optimize over your network topology. Finally, you can do multi-pass sorts partially locally by running sorters on your disk nodes.
@cbsmith Just in case you have links handy could you post them. Much appreciated.
Much of this stuff has been heavily refactored, particularly with Hadoop 2.0, so the problems are at least much smaller than they used to be. Early Hadoop was not exactly a masterful work in terms of the code & algos. What was great about it was the community that was built around it.
[1] https://www.quantcast.com/inside-quantcast/2012/10/letter-fr...
[2] https://www.quantcast.com/inside-quantcast/2012/10/quantsort...
[3] https://www.quantcast.com/inside-quantcast/2013/12/petabyte-...
I'm going to say... Netseer. ;-)
The tools for HDFS balancing are quite primitive, which is why a lot of people like using MapR or Cassandra, but the rebalancing tools built in to HDFS after a 0.18.x along with a few good practices would get you there... just not the most efficient way possible.
Further, avoiding having every reducer have to have high throughput network to every mapper is key; the hadoop design makes it difficult to optimize over network topology like higher within-rack bandwidth.
Hadoop was designed around rack topologies where bandwidth within racks was higher than bandwidth between them. That is why it tries to place mappers near where the data is that the mappers will be reading. The general assumption is that mappers will be generating a much smaller data set than what they're reading. If that's not true, then yes, you may have locality issues. You can solve this problem by writing your own InputFormat.
There are far better designs implemented: never write to local disk from the mappers, instead writing to a global dfs then performing the sort in the dfs. I worked for a company that took hadoop 0.17 and made it roughly 10x more performant (and not on toy datasets: on petabyte scale datasets) with roughly 10 engineer-years with this design.
There have been a bunch of projects to optimize MapReduce, like Sailfish. But ultimately higher-level optimizations like avoiding spilling between different phases of a multi-stage job (like Spark implements) are more important. And Impala implements a ton of join optimizations, of course, as well as doing absolutely everything it can in memory.
Finally, the hdfs design is pretty shitty, on two fronts: the namenode gets slow (choking everyone), and the disk-based format is missing lots of things to enable optimizations like certain types of append, or indexing. Indexing is incredibly powerful for things like join, enabling a map-side join instead of having to read all your data once, sort by join keys, then reread in your reducers.
In practice, the HDFS NameNode only gets slow if you create lots of small files.
There are plenty of external indices you can put on top of HDFS. For example, you can run HBase. Or if you don't want to run another daemon, you can use Apache Parquet or Apache ORCFile... file formats which are inherently indexed.
Another problem with hdfs design is capacity adding; hdfs has shitty rebalancing tools. The typical problem is this: hot data is new data. So if you add capacity to hdfs, your old capacity is say 80% full and your new capacity is 0% full so where does all new data go? To the new capacity. Which is quickly 100% saturated and choking your jobs.
This is completely untrue. By default, HDFS doesn't make any attempt to load balance between disks or nodes at the moment; it will keep writing until the disk / node fills up. The sysadmin must run the balancer to move things around between full and less-full nodes. I don't think any version of HDFS ever implemented the behavior you talk about as the default.
hdfs' use of triple replication (well, you can actually set the replication level) instead of better algorithms like erasure encoding does no favors to performance as well.
Erasure encoding isn't better for all use cases, only for cold data. Sometime you want those replicas. At some point, I expect HDFS will get RAID support, as well.
To be honest, it's rare to see a cluster where lack of disk space is a problem. With 20 4 TB disk drives on each node, optimizing for reduced disk space is a bit foolish unless you profile and find that I/O is indeed the bottleneck.
It's really depressing to read anything about Hadoop on Hacker News. It seems like people talk about 2008 software and Hadoop 0.20 as though it were current. Nobody talks about Windows XP as though that were the latest version of Windows (which it was in 2006), but somehow talk...
And that's about it; I was going to respond point by point, but your post displays a general lack of knowledge of large-scale hadoop deployments, so it's basically a waste of time.
You can't even respond to a single point? I'm sorry to hear that. I have been polite and respectful to you, can you pay me the same courtesy?