subatomic

I've just committed HADOOP-13786 Add S3A committer for zero-rename commits to S3 endpoints.. Contributed by Steve Loughran and Ryan Blue.

This is a serious and complex piece of work; I need to thank:

1. Thomas Demoor and Ewan Higgs from WDC for their advice and testing. They understand the intricacies of the S3 protocol to the millimetre.
2. Ryan Blue for his Staging-based S3 committer. The core algorithms and code will be in hadoop-aws come Hadoop 3.1.
3. Colleagues for their support, including the illustrious Sanjay Radia, and Ram Venkatesh for letting me put so much time into this.
4. Reviewers, especially Ryan Blue, Ewan Higgs, Mingliang Liu and extra especially Aaron Fabbri @ cloudera. It's a big piece of code to learn. First time a patch of mine has ever crossed the 1MB source barrier

I now understand a lot about commit protocols in Hadoop and Spark, including the history of interesting failures encountered, events which are reflected in the change logs of the relevant classes. Things you never knew about the Hadoop MapReduce commit protocol

1. The two different algorithms, v1 and v2 have very different semantics about the atomicity of task and job commits, including when output becomes visible in the destination directory.
2. Neither algorithm is atomic in both task and job commit.
3. V1 is atomic in task commits, but O(files) in its non-atomic job commit. It can recover from any job failure without having rerun all succeeded tasks, but not from a failure in job commit.
4. V2's job commit is a repeatable atomic O(1) operation, because it is a no-op. Task commits do the move/merge, which are O(file), make the output immediately visible, and as a consequence, mean that failure of a job means the output directory is in an unknown state.
5. Both algorithms depend on the filesystem having consistent listings and Create/Update/Delete operations
6. The routine to merge the output of a task to the destination is a real-world example of a co-recursive algorithm. These are so rare most developers don't even know the term for them -or have forgotten it.
7. At-most-once execution is guaranteed by having the tasks and AM failing when they recognise that they are in trouble.
8. The App Master refuses to commit a job if it hasn't had a heartbeat with the YARN Resource Manager within a specific time period. This stops it committing work if the network is partitioned and the AM/RM protocol fails...YARN may have considered the job dead and restarted it.
9. tasks commit iff they get permission from the AM; thus they will not attempt to commit if the network partitions.
10. if a task given permission to commit does not report a successful commit to the AM; the V1 algorithm can rerun the task; v2 must conclude its in an unknown state and abort the job.
11. Spark can commit using the Hadoop FileOutputCommitter; its Parquet support has some "special" code which refuses to work if the committer is not a subclass of ParquetOutputCommitter
. That is: it's special code makes it the hardest thing to bind to this: ORC, CSV, Avro: they all work out the box.
12. It adds the ability for tasks to provide extra data to its job driver for use in job commit; this allows committers to explicitly pass commit information directly to the driver, rather than indirectly via the (consistent) filesystem.
13. Everyone's code assumes that abort() completes in a bounded time, and does not ever throw that IOException its signature promises it can.
14. There's lots of cruft in the MRv2 codebase to keep the MRv1 code alive, which would be really good to delete

This means I get to argue the semantics of commit algorithms with people, as I know what the runtimes "really do", rather than believed by everyone who has neither implemented part of it or stepped throught the code in a debugger.

If we had some TLA+ specifications of filesystems and object stores, we could perhaps write the algorithms as PlusCal examples, but that needs someone with the skills and the time. I'd have to find the time to learn TLA+ properly as well as specify everything, so it won't be me.

Returning to the committers, what do they do which is so special?

They upload task output to the final destination paths in the tasks, but don't make the uploads visible until the job is committed.

No renames, no copies, no job-commit-time merges, and no data visible until job commit. Tasks which fail/fail to commit do not have any adverse side effects on the destination directories.

First, read S3A Committers: Architecture and Implementation.

Then, if that seems interesting look at the source.

A key feature is that we've snuck in to FileOutputFormat a mechanism to allow you to provide different committers for different filesystem schemas.

