May 13, 2009

Hadoop computes the 10^15+1st bit of π

I used Yahoo's Hadoop clusters to compute the 1,000,000,000,000,001st bit of π. The 7 hexadecimal digits of π starting at the 10^15+1 bit are:

6216B06

Although Hadoop is primarily used for data-intensive applications, it can also be used to run CPU-intensive jobs on many machines. Computing a range of bytes in π using a BPP-type formula, requires a lot of arithmetic operations and therefore CPU, but not much storage. When computing the 10^15+1st bit of π, the first 30% of the computation was done in idle slots of our Hadoop clusters spread over 20 days. The remaining 70% was finished over a weekend on the Hammer cluster, which was also used for the petabyte sort benchmark.

This validates the results calculated by PiHex, which took more than 2 years on 1734 computers from 56 different countries.

My program was written entirely in Java and ran on Hadoop 0.20. An earlier version is checked in as a Hadoop example named BaileyBorwinPlouffe. The new code will be uploaded soon.

-- Tsz Wo (Nicholas), Sze

Posted by aanand at 8:00 AM | Comments (3) | TrackBack | Permalink

May 11, 2009

Hadoop Sorts a Petabyte in 16.25 Hours and a Terabyte in 62 Seconds

We used Apache Hadoop to compete in Jim Gray's Sort benchmark. Jim's Gray's sort benchmark consists of a set of many related benchmarks, each with their own rules. All of the sort benchmarks measure the time to sort different numbers of 100 byte records. The first 10 bytes of each record is the key and the rest is the value. The minute sort must finish end to end in less than a minute. The Gray sort must sort more than 100 terabytes and must run for at least an hour. The best times we observed were:

Within the rules for the 2009 Gray sort, our 500 GB sort set a new record for the minute sort and the 100 TB sort set a new record of 0.578 TB/minute. The 1 PB sort ran after the 2009 deadline, but improves the speed to 1.03 TB/minute. The 62 second terabyte sort would have set a new record, but the terabyte benchmark that we won last year has been retired. (Clearly the minute sort and terabyte sort are rapidly converging, and thus it is not a loss.) One piece of trivia is that only the petabyte dataset had any duplicate keys (40 of them).

We ran our benchmarks on Yahoo's Hammer cluster. Hammer's hardware is very similar to the hardware that we used in last year's terabyte sort. The hardware and operating system details are:

• approximately 3800 nodes (in such a large cluster, nodes are always down)
• 2 quad core Xeons @ 2.5ghz per node
• 4 SATA disks per node
• 8G RAM per node (upgraded to 16GB before the petabyte sort)
• 1 gigabit ethernet on each node
• 40 nodes per rack
• 8 gigabit ethernet uplinks from each rack to the core
• Red Hat Enterprise Linux Server Release 5.1 (kernel 2.6.18)
• Sun Java JDK (1.6.0_05-b13 and 1.6.0_13-b03) (32 and 64 bit)

We hit a JVM bug that caused a core dump in 1.6.0_05-b13 on the larger sorts (100TB and 1PB) and switched over to the later JVM, which resolved the issue. For the larger sorts, we used 64 bit JVMs for the Name Node and Job Tracker.

Because the smaller sorts needed lower latency and faster network, we only used part of the cluster for those runs. In particular, instead of our normal 5:1 over subscription between racks, we limited it to 16 nodes in each rack for a 2:1 over subscription. The smaller runs can also use output replication of 1, because they only take minutes to run and run on smaller clusters, the likelihood of a node failing is fairly low. On the larger runs, failure is expected and thus replication of 2 is required. HDFS protects against data loss during rack failure by writing the second replica on a different rack and thus writing the second replica is relatively slow.

Below are the timelines for the jobs counting from the job submission at the Job Tracker. The diagrams show the number of tasks running at each point in time. While maps only have a single phase, the reduces have three: shuffle, merge, and reduce. The shuffle is the transfer of the data from the maps. Merge doesn't happen in these benchmarks, because none of the reduces need multiple levels of merges. Finally, the reduce phase is where the final merge and writing to HDFS happens. I've also included a category named waste that represents task attempts that were running, but ended up either failing, or being killed (often as speculatively executed task attempts).

If you compare this years charts to last year's, you'll notice that tasks are launching much faster now. Last year we only launched one task per heartbeat, so it took 40 seconds to get all of the tasks launched. Now, Hadoop will fill up a Task Tracker in a single heartbeat. Reducing that job launch overhead is very important for getting runs under a minute.

