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)

Stocator: A High Performance Object Store Connector for Spark

IBM have published a lovely paper on their Stocator 0-rename committer for Spark

Stocator is:

1. An extended Swift client
2. magic in their FS to redirect mkdir and file PUT/HEAD/GET calls under the normal MRv1 __temporary paths to new paths in the dest dir
3. generating dest/part-0000 filenames using the attempt & task attempt ID to guarantee uniqueness and to ease cleanup: restarted jobs can delete the old attempts
4. Commit performance comes from eliminating the COPY, which is O(data),
5. And from tuning back the number of HTTP requests (probes for directories, mkdir 0 byte entries, deleting them)
6. Failure recovery comes from explicit names of output files. (note, avoiding any saving of shuffle files, which this wouldn't work with...spark can do that in memory)
7. They add summary data in the _SUCCESS file to list the files written & so work out what happened (though they don't actually use this data, instead relying on their swift service offering list consistency). (I've been doing something similar, primarily for testing & collection of statistics).

Page 10 has their benchmarks, all; of which are against an IBM storage system, not real amazon S3 with its different latencies and performance.

Table 5: Average run time

	Read-Only 50GB	Read-Only 500GB	Teragen	Copy	Wordcount	Terasort	TPC-DS
Hadoop-Swift Base	37.80±0.48	393.10±0.92	624.60±4.00	622.10±13.52	244.10±17.72	681.90±6.10	101.50±1.50
S3a Base	33.30±0.42	254.80±4.00	699.50±8.40	705.10±8.50	193.50±1.80	746.00±7.20	104.50±2.20
Stocator	34.60±0.56	254.10±5.12	38.80±1.40	68.20±0.80	106.60±1.40	84.20±2.04	111.40±1.68
Hadoop-Swift Cv2	37.10±0.54	395.00±0.80	171.30±6.36	175.20±6.40	166.90±2.06	222.70±7.30	102.30±1.16
S3a Cv2	35.30±0.70	255.10±5.52	169.70±4.64	185.40±7.00	111.90±2.08	221.90±6.66	104.00±2.20
S3a Cv2 + FU	35.20±0.48	254.20±5.04	56.80±1.04	86.50±1.00	112.00±2.40	105.20±3.28	103.10±2.14

The S3a is the 2.7.x version, which has the stabilisation enough to be usable with Thomas Demoor's fast output stream (HADOOP-11183). That stream buffers in RAM & initiates the multipart upload once the block size threshold is reached. Provided you can upload data faster than you run out of RAM, it avoids the log waits at the end of close() calls, so has significant speedup. (The fast output stream has evolved into the S3ABlockOutput Stream (HADOOP-13560) which can buffer off heap and to HDD, and which will become the sole output stream once the great cruft cull of HADOOP-14738 goes in)

That means in the doc, "FU" == fast upload, == incremental upload & RAM storage. The default for S3A will become HDD storage, as unless you have a very fast pipe to a compatible S3 store, it's easy to overload the memory

Cv2 means MRv2 committer, the one which does  single rename operation on task commit (here the COPY), rather than one in task commit to promote that attempt, and then another in job commit to finalise the entire job. So only: one copy of every byte PUT, rather than 2, and the COPY calls can run in parallel, often off the critical path

 Table 6: Workload speedups when using Stocator

	Read-Only 50GB	Read-Only 500GB	Teragen	Copy	Wordcount	Terasort	TPC-DS
Hadoop-Swift Base	x1.09	x1.55	x16.09	x9.12	x2.29	x8.10	x0.91
S3a Base	x0.96	x1.00	x18.03	x10.33	x1.82	x8.86	x0.94
Stocator	x1	x1	x1	x1	x1	x1	x1
Hadoop-Swift Cv2	x1.07	x1.55	x4.41	x2.57	x1.57	x2.64	x0.92
S3a Cv2	x1.02	x1.00	x4.37	x2.72	x1.05	x2.64	x0.93
S3a Cv2 + FU	x1.02	x1.00	x1.46	x1.27	x1.05	x1.25	x0.93

Their TCP-DS benchmarks show that stocator & swift is slower than TCP-DS Hadoop 2.7 S3a + Fast upload & MRv2 commit. Which means that (a) the Hadoop swift connector is pretty underperforming and (b) with fadvise=random and columnar data (ORC, Parquet) that speedup alone will give better numbers than swift & stocator. (Also shows how much the TCP-DS Benchmarks are IO heavy rather than output heavy the way the tera-x benchmarks are).

As the co-author of that original swift connector then, what the IBM paper is saying is "our zero rename commit just about compensates for the functional but utterly underperformant code Steve wrote in 2013 and gives us equivalent numbers to 2016 FS connectors by Steve and others, before they started the serious work on S3A speedup". Oh, and we used some of Steve's code to test it, removing the ASF headers.

Note that as the IBM endpoint is neither the classic python Openstack swift or Amazon's real S3, it won't exhibit the real issues those two have. Swift has the worst update inconsistency I've ever seen (i.e repeatable whenever I overwrote a large file with a smaller one), and aggressive throttling even of the DELETE calls in test teardown. AWS S3 has its own issues, not just in list inconsistency, but serious latency of HEAD/GET requests, as they always go through the S3 load balancers. That is, I would hope that IBM's storage offers significantly better numbers than you get over long-haul S3 connections. Although it'd be hard (impossible) to do a consistent test there, I 'd fear in-EC2 performance numbers to be actually worse than that measures.

I might post something faulting the paper, but maybe I'll should to do a benchmark of my new committer first. For now though, my critique of both the swift:// and s3a:// clients is as follows

Unless the storage services guarantees consistency of listing along with other operations, you can't use any of the MR commit algorithms to reliably commit work. So performance is moot. Here IBM do have a consistent store, so you can start to look at performance rather than just functionality. And as they note, committers which work with object store semantics are the way to do this: for operations like this you need the atomic operations of the store, not mocked operations in the client.

People who complain about the performance of using swift or s3a as a destination are blisfully unaware of the key issue: the risk of data loss due inconsistencies. Stocator solves both issues at once.

Anyway, means we should be planning a paper or two on our work too, maybe even start by doing something about random IO and object storage, as in "what can you do for and in columnar storage formats to make them work better in a world where a seek()+ read is potentially a new HTTP request."

(picture: parakeet behind Picton Street)

Nobody ever got fired for using Hadoop on a cluster

Over a weekend in London I enjoyed reading a recent Microsoft Research paper from Rowstron et al., Nobody ever got fired for using Hadoop on a cluster.

The key points they make are

1. A couple of hundred GB of DRAM costs less than a new server
2. Public stats show that most Hadoop analysis jobs use only a few tens of GB.
3. If you can load all the data into DRAM on a single server, you can do far more sophisticated and stateful algorithms than MapReduce

They then go on to demonstrate this with some algorithms and examples examples that show that you can do analysis in RAM way, way faster than by streaming HDD data past many CPUs.


This paper makes a good point. Stateless MapReduce isn't ideal way to work with things, which is why counters (cause JT scale problems), per-task state (dangerous but convenient) and multiple iterations (purer; inefficient) come out to play. It's why graph stuff is good.

Think for a moment though, what Hadoop HDFS+MR delivers.

• Very low cost storage of TB to PB of data -letting you keep all the historical data for analysis.
• I/O bandwidth that scales with the #of HDDs.
• An algorithm that, being isolated and stateless, is embarassingly parallel
• A buffered execution process that permits recovery from servers that fail
• An execution engine that can distribute work across tens, hundreds or even thousands of machines.

HDFS delivers the storage, MapReduce provides a way to get at the data. As Rowstron and colleagues note, for "small" datasets, datasets that fit into RAM, it's not always the best approach. Furthermore, that falling cost of DRAM means that you can start predicting cost/GB of RAM in future, and should start thinking "what can I do with 256GB RAM on a single server?" The paper asks the question: Why then, Hadoop? Well, one reason the paper authors note themselves:

The load time is significant, representing 20–25% of the total execution time. This suggests that as we scale up single servers by increasing memory sizes and core counts, we will need to simultaneously scale up I/O bandwidth to avoid this becoming a bottleneck.

A server with 16 SFF HDDs can give you 16 TB of storage today; 32 TB in the future, probably matched by 32 cores at that time. The IO bandwidth, even with those 16 disks, will be a tenth of what you get from ten such servers. It's the IO bandwidth that was a driver for MapReduce -as the original Google paper points out.  Observing in a 2012 paper that IO bandwidth was lagging DRAM isn't new -that's a complaint going back to the late 1980s.

If you want great IO bandwidth from HDDs, you need lots of them in parallel. RAID-5 filesystems with striped storage deliver this at a price; HDFS delivers it at a tangibly lower price. As the cost of SDDs falls, when they get integrated into the motherboards, you'll get something with better bandwidth and latency numbers (I'm ignoring wear levelling here, and hoping that at the right price point SSD could be used for cold data as well as warm data). SSD at the price/GB of today's HDDs would let you store hundreds of TB in servers, transform the power budget of a Hadoop cluster, and make random access much less expensive. That could be a big change.