Normal file output formats (i.e. not Parquet) will automatically get the committer for the target filesystems, which, for S3A, can be changed from the default FileOutputCommitter to an S3A-specific one. And any other object store which also offers delayed materialization of uploaded data can implement their own and run it alongside the S3A ones, which will be something to keep the Azure, GCS and openstack teams busy, perhaps.

For now though: users of Hadoop can use Amazon S3 (or compatible services) as the direct destination of Hadoop and Spark workloads without any overheads of copying the data, and the ability to support failure recovery and speculative execution. I'm happy with that as a good first step.

(photo: street vendors at the Kenya/Tanzania Border)

Defending HDFS

It seems like everyone is picking on HDFS this week.

Why?


Some possibilities

1. There are some fundamental limitations of HDFS that suddenly everyone has noticed.
2. People who develop filesystems have noticed that Hadoop is becoming much more popular and wish to help by contributing code, tests and documentation to the open source platform, passing on their experiences running the Hadoop application stack and hardening Hadoop to the specific failure modes and other quirks of their filesystems.
3. Everyone whose line of business is selling storage infrastructure has realised that not only are they not getting new sales deals for hadoop clusters
4. Hadoop HDFS is making it is harder to justify the prices of "Big Iron" storage.

If you look at the press releases, action two, "test and improve the Hadoop stack" isn't being done by the "legacy" DFS vendors. These are the existing filesystems that are having Hadoop support retrofitted - usually by adding locality awareness to SAN-hosted location independence and a new filesystem driver with topology information for Hadoop. A key aid to making this possible is Hadoop's conscious decision to not support full Posix semantics, so it's easier to flip in new filesystems (a key one being Amazon S3's object store, which is also non-Posix).

I've looked at NetApp and Lustre before. Whamcloud must be happy that Intel bought them this week, and I look forward to consume beer with Eric Barton some time this week. I know they were looking at Hadoop integration -and have no idea what will happen now.

GPFS, well, I'll just note that they don't quote a price, instead say "have an account team contact you". If the acquisition process involves an account team, you know it wont be cents per GB. Client-side licensing is something I thought went away once you moved off Windows Server, but clearly not.

CleverSafe. This uses erasure coding as a way of storing data efficiently; it's a bit like parity encoding in RAID but not quite, because instead of the data being written to disks with parity blocks, the data gets split up into blocks and scattered through the DFS. Reading in the blocks involves pulling in multiple blocks and merging them. If you over-replicate the blocks you can get better IO bandwidth -grab the first ones coming in and discard the later ones.

Of course, you then end up with the bandwidth costs of pulling in everything over the network -you're in 10GbE territory and pretending there aren't locality issues, as well as worrying about bisection bandwidth between racks.

Or you go to some SAN system with its costs and limitations. I don't know what CleverSafe say here -potential customers should ask that. Some of the cloud block stores use e-coding; it keeps costs down and latency is something the customers have to take the hit on.

I actually think there could be some good opportunities to do something like this for cold data or stuff you want to replicate across sites: you'd spread enough of the blocks over 3 sites that you could rebuild them from any two, ideally.


Ceph: Readingthe papers, it's interesting. I haven't played with it or got any knowledge of its real-world limitations.


MapR. Not much to say there, except to note the quoted Hadoop scalability numbers aren't right. Today, Hadoop is in use in clusters up to 4000+ servers and 45+PB of storage (Yahoo!, Facebook). Those are real numbers, not projections from a spreadsheet.

There are multiple Hadoop clusters running at scales of 10-40 PB clusters, as well as lots of little ones from gigabytes up. From those large clusters, we in the Hadoop dev world have come across problems, problems we are, as an open source project, perfectly open about.

This does make it easy for anyone to point at the JIRA and say "look, the namenode can't...", or "look, the filesystem doesn't..." That' something we just have to recognise and accept.

