Hadoop in Practice - "Applied Hadoop"

Recent train journeys to and from London have given me a chance to get the laptop out and read some of the collected PDFs of things I know I should read.



I was given a PDF copy of Hadoop in Practice [Holmes, 2012] on account the fact of I'd intermittently been in the preview program -but I'd not looked at it in any detail until now. The (unexpectedly ) slow train journeys to and from London have been an opportunity to unfold the laptop and read it -and, at home, while I wait for EC2 do respond to whirr requests, to read it to the end -though not in as much detail as it deserves.

The key premises of this book are

1.  You've read one of the general purpose "this is Hadoop" books -either the Definitive Guide or Hadoop in Action.
2.  You want to do more with Hadoop.
3.  You aren't concerned with managing the cluster.
4.  You are concerned about how to integrate a Hadoop cluster with the rest of your organisation.

#3 means that there's nothing here on metrics, logging or low-level things. This is a book for developers and (yes) architects; less the operations people. Even so, the sections on integration with other systems, especially hooking up to log sources and databases that they need to know about.
Although it starts off with a quick overview of Hadoop and MapReduce, internals -such as how HDFS works- are relegated to appendices for the curious. Instead, the first detailed chapter looks at Ingress and Egress, or, so as not to scare readers, "Moving Data in and Out Hadoop", looking mostly at Flume, mentioning Chukwa and Scribe, and then into using Oozie-scheduled MR Jobs to pull data -something in an example in the book.

It doesn't delve into the aspects of this problem you'd need to worry about in production -data rates, the risk that MR pull jobs can either overload the endpoints or, unless they are split up well, can create imbalanced filesystems. Ops problems -or just too much to worry about right now.  What it does do is show why a workflow engine like Oozie is useful: to automate the regular work.

It glues the Hadoop ecosystem together. Want to parse XML? grab the XML input reader from Mahout.  Want to work with JSON? Twitter's Elephant Bird… etc. In fact the serialization chapter went into the depths XML and JSON parsing -and showed the problems, so justifying the next stage: Protobuf, Avro and Thrift.

There's a chapter on tuning problems which focuses more on code-level issues than hardware; this is where the line between ops & developers gets blurred. I think I'd have approached the problem in a different order, but the tactics are all valid.

Installation-wise, Alex points everyone at a version of CDH without LZO support; he has to talk people through building it. I don't know where Cloudera stand on that, as I know yum -y install hadoop-lzo works for HDP., and is up there with hadoop-native hadoop-pipes hadoop-libhdfs  and snappy as RPMs to add (update: see below). I'd have liked to seen bigtop as the centre of the universe, so be more neutral -something to hope for in the second edition

There's a few chapters on "data science" stuff: bloom filters, simple graph operations, R & Hadoop integration. I get the feeling that this section is very handy if you know your statistics and want to do work with a new toolset. The problem I have there is a personal one: I've forgotten too much of what I new about statistics. min, max, mean, Poisson, Gaussian and Weibull distributions;the notion of Markov chains are all concepts I know about -but ask me the equation behind a Poisson distribution and I stare as blankly at the questioner as our pet rabbit does when asked why he's been chewing power cables: there's no comprehension going on behind the eyeballs. I really need something that covers "statistics for people who used to know it vaguely -using R & Pig as the tools". There's a good argument for all developers to know more stats. This book isn't that -it does assume you know your statistics, at least better than I do.

Alex Holmes delves into MRUnit, which is a good way for unit testing individual operations. I tend to do something else: MiniMRCluster -but that one, while more authentic, can push problems onto different threads and so make it harder to identify root causes of problems -or isolate tests. MRUnit doesn't have that flaw, and nor does LocalJobRunner -which also gets coverage. The only thing that grated against me there was that the tests were done in Java -I've been using Groovy as my test language for the whole of 2012, and sheer verbosity of setting up lists in Java, and the crudeness of JUnit's assertions compared to Groovy's assert statements is painful to look at.

For anyone who's never used Groovy, its assert statement takes advantage of the compile-on-demand features of the language. On an assertion failure, the output walks through the entire expression tree, evaluates every part in turn and gives you the complete tree for your debugging pleasure. You can write one all-encompassing assertion, rather than break down each part of a large query into various assertNotNull, assertTrue, assertEquals calls -and if the single assert fails, there should be enough information for you to track down the cause.  That's why I like testing in Groovy, irrespective of whether or not your production code is in Java.

Other points: the ebook comes with your email address at the bottom, but no epub-esque security. This works on your Linux workstation as well as whatever tablet you choose to own -and relies on publicity & guilt to stop sharing. Which is probably a good strategy. That eBook comes with a feature I've never seen before: the page numbers in the contents match exactly the page numbers in the book -there must be some Framemaker magic that tells Preview &c the offset to apply after the user hits the "go to page" button.

Summary: this isn't book for newbies -precisely because it delves into Applied Hadoop. Even so, it's something you ought to have to hand, just so you aren't one of the people posting questions to user@hadoop that everyone else stares and generally refuses to answer., the "hello, I have got a pseudo-distributed cluster that cannot find localhost, here is the screenshot of the DOS console, please help!!!" -while forgetting to even include the screenshot of their hadoop.bat command line failing as they've forgotten to do something foundational like install Java.