As with last year, we ran with significantly larger tasks than the defaults for Hadoop. Even with the new more aggressive shuffle, minimizing the number of transfers (maps * reduces) is very important to the performance of the job. Notice that in the petabyte sort, each map is processing 15 GB instead of the default 128 MB and each reduce is handling 50 GB. When we ran the petabyte with more typical values 1.5 GB / map, it took 40 hours to finish. Therefore, to increase throughput, it makes sense to consider increasing the default block size, which translates into the default map size, to at least up to 1 GB.

We used a branch of trunk with some modifications that will be pushed back into trunk. The primary ones are that we reimplemented shuffle to re-use connections, and we reduced latencies and made timeouts configurable. More details including the changes we made to Hadoop are available in our report on the results.

-- Owen O'Malley and Arun Murthy

Posted by aanand at 3:00 PM | Comments (18) | TrackBack | Permalink

May 5, 2009

Hadoop Summit 2009 - Open for registration

This year’s Hadoop Summit is confirmed for June 10th at the Santa Clara Marriott, and is now open for registration.

We have a packed agenda, with three tracks – for developers, administrators, and one focused on new and innovative applications using Hadoop. The presentations include talks from Amazon, IBM, Sun, Cloudera, Facebook, HP, Microsoft, and the Yahoo! team, as well as leading universities including UC Berkeley, CMU, Cornell, U of Maryland, U of Nebraska and SUNY.

From our experience last year with the rush for seats, we would encourage people to register early.

Posted by aanand at 2:59 PM | Comments (1) | TrackBack | Permalink

May 4, 2009

Using ZooKeeper to tame system test for large-scale services

In the last release I took on the task of setting up a true system test environment for Apache ZooKeeper. Our previous environment ran the system test in a single JVM instance, which meant that there were some test scenarios that we just couldn't reproduce. In this new environment we wanted to be able to run tests across multiple hosts and deal with different numbers of machines and cluster environments.

My first attempt used ssh and scripts to fire off servers and clients on a cluster of machines. I soon realized that there was a significant amount of hardcoded configuration and envirnment
assumptions that would make the setup inflexible and impractical. So, I started over from scratch.

For the system test I wanted to use some set of machines, M, to host some number of ZooKeeper servers, S, and some number of clients C. Ideally the system test could just discover M at runtime, startup S servers on some subset of M and then startup C clients on the rest. I realized this is a perfect application for ZooKeeper. I also realized that such a use case can apply to more than just system tests.

What I needed to do was very similar to the rendezvous problem: you have two processes that startup with no knowledge of where each other is running and they need to find each other. Common solutions to this involve service discovery by broadcasting on well known ports, but for our network topologies broadcast based solutions will not work. But, ZooKeeper makes rendezvous easy. Each process creates an ephemeral znode with contact information as a child of a previously agreed upon znode (the rendezvous point). For example, let's use "/systest/available" as the rendezvous point. If four hosts startup, /systest/available would have the following children:

/systest/available/host1
/systest/available/host2
/systest/available/host3
/systest/available/host4

I can now see what machines are available for use in our system test.

Now that I know the machines that I can use, I want to start assigning work to them. The Java process that created /systest/available/host1 also created /systest/assignments/host1 and is watching that znode for children to appear. So when I need to assign task1 to a machine and assuming host1 is the least loaded machine, the system test will create the znode /systest/assignments/host1/task1, which will contain information about what class to instantiation and start as well as configuration parameters. The creation of /systest/assignments/host1/task1 will trigger a watch on the agent process running on that machine. The agent running on that machine will read the information about the
new task and start it. I also use the task znode to stop tasks, change their configuration, and get task status.

The end result is a very clean system test environment. The test classes themselves just interact through ZooKeeper, so there aren't any hardcoded assumptions about the environment needed. This also serves as a nice illustration of how Apache ZooKeeper can be used for more than just leader election and locking. Of course the next step would be to combine this scheme with something like OSGi to provide automatic management of the class dependences of the instances.

See the Apache ZooKeeper website to give it a try. The system test describe here is in the latest release: 3.1.1

Ben Reed
Yahoo!

Posted by aanand at 3:21 PM | Comments (2) | TrackBack | Permalink