Even with SSDs, you need lots in parallel to get the bandwidth -more than a single server's storage capacity. If, in, say 5-10 years you could get away with a "small" cluster with a few tens of machines, ideally SSD storage, and lots of DRAM per server, you'd get an interesting setup. A small enough set of machines that a failure would be significantly less likely, changing the policy needed to handle failures. Less demand for stateless operations and use of persistent storage to propagate results; more value in streaming across stages. Less bandwidth problems, especially with multiple 10GBe links to every node. Lots of DRAM and CPUs.

This would be a nice "Hadoop 3.x" cluster.

Use an SSD-aware descendent of HDFS that was wear-leveling aware, maybe mix HDD for cold storage, SSD for newly added data. Execute work that used RAM not just for the computations, but for the storage of results. Maybe use some of that RAM for replicated storage -as work is finished, forward it to other nodes that just keep it in RAM for the next stages in a batch mode, stream directly to them for other uses. You'd gain storage capacity and bandwidth that a single server will always lag compared to a set of servers, while being able to run algorithms that can be more stateful.

In this world, for single rack clusters, you'd care less about disk locality (the network can handle that better than before), and more about which tier of storage the data was. Which is what Ananthanarayanan et al., argued in Disk-Locality in Datacenter Computing Considered Irrelevant. You could view that rack as storage system with varying degrees of latency to files, request server capacity 64 GB at a time. The latter, of course, is what the YARN RM lets you do, though currently its memory slots are measured in smaller numbers like 4-8GB.