Everyone but @castagna will learn something new -in fact maybe even him, because he needs something to read on test runs and trains to London (which is where I'm writing this, somewhere between Reading and London Paddington)

Update: Eric Sammer says of the LZO thing "hadoop-lzo in cdh, it's because of license concerns that we don't distrib."

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]

Towards a Topology of Failure

The Apache community is not just the mailing lists and the get togethers: it is the planet apache aggregate blog; this lets other committers share their thoughts -and look at yours. This makes for unexpected connections.

After I posted my comments on Availability in Globally Distributed Storage Systems, Phil Steitz posted wonderful article on the mathematics behind it. This impressed me, not least because of his ability to get TeX-grade equations into HTML. What he did do was look at the real theory behind it, and even attempted to implement Dynamic Programming solution the problem.

I'm not going to be that ambitious, but I will try and link this paper -and the other ones on server failures, into a new concept, "Failure Topology". This is an excessively pretentious phrase, but I like it -if it ever takes off I can claim I was the first person to use it, as I can do with "continuous deployment"

The key concept of Failure Topology is that failures of systems often follow topologies. Rack outages can be caused by rack-level upgrades. Switch-level outages can take out 1* racks and are are driven by the network topology. Power outages can be caused by the failure of power supplies to specific servers, sets of servers, specific racks or even quadrants of a datacentre. There's the also the notion of specific hardware instances, such as server batches with the same run of HDDs or CPU revisions.

Failure topologies, then, are maps of the computing infrastructure that show how these things are related, and where the risk lies. A power topology would be a map of the power input to the datacentre. We have the switch topology for Hadoop, but it is optimised for network efficiency, rather than looking at the risk of switch failure. A failure-aware topology would need to know which racks were protected by duplicate ToR switches and view them as less at risk then single-switch racks. Across multiple sites you'd need to look at the regional power grids, the different telcos. Then is the politics overlay: what government controls the datacentre sites; whether or not that government is part of the EU and hence has data protection rights, or whether there's some DMCA-style takedown rules.

You'd also need to look at physical issues: fault lines, whether the sites were downwind of Mt St Helen's class volcanoes. That goes from abstract topologies to physical maps.

What does all this mean? Well, in Disk-Locality in Datacenter Computing Considered Irrelevant, Ganesha Ananthanarayanan argues that as switch and backplane bandwidth increases you don't have to worry about where your code runs relative to the data. I concur: with 10GbE and emerging backplanes, network bandwidth means that switch-local vs switch-remote will become less important. Which means you can stop worrying about Hadoop topology scripts driving code execution policies. Doing this now opens a new possibility:

Writing a script to model the failure topology of the datacentre.

You want to move from a simple "/switch2/node21" map to one that includes power sources, switches, racks, shared-PSUs in servers, something like "/ups1/switch2/rack3/psu2/node21". This models not the network hierarchy, but the failure topology of the site. Admittedly, it assumes that switches and racks share the same UPS, but if the switch power source goes away, the rack is partitioned and effectively offline anyway -so you may as well do that.

I haven't played with this yet, but as I progress with my patch to allow Hadoop to focus on keeping all blocks on separate racks for availability, this failure topology notion could be the longer term topology that the ops team need to define.

[Photo: sunrise from the snowhole on the (dormant) volcano Mt Hood -not far from Google's Dalles datacentre]

301 Moved Permanently

I've moved my blog from 1060.org here. Why? Well, the team at 1060research have pushed out a new release of their NetKernel product, on which the blog was running, and if I wanted to retain the URLs I'd have to upgrade the code myself.  Being lazy and all, I opted not to.

What have I been up to since going offline

1. On twitter @steveloughran. Idle chatter.
2. Finishing up some major project at work that has kept be busy for the past 12-18 months. I am feeling more relaxed now. 
3. Coding in Groovy. It's like Java only better, and trivial to switch between the too. There's great IDE support in IntelliJ IDEA too.
4. Doing some proper Computer Science stuff, as opposed to Software Engineering.
5. Paper Reading. This may seem dull, but there is a lot of interesting stuff out there. If you spend too much time knee-deep in various projects' codebases you get sucked into various issues (log4j configuration etc), and out of touch with higher level problems. 
6. Holiday on the south coast of England. Not far; lazy. I lost a camera, which is a pity, but I've replaced it already.

My most recent set of readings was all the big datacentre papers on DRAM and HDD failures. A good way to revise on concepts like Poisson Distributions, Gauss Distributions, Weibull Distributions (which I never knew of before), and lots of other math-hard problems. It's shocking how much maths I have forgotten; I keep seeing these bits on the paper where they do integration or differentiation and say "clearly then" and the words mean nothing to me. I will have to revise some fundamentals.

With the work I've been doing wrapping up, I'm hoping to get more involved in Hadoop and Hadoop related work. I don't have a schedule for that -I'm just reading now.