Fine: other people can point to the large HDFS clusters and say "it has limits", but remember this: they are pointing at large HDFS clusters. Nobody is pointing at large Hadoop-on-filesystem-X clusters, for X != HDFS, -because there aren't any public instances of those.

All you get are proof of concepts, powerpoint and possibly clusters of a few hundred nodes -smaller than the test cluster I have access to.

If you are working on a DFS, well, Hadoop MapReduce is another use case you've got to deal with -and fast. The technical problem is straightforward -a new filesystem client class. The hard part is solving the economics problem of a filesystem that is designed to not only store data on standard servers and disks -but to do the computation there.

Any pure-storage story has to explain why you also need a separate rack or two of compute nodes, and why SAN Failures aren't considered a problem.

Then they have to answer a software issue: how can they be confident that the entire Hadoop software stack runs well on their filesystem? And if it doesn't, what processes have they in place to get errors corrected -including cases where the Hadoop-layer applications and libraries aren't working as expected?
Another issue is that some of the filesystems are closed source. That may be good from their business model perspective, but it means that all fixes are at the schedule of the sole organisation with access to the source. Not having any experience of those filesystems, I don't know whether or not that is an issue. All I can do is point out that it took Apple three years to make the rename() operation atomic, and hence compliant with POSIX.. Which is scary as I do use that OS on my non-Linux boxes. And before that, I used NTFS, which is probably worse.

Hadoop's development is in the open; security is already in HDFS (remember when that was a critique? A year ago?), HA is coming along nicely in the 1.x and 2.x lines. Scale limits? Most people aren't going to encounter them, so don't worry about that. Everyone who points to HDFS and says "HDFS can't" is going to have to find some new things to point too soon.

For anyone playing with alternate filesystems to hdfs:// file:// and s3://, then -here are some things to ask your vendor:

1. How do you qualify the Hadoop stack against your filesystem?
2. If there is an incompatibility, how does it get fixed?
3. Can I get the source, or is there an alternative way of getting an emergency fix out in a hurry?
4. What are the hardware costs for storage nodes?
5. What are the hardware costs for compute nodes?
6. What are the hardware costs for interconnect.
7. How can I incrementally expand the storage/compute infrastructure.
8. What are the licencing charges for storage and for each client wishing to access it?
9. What is required in terms of hardware support contracts (replacement disks on site etc), and cost of any non-optional software support contracts?
10. What other acquisition and operational costs are there?

I don't know the answers to those questions -they are things to ask the account teams. From the Hadoop perspective:

1. Qualification is done as part of the release process of the Hadoop artifacts.
2. Fix it in the source, convince someone else too (support contracts, etc)
3. Source? See http://hadoop.apache.org/
4. Server Hardware? SATA storage -servers depend on CPU and RAM you want.
5. Compute nodes? See above.
6. Interconnect? Good q. 2x1GbE getting more popular, I hear. 10 GbE still expensive
7. Adding new servers is easy, expanding the network may depend on the switches you have.
8. Licensing? Not for the Open Source bits.
9. H/W support: you need a strategy for the master nodes, inc. Namenode storage.
10. There's support licensing (which from Hortonworks is entirely optional), and the power budget of the servers.

Server power budget is something nobody is happy about. It's where reducing the space taken up by cold data would have add-on benefits -there's a Joule/bit/year cost for all data kept on spinning-media. The trouble is: there's no easy solution.

I look forward to a time in the future when solid state storage competes with HDD on a bit by bit basis, and that cold data can be moved to it -where wear levelling matters less precisely because it is cold- and warm data can live on it for speed of lookup as well as power. I don't know when that time will be -or even if.

