Let's Build a Modern Hadoop
medium.com
medium.com
Being more explicitly Linux-centric and dropping the JVM paves the way for some major performance optimizations. In practice, virtually everyone uses Linux for these kinds of applications so portability is not a significant concern. Efficiency in the data center is becoming a major concern, both in terms of CapEx and OpEx, and the basic Hadoop stack can be exceedingly suboptimal in this regard. Free software is not cost effective if you require 10-100x the hardware (realistic) of a more optimal implementation.
However, I would use a different architecture generally rather than reimplementing the Hadoop one. The Achille's Heel of Hadoop (and Spark) is its modularity. Throughput for database-like workloads, which matches a lot of modern big data applications, is completely dependent on tightly scheduling execution, network, and storage operations without any hand-offs to other subsystems. If the architecture requires network and storage I/O to be scheduled by other processes, performance falls off a cliff for well-understood reasons (see also: mmap()-ed databases). People know how to design I/O intensive data processing engines, it was just outside the original architectural case for Hadoop. We can do better.
A properly designed server kernel running on cheap, modern hardware and a 10 GbE network can do the following concurrently on the same data model:
- true stream/event processing on the ingest path at wire speed (none of that "fast batch" nonsense)
- drive that ingest stream all the way through indexing and disk storage with the data fully online
- execute hundreds of concurrent ad hoc queries on that live data model
These are not intrinsically separate systems, it is just that we've traditionally designed data platforms to be single purpose. However, it does require a single scheduler across the complete set of operations required in order to deliver that workload.
From a software engineering standpoint it is inconvenient that good database kernels are essentially monolithic and incredibly dense but it is unavoidable if performance matters.
Long term we don't want to mandate either of these technologies. We'd like to offer users a variety of job formats (docker and rocket come to mind more domain specific things like SQL are interesting as well) and a variety of storage options (zfs, overlayfs, aufs and in-memory to name a few). However we had to be pragmatic early on and pick what we thought were the highest leverage implementations so that we could ship something that worked.
We'll certainly be looking in to getting rid of this mandate in the near future, we like giving people a variety of choices.
> Contrast this with Pachyderm MapReduce (PMR) in which a job is an HTTP server inside a Docker container (a microservice). You give Pachyderm a Docker image and it will automatically distribute it throughout the cluster next to your data. Data is POSTed to the container over HTTP and the results are stored back in the file system.
Now I'm not even sure anymore whether they are trying to troll people or if they are that out of touch with reality.
- No focus on trying to take on the real issues, but working around a perceived problem which would have been a non-issue after 5 minutes of research? Check.
- Trying to improve things by looking at bad ideas which weren't even state-of-the-art ten years ago? Check.
- Being completely clueless about the real issues in distributed data processing? Check.
- Whole thing reads like some giant buzzword-heavy bullshit bingo? Check.
- Some Silicon-Valley-kids-turned-JavaScript-HTML-"programmers"-turned-NodeJS-developers-turned-Golang-experts thinking that they can build a Git-like distributed filesystem from scratch? Check.
- "A git-like distributed file system for a Dockerized world." ... Using Go? ... Please just kill me.
If I ever leave the software industry, it's because of shit like this.
Zookeeper isn't Modular? No mention of HTCondor+Dagman? No mention of Spark?
I've written a general DAG processor/workflow engine/metascheduler/whatever you want to call it. It's used by various physics and astronomy experiments. I've interfaced it with the Grid/DIRAC, Amazon, LSF, Torque, PBS, SGE, and Crays. There's nothing in it that precludes Docker jobs from running. I've implemented a mixed-mode (Streaming+batch processing + more DAG) version of it which just uses job barriers to do the setup and ZeroMQ for inter-job communication. I think something like Spark's resilient distributed dataset would be nice here as well.
We don't use hadoop because, as was said, because it's narrow in scope and it's not a good general purpose DAG/Workflow Engine/Pipeline.
I think this is a small improvement, but I don't really see it being much better than hadoop, or HTCondor, or Spark.
HTCondor is pretty amazing. I think somebody should be building a modern HTCondor, not a modern Hadoop.
A bug-free, cross platform, rock-solid HTCondor would be great. Sadly the existing HTCondor is a buggy turd. I've fixed some stuff in it and a lot of the code is eye-roll worthy. I can practically guarantee you hours of wondering why something that by any reasonable expectation should be working isn't working, and not giving any meaningful error message.
But like you said, a really solid HTCondor rewrite with the Stork stuff baked in would pretty much be all anyone should need, I think. It's not like you'd even really need to dramatically re-architect the thing. The problems with HTCondor really are mostly just code quality and the user experience.
How big are the data sets? How does it compare with the MapReduce paradigm?
I see there is this Dagman project which sounds similar to some of the newer Big Data frameworks that use a DAG model. I will make an uneducated guess that it deals with lots of computation on smaller sized data (data that fits on a machine). Maybe you have 1 GB or 10 GB of data that you want to run through many (dozens?) of transformations. So you need many computers to do the computation, but not necessarily to store the data.
I would guess another major difference is that there is no shuffle (distributed sort)? That is really what distinguishes MapReduce from simple parallel batch processing -- i.e. the thing that connects the Map and Reduce steps (and one of the harder parts to engineer).
In MapReduce you often have data much bigger than a single machine, e.g. ~100 TB is common, but you are doing relatively little computation on it. Also MapReduce tolerates many node failures and thus scales well (5K+ machines for a single job), and tries to reign in stragglers to reduce completion time.
I am coming from the MapReduce paradigm... I know people were doing "scientific computing" when I was in college but I never got involved in it :-( I'm curious how the workloads differ.
Our feeling is that there's a big difference between having projects like this in the ecosystem and having a canonical implementation that comes preloaded in the software. "Batteries included but removable" is the best phrasing of this we've found (stolen from Docker). I've seen tons of pipeline management tools in the Hadoop ecosystem (even contributed to some) it's not that they're bad it's just that they're not standardized. This winds up being a huge cost because if 2 companies use different pipeline management it gets a lot harder to reuse code.
I think etcd is a pretty good example of this. Etcd is basically a database (that's the way CoreOS describes it) by database standards it's pretty crappy. But it's simple and it's on every CoreOS installation so you can use it without making your software a ton harder to deploy.
Hope this clears some stuff up!
Switching to Docker feels like a real step backwards. Rather than a jar containing strictly-specified bytecode, cluster jobs are going to be random linux binaries that could do any combination of system calls. You need another layer of build/management system to make your docker containers useable. Worst of all, rather than defining common shared services for e.g. job scheduling, this is going to encourage users to put any random component in their docker image. We'll end up with a more fragmented ecosystem, not less.
Personally, I dig Spark on EMR or Spark on Mesos (with Scala as a domain language), and I'm not sure how this plays with the rest of the world.
I just need to take a big breath every time I say it though.
Our feeling isn't that Java is bad it's that it shouldn't be the only option for big data processing. There are a lot of people that want to do this type of computing and a lot of them hate Java. We want to build a solution that can serve them better and Docker is actually an incredibly high leverage way to do it.
I feel like I don't fully understand your criticism of choosing Docker as platform.
> cluster jobs are going to be random linux binaries that could do any combination of system calls
Could you give an example of what could go wrong here? Docker prevents you from talking to the outside world so you can't call out to an outside service in the middle of a job or anything like that.
> Worst of all, rather than defining common shared services for e.g. job scheduling
In Pachyderm job scheduling happens outside the user submitted containers. The job scheduler is what actually spins up the containers and pushes data through them. I'm afraid I don't understand what it would even mean for one of those containers to then do job scheduling itself.
Hope this clears up our thoughts on this a bit.
I can see the argument for managing workload definitions as an {app repo, VM descriptor} tuple—kind of like Heroku app definitions—rather than a single app+VM image blob. But when you're actually deploying them to run on machines, nothing beats baking everything into a reproducable static-binary-equivalent slug.
You don't need to let people run arbitrary docker images as workloads, but the ability to let people pick any VM—rather than just your preferred VM—while having this be just as safe and sandboxed, opens up a lot of flexibility.
(This is the same argument I give for Chromium's PNaCl, by the way: the advantage doesn't come from your ability to run one-off native apps in the browser; the true advantage is in <script>s that can specify exactly the interpreter engine—a PNaCl-binary VM—they need. Nobody seems to see eye-to-eye on this one, either.)
Thank you.
JVM != Java. It runs a wide variety of languages, and this is not just theory - you can do Spark in Python right now. "Every language I care about" was perhaps overly flippant, but between Groovy, Clojure and Scala there's something to suit most programming tastes.
Meanwhile I hate Linux, and Docker doesn't have a good answer for that; the suggestions I've seen for developing with Docker on other platforms boil down to "run Linux in a VM".
> Could you give an example of what could go wrong here? Docker prevents you from talking to the outside world so you can't call out to an outside service in the middle of a job or anything like that.
I'm more worried about forward compatibility, standardization - and partly attack surface I guess. I can run jars from 15 years ago, on today's JVM - and not because it bundles an old version; I can link today's programs against 15-year-old libraries and they'll just work. If a library I built against has disappeared from the internet, that's fine.
Meanwhile I can't run linux binaries from 15 years ago (I did try with some of the loki games - maybe it's possible but it's decidedly nontrivial). Even if I found the right .sos, if I wanted to use a library with the ABI of ten years ago in my program, I'd have to rebuild the whole world. I guess I'm meant to rebuild my docker image, but that doesn't work if the library source is no longer available, and I'm not at all confident that the tooling does fully reproducible builds to the same extent as e.g. maven (from what I can see plenty of dockerfiles use unpinned versions).
Finally, the spec for the JVM is in one place, and it's published; apache harmony shows that a ground-up reimplementation is possible (that the TCK isn't open is very unfortunate, but the patents will expire sooner or later). I have a reasonable level of confidence that we'll always be able to run the jars we're producing today, without having to emulate x86s or what have you. Linux doesn't have a spec like that (yes the C code is available), and I don't think a ground-up reimplementation would be feasible.
> In Pachyderm job scheduling happens outside the user submitted containers. The job scheduler is what actually spins up the containers and pushes data through them. I'm afraid I don't understand what it would even mean for one of those containers to then do job scheduling itself.
Job scheduling was a poor example; I guess I'm thinking more things like e.g. running your own database on the cluster nodes, instead of talking to one outside the cluster via a proper interface. (Admittedly that's possible on hadoop with derby or h2).
Thinking about it more I guess it's less the service part and more the interface part that I'm worried about. A docker image could implement critical functionality with shell scripts. I bet they will. I've already spent too much of my life trying to get people to move away from shell scripts into something safer and more maintainable.
> the spec for the JVM is in one place, and it's published
Between Groovy, Clojure and Scala, there's something to suit most tastes on the scale of how much language specification there is. Scala has a published spec (along with many white papers on the R.I. internals), Clojure has its published spec in the comments in the source code, and Groovy actively changes the spec between versions to prevent addons and alternative implementations being built.
There is a good reason most of Map-Reduce frameworks are written on top of JVM - it is VERY easy to serialize Java/Scala code ship it to remote host and execute there.
This is not trivial in compiled languages ( I believe some hacks for Haskell exists )
What we need in Big Data space is more people that understand how computers really work. Some effort must be put into supporting TCP/IP Stacks in Userspace using new 10GBe cards or support for IB verbs. Also the long forgotten art of writing zero-copy code.
Obviously my understanding of networks is not at that level yet. What would a layman do to become somewhat vested in this? Any books or resrouces?
It operates very differently from Hadoop for us. SparkSQL allows third parties apps e.g. analytics to use JDBC/ODBC rather than going HDFS. And the in memory model and ease of caching data from HDFS allows for different use cases. We do most work now via SQL.
Combining Spark with Storm, ElasticSearch etc also permits a true real time ingestion and searching architecture.
This is something we probably should have spoken directly to in the article because of how popular spark is. The main reason we didn't is that we like Spark and don't feel it needs to be replaced. It doesn't seem to have most of the ecosystem problems we discuss because it's got a company behind it. My understanding of Spark is that it's designed to be used with different storage backends and I'm very curious to see what would happen if we got it talking to pfs. I think it could work very well because Spark's notion of immutable RDDs seems very similar to the way pfs handles snapshots.
Hope this clears a few things up and apologizes if I've mischaracterized Spark here I've only used it a little bit.
The bit about "We'll get around to Hive and Pig eventually" is also odd, as they're approaching legacy status too. (Albeit not there yet; please recall that I'm somewhat of an Impala skeptic, and of course there's the Tez booster to Hive.)
Also bizarre are some of the history claims in the article, e.g. that YARN is part of more or less original Hadoop.
My experience is that Hive is pretty widely used, I have a lot fewer data points about Pig. We're not particularly attached to either of these. That sentence was more to indicate that we'd be adding some sort of interfaces for specific data types.
Seems like we got our history wrong here. When was YARN added we'll add a correction.
Hope this helps clear things up.
Note that YARN also supports Docker containers as of Hadoop 2.6...
YARN-1964: Create Docker analog of the LinuxContainerExecutor in YARN https://issues.apache.org/jira/browse/YARN-1964
I'll stick with http://www.dbms2.com/2014/04/30/spark-on-fire/ however, for the most part. (The mid-2014 PR blitz I thought I was engineering didn't actually happen.)
I have terabytes of new data arriving per day that needs to be verified, ingested and translated. There simply is no way to do this via online or stream processing. If you ingesting clickstream or Twitter data then sure it will work. But more often than not you need to work with sets of data. And for that batch processing is the only option.
Are you folks doing sensor analysis or something, or eating logfiles, or what?
What we can get for that data is a major competitive advantage. We can offer much cheaper financial products since we model risk individually rather than as a cohort. It also allows us to have a single customer view despite reselling other companies products.
Building a single customer view with lots of disparate data sets is a big trend right now.
This is generally one of the things that's frustrated us about the Hadoop ecosystem. Hadoop's implementation of MapReduce, which only supports batch processing, has become conflated with MapReduce as a paradigm, which can be used for way more than just batch processing.
Hope this clears a few things up!
Our storage layer is built on top of btrfs, we haven't put together comprehensive benchmarks yet but our experience with btrfs is that there isn't a meaningful penalty to reading a batch of data from a commit based system compared to more traditional storage layers. I really wish I had concrete performance numbers to give you but we haven't had the time to put them together yet. I will say that our observations match with btrfs' reported performance numbers and are what I'd expect given the algorithmic complexity of their system.
Also, why yet another DFS that makes excuses for shitty local filesystems instead of leveraging ZFS? Use ZFS as the base for integrity, snapshots, send/receive and make a small service on top for managing cluster metadata.
We think Manta is pretty cool but it wasn't actually an inspiration for us.
I'm not sure I fully understand your criticism of the filesystem. Pfs was actually initially conceived as almost exactly what you're describing, a small service on top of ZFS. But we realized early on that ZFS was going to be a pain because of its patent status so we used btrfs instead. I think btrfs is worse than ZFS but I'm not sure I'd call it a shitty local filesystem.
But I am not in the favour of creating another stack as Hadoop replacement. Spark fills the role of distributed computation framework very well. It uses scala a functional language which provides neat abstractions for map,reduce,filter and other operations. It introduces RDD to abstract data. All the components graphx, sparksql, blinkdb use scala.
Also building a new map-reduce engine is a difficult task. The computation engine will lie in the middle of stack not on top as shown. Libraries will sit on top of that, so if you build an engine you will have to build everything on top of that.
See the Spark ecosystem : https://amplab.cs.berkeley.edu/software/
1) Do one thing well - Hadoop is so stacked up because of this (which has pros and cons). Every single component of Hadoop is designed specific to a job (there are many overlapped components, though)
2) Hadoop's main idea is to be data-local, yet this Pachy thing not to be: "Data is POSTed to the container over HTTP and the results are stored back in the file system". Data-locality (almost) always win.
3) There is no scheduler like YARN specified, yet fleet and etcd are not schedulers.
4) You may not like Java or JVM-based things (like I do), but there is no problem of being Java-based or JVM-based.
…
We unfortunately don't right now. It's surprisingly hard to get benchmarks that give you a true apples to apples comparison between us and Hadoop. However it seems from the comments in this thread that there's a lot of interest in those though so we'll try to put those together ASAP.
That said, I have no opinion on this article.
The key distinction is that this internal metrics data that is just used for monitoring is incidental, and can be thrown away without impacting the computation's result. It's just convenient information that humans might want to have so they can reason about the internal state of the overall system.
Anyway, I think such a tool would be highly useful. Also, "progress" is an often overlooked feature, but users actually appreciate it a lot, and often even require it. And you simply can't easily add progress-computation to a system as an after-thought. You have to "weave" it through the design as you build it.
There's also the fact that in such distributed systems, the entity that consumes the result of the computation is generally not the same entity that consumes the metrics.
I love the ability to run the cluster in any OS that has a JVM available, from big commercial UNIXes, Windows, *BSD and GNU/Linux distributions.
Hadoop is a poor imitation of a mainframe. When you are processing large lumps of data you need the following:
A task scheduler A message passing system(although in some workloads this can be proxied by the scheduler or filesystem) File storage
In mainframe land, you have a large machine with many processors all linked to the same kernel. As its one large machine, there is a scheduler that's so good you don't even know its there. However mainframes require specialist skills.
If you want shared memory cluster, you're going to need inifiband (which is actually quite cheap now, and has 40gig ethernet type majigger in it aswell)
but people are scared of that, so that leaves us with your standard commodity cluster.
In VFX we have this nailed. before leaving the industry I was looking after a ~25k CPU cluster backed to 15pb of storage. No fancy map reduce bollocks, just plain files and scripts. Everything was connected by ten gigs ethernet. Sustained write rate of 25 gigabytes a second.
firstly task placement is dead simple: https://github.com/mikrosimage/OpenRenderManagement/wiki/Int... look how simple that task mapping is. It is legitimate to point out that there are no primitives for IPC, but then your IPC needs are varied. Most log churn apps don't really need IPC, they just need a catcher to aggregate the results. Map:reduce has fooled people into thinking that everything is must be that way.
This is designed to run on bared metal. Anything else is a waste of CPU/IO. We used cgroups to limit process memory, so that we could run concurrent tasks on the same box. But everything ran the same image, anything else is just a plain arse ache.
The last item is file storage. HDFS is just terrible. the reason that its able to scale is that it provides literally nothing. Its just a key:value store, based on a bunch of key:value stores.
If you want fast super scalable posix filesystem, then use GPFS(or elastic storage). The new release has a kind of software raid that allows you to recover from a dead disk in less than an hour. (think ZFS with shards)
failing that you can use lustre (although that requires decent hardware raids, as its a glorified network raid0)
or if you are lucky you can split your file system into name spaces that reside on different NFS servers.
Either way, Hadoop, and hadoop inspired infrastructures are normally not the right answer. Measos is also probably not the answer unless you want a shared memory cluster. but then if you want this you need inifinband...
I'd love to see a Haskell-centric (performance and robust code) but ultimately language-agnostic alternative to the JVM-heavy stack that's currently in vogue. (Spark is a promising and powerful technology.) I choose Haskell because pretty much anything that the JVM does well (e.g. concurrency) Haskell also does well and, while I like Clojure and can enjoy Scala when written by competent people, Java is ugly and its community has worse aesthetic sense than a taint mole.