YARN could work as a scheduler here -with RAM-centric algorithms running in it. What you'd have to do is ensure that most of the blocks of a file is stored in the same rack (trivial in a single-rack system), so that all bandwidth consumed is rack-local. Then bring up the job on a single machine, ask for the data and hope that network bandwidth coming off each node is adequate for the traffic generated by this job and all the others. Because at load time, there will be a lot of network IO. Provided that data load is only a small fraction of the analysis -the up front load time- this may be possible.

It seems to me that the best way to keep that network bandwidth down would be to store the data in RAM for a series of queries. You'd bring up something that acted as an in-RAM cache of the chosen dataset(s), then run multiple operations against it -either sequential or, better yet, parallelised across the many cores in that server. Because there'd be no seek time penalty, you can do work in the way the algorithm wants, not sequentially. Because it's in DRAM, different threads could work across the entire set simultaneously.

Yes, you could have fun here.

People have already recognised that trend towards hundreds of GB of RAM and tens of cores -the graph algorithms are being written for that. Yet they also benefit from having more than one server, as that helps them scale beyond a single server.

Being able to host a single-machine node with 200GB of data is something HDFS+YARN would support, but so would GPFS+Platform -so where's the compelling reason for Hadoop? The answer has to be that which today justifies running Giraph or Hama in the cluster -for specific analyses that can be performed with data generated by other Hadoop code, and so that the output can be processed downstream within the Hadoop layers.

For the MS Research example, you'd run an initial MR job to get the subsets of the data you want to work with out of the larger archives, then feed that data into the allocated RAM-analysis node. If that data could be streamed in during the reduce phase, network bandwidth would be less, though restart costs higher. The analysis node could be used for specific algorithms that know that access to any part of the local data is at RAM speeds, but that any other HDFS input file is available if absolutely necessary. HDFS and HBase could both be output points.

That's the way to view this world: not an either/or situation, which is where a lot of the Anti-MapReduce stories seem to start off, but with the question "what can you do here?", where the here is "a cluster with petabytes of data able to analyse it with MapReduce and Graph infrastructures -and able to host other analysis algorithms too?". Especially as once that cluster uses YARN to manage resources, you can run new code without having to buy new machines or hide the logic in long lived Map tasks.

The challenge then becomes, not "how to process all the data loaded into RAM on a server", but, "how to work with data that is stored in RAM across a small set of servers?" -the algorithm problem. At the infrastructure level, "how to effectively place data and schedule work for such algorithms, especially if the no of servers/job and duration is such that failures can be handled by checkpointing rather than restart that shard of the job". And "we can rent extra CPU time off the IaaS layer"

Finally, for people building up clusters, that perennial question crops up: many small servers vs fewer larger servers? I actually think the larger servers have a good story, provide you stick to servers with affordable ASPs and power budgets. More chance of local data, more flexible resource allocation (you could run something that used 128GB of RAM and 8 cores), and less machines to worry about. The cost: storage capacity and bandwidth; the failure of a high-storage-capacity node generates more traffic. Oh, and you need to worry about network load more too, but less machines reduces the cost of faster switches.

[Photo: the neolithic Avebury stone circle, midway between Bristol and London]

[updated: reread the text and cleaned up]