[Artwork, Limited Press on #4 Hurlingham Road. A nice commissioned work.]

Hadoop in Cloud Infrastructures

People say "should you run Hadoop in the cloud?". I say "it depends".

I think there is value in Hadoop-in-cloud, I talked about doing it in 2010 at Berlin Buzzwords 2010; since then I've had more experience with using Hadoop and implementing cloud infrastructures.

1. If your data is stored in a cloud provider's storage infrastructure, doing the analysis locally is the only rational action. It's that "work near the data" philosophy.
2. If you are only doing some computation -say nightly- then you can rent some cluster time. Even if compute performance is worse, you can just rent some more machines to compensate.
3. You may be able to achieve better security through isolation of clusters (depends on your IaaS vendor's abilities).
4. No upfront capex; fund from ongoing revenue.
5. Easier to expand your cluster; no need to buy more racks, find more rack space.
6. You don't need to care about the problems of networking.
7. Less of a problem of heterogenous clusters if you expand later.

Against that

1. Cost of data storage can only increase at a rate proportional to ingress/retention rates.
2. Cost of cluster time increases at a rate proportional to analysis performed. There is no "spare cluster time" for low priority work.
3. Even if CPU time can scale up, IO rate of persistent data may not.
4. Hadoop contains lots of assumptions about running in a static infrastructure; it's scheduling and recovery algorithms assume this.

Some examples of where Hadoop's assumptions diverge from that of cloud infrastructures:

• HDFS assumes failures are independent, and places data accordingly (Google's Availability in Globally Distributed Storage Systems paper shows this doesn't hold in physical infrastructures, my notion of failure topologies expands on that)
• MR blacklists failing machines, rather than releasing them and requesting new ones.
• Worker nodes handle failure of master nodes by spinning on the hostname, not querying (dynamic) configuration data for new hostnames. Some of the HA HDFS may address that, I'm not tracking it enough.
• Topology scripts are static. I've been slowly tweaking topology logic in 0.23+ but haven't put the dynamicness in there yet (HDFS and MR cache (name->rack) mappings on the assumption that the data is coming from slow to exec scripts, not fast & refreshable in-VM data).
• Schedulers assume #of machines are static, don't allocate and release compute nodes based on demand and with knowledge of cost and quantum of CPU rentals. (I'm not sure quantum is the right term there, I mean the fact that VMs may be rented by the hour, 15 minutes, etc, so your scheduler should retain them for 59 minutes after acquiring them.
• Scheduling doesn't bill different users for their cluster use in a way that is easily mapped to cluster time.

A lot of these are tractable, you just have to put in the effort. The Stratosphere team in Berlin are doing lots of excellent work here, including taking a higher level query language and generating an execution plan that is IaaS aware -you can optimise for fast (many machines) or lower cost (use less machines more efficiently).

In comparison, a physical cluster:

• Offers a lower cost/TB of any corporate filestore to date other than people's desktop computers (which have a high TCO and power cost that is generally ignored), so enables you to store lots of stuff you would otherwise discard.
• Let's you choose the hardware optimised for your current and predicted workloads.
• Has free time for the low priority background work as well as the quicker queries that near-real-time UIs like.
• May be directly accessible from desktops in the organisation (depends on security model of cluster).
• Is easily hooked up to Jenkins infrastructure for execution of work as CI jobs.
• Let's you do fancy tricks like striping of different MR versions across the racks for in-rack locality and different sets of task trackers for foreground vs background work, and different JTs (reduces memory use, cost of failure, etc).
• Is way, way easier to hook up to internal databases, log feeds. To do ETL into your corporate oracle servers, you will need to run something behind the firewall to fetch it off the IaaS storage layer, rather than have your reducers push it to the RDBMS itself.

If you are generating data in house, in house clusters make a lot of sense.
This is why I say "it depends" -it depends on where you collect your data and what you plan to do with it.

As for the way Hadoop doesn't currently work so well in such infrastructures, well the code is there for people to fix. It's also a lot easier to test in-cloud behaviour, including resilience to failure, than it is with physical clusters.

[Photo: Descending Mt Ranier, 2000]

Solving the Netapp Open Solution for Hadoop Solutions Guide

I have been staring at the Netapp Open Solution for Hadoop Solutions Guide. I now have a headache.

Where do I begin? It's interesting to read a datasheet about Hadoop that is targeted at enterprise filesystem people. Not management "Hadoop solves all your problems", not developer "you have to rewrite all your apps", but something designed to convince people who worry about file systems that HDFS is a wrongness on the face of the planet. That's the only reason it would make sense. And like anyone trying to position a competing product to HDFS they have to completely discredit HDFS.

This paper, which tries to rejustify splitting the storage and compute platforms -the whole thing that the HDFS/Hadoop is designed to eliminate on cost grounds- has to try pretty hard.

In fact, I will say that after reading this paper that the MapR/EMC story makes a lot more sense. As what the Netapp paper is trying to say -unlike EMC-  is "the Big Data platform that Hadoop is wonderful, provided you ignore the cost model of SATA-on-server that HDFS and equivalent filesystems offer for topology-aware applications".

This must have been a hard article to write, so I must give the marketing team some credit for the attempt, even if they got it so wrong technically.


First: did we really need new acronyms? Really? By Page 4 there is a new one, "REPCOUNT", that I have never come across before. Then the acronym "HDFS3" that I first thought meant version 3 of HDFS, but before I could switch to JIRA and see if someone had closed an issue "design, implement and release V3 of HDFS", I see it really means "A file saved to HDFS with the block.replication.factor attribute set to 3", or more simply, 3x replicated blocks. No need to add another acronym to a document that is knee deep in them.


Now, for some more serious criticisms, starting at page 5: 

"For each 1 petabyte of data, 3 terabytes of raw capacity are needed".

If that were true, this would not be a criticism. No, it would be a sign of someone getting such a fantastic compression ration that Facebooks leading edge 30 PB server would fit into 30x3TB LFF HDDs, which you could fit into about 3U's worth of rack. If only. And don't forget that RAID uses 1.6X raw storage: it replicates too, just more efficiently.

Then there's the ingest problem. A whole section of this paper is dedicated to scaring people that they won't be able to pull in data fast enough because the network traffic will overload the system. We are not -outside my home network- using Ether over Power as the backplane any more. You can now get 46 servers with 12x3TB HDDs in them onto a single rack, with a 10GbE switch that run "with all the lights on". On a rack like that - which offers 500+TB of storage- you can sustain 10 GbE between any two servers. If the ingress and egress servers are hooked up to the same switch you could in theory move a Gigabyte per second between any two nodes or in and out the cluster. Nobody has an ingress right in that range, except maybe the physicists and their "odd" experiments. Everyone else can predict their ingress rate fairly simply, it is "look around at the amount of data you discard today". That's your ingress rate. Even a terabyte per day is pretty significant -yet on a 10GbE switched you could possibly import that from a single in-centre-off-HDFS-node in under 20 minutes. For anything out of the datacentre, your bandwidth will probably be less than 10 Gigabits, unless your name is Facebook, Yahoo!, Google or Microsoft.


Summary: ingress rate is unlikely to be bottlenecked by in-rack bandwidth even with 3x replication.

Next: Intra-HDFS bandwidth issues.

The paper says today that "A typical Hadoop node has 2GbE interfaces". No. A typical node tends to have a single 1GbE interface; bonded 2x1 GbE is rarer, as it's harder to commission.
Go that way and add on separate ToR switches and all concerns about "network switch failure" go away. You'd have to lose two network switches or power to the entire rack to trigger re-replication. At least NetApp did notice that 10GbE is going on the mainboard and said "Customers building modest clusters (of less than 128 nodes) with modern servers should consider 10GbaseT network topologies."

I'd say something different, which is "128 nodes is not a modest cluster". If you are building your rack from paired six-core CPUs with 12 LFF HDDs each with 3TB of SATA (how's that for acronyms?), then your cluster has 1536 individual cores. It will have 12*3*128 = 4608TB of storage: four petabytes. That's is not modest. That is something that can sort a petabyte of data with. Fit that up with a good 10GbE switch and enough RAM -and the latest 0.23 alpha release with Todd's CRC32c-in-SSE patch- and you could probably announce you have just won the petabyte sorting record.

Summary: "a 128 node cluster built of current CPUs and SATA storage is not modest".

It's a modest size -three racks- but it will probably be the largest file system your organisation owns. Think about that. And think about the fact that when people say "big data" what they mean is "lots of low value data". If you don't store all of it (aggressively compressed), you will regret it later.

Page 6: Drive failures. This  page and its successors wound me up so much I had to stop. You see, I've been slowly doing some changes to trunk and 0.23 to tune HDFS's failure handling here, based on papers on failures by Google, Microsoft and -um- Netapp. This means I am starting to know more about the subject, and defend my statements.

The Netapp paper discusses the consequences of HDD Failure rates using the failure rate on 5K-node cluster to scare people. Yes, there will be a high failure rate there, but it's a background noise. The latest version of Hadoop 0.20.20x doesn't  require a DataNode restart when a disk fails -take it away and only that 1-3 TB of data fails. When a disk fails -and this is where whoever wrote the paper really misses the point - that 2TB of missing data is still scattered across all other clusters in the rack.

If you have 128 servers in your "modest" cluster, 2TB disks and a block size of 256MB, then there were about 8000 blocks in the disk, which are now scattered across (128-1) servers. Sixtyish blocks per server. With twelve disks per server, that's about five blocks per SATA disk (=2500 MB). Even if -as claimed- the throughput of a single SATA disk is 30MB/s, those five blocks will take under two minutes to be read off disk. I'm not going to follow this chain through to network traffic as you'd also have to factor in the fact that servers are reading network packets from blocks being replicated to it at the same time and saving them to disk too (it should expect 60 blocks incoming), but the key point is this: on a 128 node cluster, the loss of a single 2TB cluster will generate a blip of network traffic, and your cluster will carry on as normal. No need for "24-hour staffing of data center technicians". No need for the ops teams to even get texted when a disk fails. Yes, loss of a full 36TB server is a bit more dramatic, but with 10 GbE it's still manageable. This paper is just trying to scare you based on a failure to understand how block replication is implicit striping of a file across the entire cluster, and they haven't looked at Hadoop's source to see what it does on a failure. Hint: look at BlockManager.java. I have.


To summarise:


The infrastructure handles failures, the apps are designed for it. This is not your old enterprise infrastructure. We have moved out of the enterprise world of caring about every disk that fails, and into a world of statistics. 


I'm going to stop here. I can't be bothered to read the low level details if the high level stuff is so painfully wrong, either through inadequate technical review of the marketing team's world view, or deliberately through a failure to understand HDFS.

[Update 21:57 GMT]. Actually it is weirder than I first thought. This is still HDFS, just running on more expensive hardware. You get the (current) HDFS limitations: no native filesystem mounting, a namenode to care about, security on a par with NFS, without the cost savings of pure-SATA-no-licensing-fees. Instead you have to use RAID everywhere, which not only bumps up your cost of storage, puts you at risk of RAID controller failure and errors in the OS drivers for those controller (hence their strict rules about which Linux releases to trust). If you do follow their recommendations and rely on hardware for data integrity, you've cut down the probability of node-local job execution, so all FUD about replication traffic is now moot as at least 1/3 more of your tasks will be running remote -possibly even with the Fair Scheduler, which waits for a bit to see if a local slot becomes free. What they are doing then is adding some HA hardware underneath a filesystem that is designed to give strong availability out of medium availability hardware. I have seen such a design before, and thought it sucked then too.  Information week says this is a response to EMC, but it looks more like NetApp's strategy to stay relevant, and Cloudera are partnering with them as NetApp offered them money and if it sells into more "enterprise customers" then why not? With the extra hardware costs of NetApp the cloudera licenses will look better value, and clearly both NetApp and their customers are in need of the hand-holding that Cloudera can offer.

I just wish someone from Cloudera had reviewed that solutions paper for technical validity before it got published.




[Artwork: ARYZ, Nelson Street]