Yahoo! Hadoop Blog

August HUG Recap

Wed, 01 Sep 2010 11:00:00 -0800

Thanks to the around 175 developers who came to Yahoo! recently for our monthly Hadoop User Group meeting. The energy in the packed room was phenomenal, and conversations continued long after the formal sessions.

Hundreds of Hadoop Fans Flock to Yahoo! for the Hadoop User Group

The event started with Arun Murthy from Yahoo! describing the best practices for developing MapReduce applications. Arun introduced the concept of a Grid Pattern which, similar to Design Pattern, represents a general reusable solution for applications running on the Grid. Finally, Arun talked about the anti-patterns of applications running on the Apache Hadoop clusters.

Part 1:

Part 2:

Next, Stefan Groschupf, the co-founder and CTO of Datameer, discussed the challenges in social media analytics and how to overcome these using big data analytics built on Hadoop in his “Social Media: What’s Really the Buzz?” talk. The demo was very helpful in visualizing the true thought leads and influencers in social media conversations. These leaders and influences are becoming increasingly important, so that companies can better understand who is having an impact on their customers' buying decisions. This talk gave a very good perspective of how the power of Hadoop can be used to crunch large amounts of data and then visually rendered.

Part 1:

Part 2:

Part 3:

Finally, Matei Zaharia from UC Berkeley talked about Mesos: A Flexible Cluster Resource Manager. The talk highlighted Mesos features and how organizations can consolidate multiple application workloads into a single cluster. The demo showed off the benefits of Mesos and highlighted its ability to run multiple isolated instances of Hadoop on the same cluster. The fault tollerance of Mesos was successfully demonstrated too. Subsequent to the mail session, Matei and team talked about Spark, MapReduce-like framework that adds support for iterative jobs. Spark functional programming model similar to MapReduce was demonstrated capable of caching data between iterations making it very efficient for interactive analysis of big datasets. Spark in addition was demonstrated on the same cluster running alongside Hadoop.

Part 1:

Part 2:

Part 3:

We at Yahoo! embrace Hadoop, and are looking for exciting technologies and experiences you want to share. Please contact me via the Hadoop Bay Area User Group Meetup page.

Susheel Kaushik
Director, Product Management
Cloud Computing at Yahoo!

Apache Hadoop: Best Practices and Anti-Patterns

Wed, 18 Aug 2010 17:00:00 -0800

Apache Hadoop is a software framework to build large-scale, shared storage and computing infrastructures. Hadoop clusters are used for a variety of research and development projects, and for a growing number of production processes at Yahoo!, EBay, Facebook, LinkedIn, Twitter, and other companies in the industry. It is a key component in several business critical endeavors representing a very significant investment and technology component. Thus, appropriate usage of the clusters and Hadoop is critical in ensuring that we reap the best possible return on this investment.

This blog post represents compendium of best practices for applications running on Apache Hadoop. In fact, we introduce the notion of aGrid Pattern which, similar to a Design Pattern, represents a general reusable solution for applications running on the Grid.

This blog post enumerates characteristics of well behaved applications and provides guidance on appropriate uses of various features and capabilities of the Hadoop framework. It is largely prescriptive in its nature; a useful way to look at this document is to understand that applications that follow, in spirit, the best practices prescribed here are very likely to be efficient, well-behaved in the multi-tenant environment of the Apache Hadoop clusters, and unlikely to fall afoul of most policies and limits.

This blog post also attempts to highlight some of the anti-patterns for applications running on the Apache Hadoop clusters.

Overview

Applications processing data on Hadoop are written using the Map-Reduce paradigm.

A Map-Reduce job usually splits the input data-set into independent chunks, which are processed by the map tasks in a completely parallel manner. The framework sorts the outputs of the maps, which are then input to the reduce tasks. Typically both the input and the output of the job are stored in a file-system. The framework takes care of scheduling tasks, monitoring them and re-executes the failed tasks.

Map-Reduce applications specify the input/output locations and supply map and reduce functions via implementations of appropriate Hadoop interfaces, such as Mapper and Reducer. These, and other job parameters, comprise the job configuration. The Hadoop job client then submits the job (jar/executable, etc.) and configuration to the JobTracker, which then assumes the responsibility of distributing the software/configuration to the slaves, scheduling tasks and monitoring them, providing status and diagnostic information to the job-client.

The Map/Reduce framework operates exclusively on <key, value> pairs — that is, the framework views the input to the job as a set of <key, value> pairs and produces a set of <key, value> pairs as the output of the job, conceivably of different types.

Here is the typical data-flow in a Map-Reduce application:

The vast majority of Map-Reduce applications executed on the Grid do not directly implement the low-level Map-Reduce interfaces; rather they are implemented in a higher-level language, such as Pig.

Oozie is the preferred workflow management and scheduling solution on the Grid. Oozie supports multiple interfaces for applications (Hadoop Map-Reduce, Pig, Hadoop Streaming, Hadoop Pipes, etc.) and supports scheduling of applications based on either time or data-availability.

Grid Patterns

This section covers the best practices for Map-Reduce applications running on the Grid.

Input

Hadoop Map-Reduce is optimized to process large amounts of data. The maps typically process data in an embarrassingly parallel manner, typically at least 1 HDFS block of data, usually 128MB.

By default, the framework processes at most 1 HDFS file per-map. This means that if an application needs to processes a very large number of input files, it is better to process multiple files per-map via a special input-format such as MultiFileInputFormat. This is true even for applications processing a small number of tiny input files, processing multiple files per map is significantly more efficient.
If the application needs to process a very large amount of data, even if they are present in large-sized files, it is more efficient to process more than 128MB of data per-map (see section on Maps).

Grid Pattern: Coalesce processing of multiple small input files into smaller number of maps and use larger HDFS block-sizes for processing very large data-sets.

Maps

The number of maps is usually driven by the total size of the inputs, that is, the total number of blocks of the input files. Thus, if you expect 10TB of input data and have a block-size of 128MB, you'll end up with 82,000 maps.

Task setup takes awhile, so it is best if the maps take at least a minute to execute for large jobs.

As explained in the section above on input of applications, it is more efficient to process multiple-files per map for jobs with very large number of small input files.

Even if an application is processing large input files, such that each map is processing a whole HDFS block of data, it is more efficient to process large chunks of data per-map. For example, one way to process more data per map is to have the application process input data with larger HDFS block size, e.g., 512M or even higher, if appropriate.

As an extreme example the Map-Reduce development team used ~66,000 maps to accomplish the PetaSort, that is, 66,000 maps to process 1PB of data (12.5G per map).

The bottom-line is that having too many maps or lots of maps with very short run-time is anti-productive.

Grid Pattern: Unless the application's maps are heavily CPU bound, there is almost no reason to ever require more than 60,000-70,000 maps for a single application.

Also, when processing larger blocks per-map, it is important ensure they have sufficient memory for the sort-buffer to speed up the map-side sort (please see the documentation for io.sort.mb and io.sort.record.percent). The performance of the application can improve dramatically if it can be arranged such that the majority of the map-output can be held in the map's sort-buffer, this will entail larger heap-sizes for the map JVM. It is important to remember that the in-memory footprint of deserialized input might significantly vary from the on-disk footprint; for example, certain class of Pig applications result in 3x-4x blow up of on-disk data in-memory. In such cases, applications might need significantly large heap-sizes for the JVM to ensure the map-input-records and map-output-records can be kept in memory.

Grid Pattern: Ensure maps are sized so that all of map-outputs can be sorted in one pass by keeping all of them in the sort-buffer.

Having the right number of maps has the following advantages for applications:

It reduces the scheduling overhead; having fewer maps means task-scheduling is easier and availability of free-slots in the cluster is higher.
It means the map-side is more efficient; provided there is sufficient memory to accommodate the map-outputs in the sort-buffer in the map.
It reduces the number of seeks required to shuffle the map-outputs from the maps to the reduces — remember that each map produces output for each reduce, thus the number of seeks is m * r where m is #maps and r is #reduces.
Each shuffled segment is larger, resulting in reducing the overhead of connection-establishment when compared to the 'real' work done, that is, moving bytes across the network.
It means that the reduce-side merge of the sorted map-outputs is more efficient, since the branch-factor for the merge is lesser, that is, fewer merges are needed since there are fewer sorted segments of map-outputs to merge.

The caveat to the above guidelines is that processing too much data per-map is bad for failure recovery, a single failed map might hurt the latency of the application.

Grid Pattern: Applications should use fewer maps to process data in parallel, as few as possible without having really bad failure recovery cases.

Combiner

Applications, which use Combiners appropriately, reap benefits of the map-side aggregation effected by them. The primary benefit of the Combiner is that, when used appropriately, it significantly cuts down the amount of data shuffled from the maps to the reduces.

Shuffle

Applications that use Combiners appropriately reap benefits of the map-side aggregation effected by them. The primary benefit of the Combiner is that, when used appropriately, it significantly cuts down the amount of data shuffled from the maps to the reduces.

It is important to remember that Combiner has a performance penalty since it entails an extra serialization/de-serialization of map-output records. Applications that cannot aggregate the map-output bytes by 20-30% should not be using combiners. Applications can use the combiner input/output records counters to measure the efficacy of the Combiner.

Grid Pattern: Combiners help the shuffle phase of the applications by reducing network traffic. However, it is important to ensure that the Combiner does provide sufficient aggregation.

Reduces

The efficiency of reduces is driven by a large extent by the performance of the shuffle.

The number of reduces configured for the application (r) is, obviously, a crucial factor.

Having too many or too few reduces is anti-productive:

Too few reduces cause undue load on the node on which the reduce is scheduled — in extreme cases we have seen reduces processing over 100GB per-reduce. This also leads to very bad failure-recovery scenarios since a single failed reduce has a significant, adverse, impact on the latency of the job.
Too many reduces adversely affects the shuffle crossbar. Also, in extreme cases it results in too many small files created as the output of the job — this hurts both the NameNode and performance of subsequent Map-Reduce applications who need to process lots of small files.

Grid Pattern: Applications should ensure that each reduce should process at least 1-2 GB of data, and at most 5-10GB of data, in most scenarios.

Output

A key factor to remember is that the number of output artifacts of an application is linear w.r.t the number of configured reduces. As discussed in the section on reduces, picking the right number of reduces is very important.

Some other factors to consider:

Consider compressing the application's output with an appropriate compressor (compression speed v/s efficiency) to improve HDFS write-performance.
Do not write out more than one output file per-reduce, using side-files is usually avoidable. Typically applications write small side-files to capture statistics and the like; counters might be more appropriate if the number of statistics collected is small.
Use an appropriate file-format for the output of the reduces. Writing out large amounts of compressed textual data with a codec such as zlib/gzip/lzo is counter-productive for downstream consumers. This is because zlib/gzip/lzo files cannot be split and processed and the Map-Reduce framework is forced to process the entire file in a single map, in the downstream consumer applications. This results in a bad load imbalance and failure recover scenarios on the maps. Using file-formats such as SequenceFile or TFile alleviates these problems since they are both compressed and splittable.
Consider using a larger output block size (dfs.block.size) when the individual output files are large (multiple GBs).

Grid Pattern: Application outputs to be few large files, with each file spanning multiple HDFS blocks and appropriately compressed.

Distributed Cache

DistributedCache distributes application-specific, large, read-only files efficiently. DistributedCache is a facility provided by the Map/Reduce framework to cache files (text, archives, jars and so on) needed by applications. The framework will copy the necessary files to the slave node before any tasks for the job are executed on that node. Its efficiency stems from the fact that the files are only copied once per job and the ability to cache archives which are un-archived on the slaves. It can also be used as a rudimentary software distribution mechanism for use in the map and/or reduce tasks. It can be used to distribute both jars and native libraries and they can be put on the classpath or native library path for the map/reduce tasks.

The DistributedCache is designed to distribute a small number of medium-sized artifacts, ranging from a few MBs to few tens of MBs. One drawback of the current implementation of the DistributedCache is that there is no way to specify map or reduce specific artifacts.

In rare cases, it might be more appropriate for the tasks themselves to do the HDFS i/o to copy the artifacts than rely on the DistributedCache, for example, if an application has a small number of reduces and need very large artifacts (e.g. greater than 512M) in the distributed-cache.

Grid Pattern: Applications should ensure that artifacts in the distributed-cache should not require more i/o than the actual input to the application tasks.

Counters

Counters represent global counters, defined either by the Map/Reduce framework or applications. Applications can define arbitrary Counters and update them in the map and/or reduce methods. These counters are then globally aggregated by the framework.

Counters are appropriate for tracking few, important, global bits of information. They are definitely not meant to aggregate very fine-grained statistics of applications.

Counters are very expensive since the JobTracker has to maintain every counter of every map/reduce task for the entire duration of the application.

Grid Pattern: Applications should not use more than 10, 15 or 25 custom counters.

Compression

Hadoop Map-Reduce provides facilities for the application-writer to specify compression for both intermediate map-outputs and the output of the application, that is, output of the reduces.

Intermediate Output Compression: As explained in the section on shuffle, compression of the intermediate map-outputs with an appropriate compression codec yields better performance by saving on network traffic between the maps and the reduces. Lzo is a reasonably optimal choice for compressing map-outputs since it provides reasonable compression at very high CPU efficiencies.
Application Output Compression: As explained in the section on application output, compression of the outputs with an appropriate compression codec and file-format yields better latency for application. Zlib/Gzip might be an appropriate choice in a majority of cases since it provides high compression at reasonable speeds; bzip2 is usually too slow to be used.

Total Order Outputs

Sampling

Occasionally, applications need to produce totally ordered output, that is, fully-sorted. In such cases, a common anti-pattern is for applications is to use a single-reducer, forcing a single, global aggregation. Clearly, it is very inefficient - this not only puts a significant amount of load on the single node on which the reduce task is executing, but also has very bad failure recovery.

A much better approach is to sample the input and use that to drive a sampling partitioner rather than the default hash partitioner. Thus, one can derive benefits of better load balancing and failure recovery.

Joining Fully Sorted Data-Sets

Another design pattern on the Grid concerns the join of two fully-sorted data-sets whose cardinality is not an exact multiple of the other; for example, one data-set has 512 buckets while the other has 200 buckets.

In such cases, ensuring the input data-sets have a total-order (as described in the previous section) means that the application can use the cardinality of either of the data-sets i.e. 512 or 200 buckets in the above example. Pig handles these joins in the efficient manner described here.

HDFS Operations & JobTracker Operations

The NameNode is a precious resource, applications need to be careful about performing HDFS operations in the Grid. In particular, applications are discouraged from doing non-I/O operations, that is, metadata operations such as stat'ing large directories, recursive stats, and more, from the map/reduce tasks at runtime.

Similarly, applications should not contact the JobTracker for cluster statistics, etc., from the backend.

Grid Pattern: Applications should not perform any metadata operations on the file-system from the backend, they should be confined to the job-client during job-submission. Furthermore, applications should be careful not to contact the JobTracker from the backend.

User Logs

The user task-logs, that is, stdout and stderr, of map/reduce tasks are stored on the local-disk of the compute node on which the task is executed.

Since the nodes are part of the shared infrastructure the Map-Reduce framework implements limits on the amount of task-logs stored on the node.

Web-UI

The Hadoop Map-Reduce framework provides a rudimentary web-ui to track the running jobs, their progress, history of completed jobs, and so on, via the JobTracker.

It is important to remember that the web-ui is meant to be used for humans and not for automated processes.

Implementing automated processes to screen-scrape the web-ui is strictly prohibited. Some parts of the web-ui, such as browsing of job-history, are very resource-intensive on the JobTracker and could lead to severe performance problems when they are screen-scraped.

If there is a need for automated statistics gathering it is better to consult the Grid Solutions, Grid SE, or the Map-Reduce development teams.

Workflows

Oozie is the preferred workflow-management and scheduling system for the Grid. Oozie manages workflows and provides scheduling either based on time or availability of data. Increasingly, latency sensitive production job pipelines are being scheduled and managed through Oozie.

A key factor to keep in mind when designing Oozie workflows is that Hadoop is better suited for batch processing of very large amounts of data. As such, it is advisable for workflows to be comprise of fewer number of medium-to-large sized Map-Reduce jobs, in terms of processing, rather than large number of small Map-Reduce jobs. As an extreme case we have seen single workflows consisting of hundreds and thousands of jobs. This is an obvious anti-pattern. The Hadoop framework, currenty, is not really suited for pipelines of this nature. It would be better to collapse these hundreds/thousands of Map-Reduce jobs into fewer jobs crunching more data — this will help both performance and latency of the workflows.

Grid Pattern: A single Map-Reduce job in a workflow should process at least a few tens of GB of data.

Anti-Patterns

This section attempts to cover some of the common anti-patterns of applications running on the Grid. These are, usually, not in keeping with the spirit of a large-scale, distributed, batch, data processing system.

This is meant to be a warning to the application developers since the Grid software stack is being hardened, particularly in the upcoming 20.Fred release, and the Grid stack will be less forgiving of transgressions to the point of rejecting applications which exhibit some of the anti-patterns described here:

Applications not using a higher-level interface such as Pig unless really necessary.
Processing thousands of small files (sized less than 1 HDFS block, typically 128MB) with one map processing a single small file.
Processing very large data-sets with small HDFS block size, that is, 128MB, resulting in tens of thousands of maps.
Applications with a large number (thousands) of maps with a very small runtime (e.g., 5s).
Straightforward aggregations without the use of the Combiner.
Applications with greater than 60,000-70,000 maps.
Applications processing large data-sets with very few reduces (e.g., 1).

Pig scripts processing large data-sets without using the PARALLEL keyword
Applications using a single reduce for total-order amount the output records

Applications processing data with very large number of reduces, such that each reduce processes less than 1-2GB of data.
Applications writing out multiple, small, output files from each reduce.
Applications using the DistributedCache to distribute a large number of artifacts and/or very large artifacts (hundreds of MBs each).
Applications using tens or hundreds of counters per task.
Applications performing metadata operations (e.g. listStatus) on the file-system from the map/reduce tasks.
Applications doing screen scraping of JobTracker web-ui for status of queues/jobs or worse, job-history of completed jobs.
Workflows comprising hundreds or thousands of small jobs processing small amounts of data.

Arun Murthy, Architect, Hadoop Team at Yahoo!

Pig and Hive at Yahoo!

Wed, 18 Aug 2010 11:45:00 -0800

Yahoo! has begun evaluating Hive for use as part of its Hadoop stack. Since, in many peoples' minds, Hive and Pig are roughly equivalent and Pig Latin is very close to SQL, this has led to some confusion. Why are we interested in using both technologies?

As we have looked at our workloads and analyzed our use cases, we have come to the conclusion that the different use cases require different tools. In this post, I will walk through our thinking on why both of these tools belong in our toolkit, and when each is appropriate.

Data preparation and presentation

Let me begin with a little background on processing and using large data. Data processing often splits into three separate tasks: data collection, data preparation, and data presentation. I will not discuss the data collection phase, because I want to focus on Pig and Hive, neither of which play a role in that phase.

The data preparation phase is often known as ETL (Extract Transform Load) or the data factory. "Factory" is a good analogy because it captures the essence of what is being done in this stage: Just as a physical factory brings in raw materials and outputs products ready for consumers, so a data factory brings in raw data and produces data sets ready for data users to consume. Raw data is loaded in, cleaned up, conformed to the selected data model, joined with other data sources, and so on. Users in this phase are generally engineers, data specialists, or researchers.

The data presentation phase is usually referred to as the data warehouse. A warehouse stores products ready for consumers; they need only come and select the proper products off of the shelves. In this phase, users may be engineers using the data for their systems, analysts, or decisionmakers.

Given the different workloads and different users for each phase, we have found that different tools work best in each phase. Pig (combined with a workflow system such as Oozie) is best suited for the data factory, and Hive for the data warehouse.

Data factory use cases

In our data factory we have observed three distinct workloads: pipelines, iterative processing, and research.

PipelinesThe classical data pipelines bring in a data feed, and clean and transform it. A common example of such a feed is logs from Yahoo!'s web servers. These logs undergo a cleaning step where bots, company internal views, and clicks are removed. We also do transformations such as, for each click, finding the page view that preceded that click.

See my previous post that discusses why we have found Pig to be the best tool for implementing these pipelines. We use it together with our workflow tool, Oozie, to construct pipelines, some of which run tens of thousands of Pig jobs a day.

Iterative processingA closely related use case is iterative processing. In this case, there is usually one very large data set that is maintained. Typical processing on that data set involves bringing in small new pieces of data that will change the state of the large data set.

For example, consider a data set that contained all the news stories currently known to Yahoo! News. You can envision this as a huge graph, where each story is a node. In this graph, there are links between stories that reference the same events. Every few minutes a new set of stories comes in, and the tools need to add these to the graph, find related stories and create links, and remove old stories that these new stories supersede.

What sets this off from the standard pipeline case is the constant inflow of small changes. These require the use of an incremental processing model to process this data in a reasonable amount of time.

For example, if the process has already done a join against the graph of all news stories, and a small set of new stories arrives, re-running the join across the whole set will not be desirable. It will take hours or days.

Instead, joining against the new incremental data and using the results together with the results from the previous full join is the correct approach. This will take only a few minutes. Standard database operations can be implemented in this incremental way in Pig Latin, making Pig a good tool for this use case.

Research

A third use case is research. Yahoo! has many scientists who use our grid tools to comb through the petabytes of data we have. Many of these researchers want to quickly write a script to test a theory or gain deeper insight.

But, in the data factory, data may not be in a nice, standardized state yet. This makes Pig a good fit for this use case as well, since it supports data with partial or unknown schemas, and semi-structured or unstructured data.

Pig integration with streaming also makes it easy for researchers to take a Perl or Python script they have already debugged on a small data set and run it against a huge data set.

Data warehouse use cases

In the data warehouse phase of processing, we see two dominant use cases: business-intelligence analysis and ad-hoc queries.

In the first case, users connect the data to business intelligence (BI) tools — such as MicroStrategy — to generate reports or do further analysis.

In the second case, users run ad-hoc queries issued by data analysts or decisionmakers.

In both cases, the relational model and SQL are the best fit. Indeed, data warehousing has been one of the core use cases for SQL through much of its history. It has the right constructs to support the types of queries and tools that analysts want to use. And it is already in use by both the tools and users in the field.

The Hadoop subproject Hive provides a SQL interface and relational model for Hadoop. The Hive team has begun work to integrate with BI tools via interfaces such as ODBC.

Pig + Hive = Hadoop toolkit

The widespread use of Pig at Yahoo! has enabled the migration of our data factory processing to Hadoop. With the adoption of Hive, we will be able to move much of our data warehousing to Hadoop as well.

Having the data factory and the data warehouse on the same system will lower data-loading time into the warehouse — as soon as the factory is finished, it is available in the warehouse.

It will also enable us to share — across both the factory and the warehouse — metadata, monitoring, and management tools; support and operations teams; and hardware.

So we are excited to add Hive to our toolkit, and look forward to using both these tools together as we lean on Hadoop to do more and more of our data processing.

Alan Gates Architect, Yahoo! Grid Team

Hadoop Bay Area User Group – August 18, Yahoo! HQ

Wed, 11 Aug 2010 17:43:32 -0800

Hi Hadoopers,

I'm excited to invite you to the next Hadoop Bay Area User Group, August 18th, 6 p.m., at the Yahoo! Sunnyvale campus.

The Hadoop community is growing — we had more than 200 attendees at the last meetup.

We invite you to attend whether you are an active submitter, developing Hadoop-based applications, or completely new to the Apache Hadoop world. In addition to interesting presentations, you will enjoy food, beer, and great networking.

The August 18 event comprises two sessions:

Arun Murthy, from the Hadoop team at Yahoo! will introduce compendium of best practices for applications running on Apache Hadoop. In fact, we introduce the notion of a Grid Pattern which, similar to Design Pattern, represents a general reusable solution for applications running on the Grid. He will even cover the anti-patterns of applications running on the Apache Hadoop clusters. Arun will enumerate characteristics of well-behaved applications and provide guidance on appropriate uses of various features and capabilities of the Hadoop framework. It is largely prescriptive in its nature; a useful way to look at the presention is to understand that applications that follow, in spirit, the best practices prescribed here are very likely to be efficient, well-behaved in the multi-tenant environment of the Apache Hadoop clusters and unlikely to fall afoul of most policies and limits.
Stefan Groschupf, the co-founder and CTO of Datameer, will discuss challenges in social media analytics and how to overcome these using big data analytics built on Hadoop, in his “Social Media: What’s Really the Buzz?” talk. Identifying true thought leads and influencers in social media conversations are becoming increasingly important, so that companies can better understand who is having an impact on their customers' buying decisions. Rather than counting mentions in limited subsets of social media data, organizations need a solution that can uncover complex relationships buried in massive volumes of social media data and a way to bring in data from multiple online data sources to determine the quality and effectiveness of user commentary.

For those of you who are not able to attend in person the session's slides and video recording will be posted on this blog after the event. Stay tuned!

Please RSVP at the Hadoop Bay Area Meetup page

Looking forward to see you all soon.

Susheel Kaushik
Director, Product Management
Cloud Computing at Yahoo!

July HUG Recap

Wed, 04 Aug 2010 14:00:00 -0800

Thanks to the over 200 developers who came to Yahoo! recently for our monthly Hadoop User Group meeting. The energy in the packed room was phenomenal, and conversations continued long after the formal sessions.

Hundreds of Hadoop Fans Flock to Yahoo! for the Hadoop User Group

The event started with Nitin Motgi from Yahoo! describing the challenge of content optimization at scale and how Yahoo! is leveraging the power of Hadoop Stack to conquer this challenge. Hadoop, along with Hive and HBase, is the technology that enables mass content personalization for Yahoo! users. Nitin talked about the high-level architecture and modeling challenges. He concluded with a summary on the learnings so far.

Download video

Anil Madan from eBay discussed the adoption of the Hadoop Stack at eBay and their future adoption plans. He also touched upon how eBay is leveraging the Hadoop clusters to enhance search relevance and extend catalog coverage for eBay. He described in detail eBay's data-sourcing patters. Anil concluded with the opportunities for the road ahead.

Download video

Next was Doug Cutting of Cloudera, who introduced Avro, a generic serialization system. He talked about the salient features of the system, contrasted the existing serialization system, and highlighted how Avro can help solve the backward-compatibility issues with Hadoop. He walked us through samples on Avro schemas and Avro Map Reduce APIs. Do view the presentation and video for a quick overview and summary of Avro.

Download video

We at Yahoo! embrace Hadoop, and we understand and relate to the hurdles eBay is facing, and stumble upon the issues that Avro promises to solve. As always, we are looking for exciting technologies and experiences you want to share. Please contact me via the Hadoop Bay Area User Group Meetup page.

Susheel Kaushik
Director, Product Management
Cloud Computing at Yahoo!

Hadoop Archive: File Compaction for HDFS

Tue, 27 Jul 2010 13:00:00 -0800

Thanks to Tsz-Wo Nicholas Sze and Mahadev Konar for this article.

The Problem of Many Small Files

The Hadoop Distributed File System (HDFS) is designed to store and process large (terabytes) data sets. At Yahoo!, for example, a large production cluster may have 14 PB disk spaces and store 60 millions of files.

However, storing a large number of small files in HDFS is inefficient. We call a file small when its size is substantially less than the HDFS block size, which is 128 MB by default. Files and blocks are name objects in HDFS and they occupy namespace. The namespace capacity of the system is naturally limited by the physical memory in the NameNode.

When there are many small files stored in the system, these small files occupy a large portion of the namespace. As a consequence, the disk space is underutilized because of the namespace limitation. In one of our production clusters, there are 57 millions files of sizes less than 128 MB, which means that these files contain only one block. These small files use up 95% of the namespace but only occupy 30% of the disk space.

Hadoop Archive (HAR) is an effective solution to the problem of many small files. HAR packs a number of small files into large files so that the original files can be accessed in parallel transparently (without expanding the files) and efficiently.

HAR increases the scalability of the system by reducing the namespace usage and decreasing the operation load in the NameNode. This improvement is orthogonal to memory optimization in NameNode and distributing namespace management across multiple NameNodes.

Hadoop Archive is also MapReduce-friendly — it allows parallel access to the original files by MapReduce jobs.

What is Hadoop Archive?

Hadoop Archive has three components: a data model that defines the archive format, a FileSystem interface that allows transparent access, and a tool for creating archives with MapReduce jobs.

The Data Model: har format

Figure 1: Archiving small files

The Hadoop Archive's data format is called har, with the following layout:

foo.har/_masterindex //stores hashes and offsets foo.har/_index //stores file statuses foo.har/part-[1..n] //stores actual file data

The file data is stored in multiple part files, which are indexed for keeping the original separation of data intact. Moreover, the part files can be accessed by MapReduce programs in parallel. The index files also record the original directory tree structures and the file statuses. In Figure 1, a directory containing many small files is archived into a directory with large files and indexes.

HarFileSystem – A first-class FileSystem providing transparent access

Most archival systems, such as tar, are tools for archiving and de-archiving. Generally, they do not fit into the actual file system layer and hence are not transparent to the application writer in that the user had to de-archive the archive before use.

Hadoop Archive is integrated in the Hadoop’s FileSystem interface. The HarFileSystem implements the FileSystem interface and provides access via the har:// scheme. This exposes the archived files and directory tree structures transparently to the users. Files in a har can be accessed directly without expanding it. For example, we have the following command to copy a HDFS file to a local directory:

hadoop fs –get hdfs://namenode/foo/file-1 localdir

Suppose an archive bar.har is created from the foo directory. Then, the command to copy the original file becomes

hadoop fs –get har://namenode/bar.har#foo/file-1 localdir

Users only have to change the URI paths. Alternatively, users may choose to create a symbolic link (from hdfs://namenode/foo to har://namenode/bar.har#foo in the example above), then even the URIs do not need to be changed. In either case, HarFileSystem will be invoked automatically for providing access to the files in the har. Because of this transparent layer, har is compatible with the Hadoop APIs, MapReduce, the shell command -ine interface, and higher-level applications like Pig, Zebra, Streaming, Pipes, and DistCp.

The Archiving Tool: A MapReduce program for creating har

To create har efficiently in parallel, we implemented an archiving tool using MapReduce. The tool can be invoked by the command

hadoop archive -archiveName <name> <src>* <dest>

A list of files is generated by traversing the source directories recursively, and then the list is split into map task inputs. Each map task creates a part file (about 2 GB, configurable) from a subset of the source files and outputs the metadata. Finally, a reduce task collects metadata and generates the index files.

Future tasks

Currently, a har cannot be modified after it has been created. Supporting modifiable har is one candidate for future work. Once HFDS supports variable length blocks, har could possibly be created by moving the blocks metadata without copying the actual data. Then, har creation would be nearly instantaneous.

Thanks to Tsz-Wo Nicholas Sze and Mahadev Konar for this article.

Hadoop 0.20.S Virtual Machine Appliance

Tue, 29 Jun 2010 06:00:00 -0800

At Yahoo!, we recently implemented a stronger notion of security for the Hadoop platform, based on Kerberos as underlying authentication system. We also successfully enabled this feature within Yahoo! on our internal data processing clusters. I am sure many Hadoop developers and enterprise users are looking forward to get hands-on experience with this enterprise-class Hadoop Security feature.

In the past, we've aided developers and users get started with Hadoop by hosting a comprehensive Hadoop tutorial on YDN, along with a pre-configured single node Hadoop (0.18.0) Virtual Machine appliance.

This time, we decided to upgrade this Hadoop VM with a pre-configured single node Hadoop 0.20.S cluster, along with required Kerberos system components. We have also included Pig (version 0.7.0), a high level SQL-like data processing language used at Yahoo!.

This blog post describes how to get started with the Hadoop 20.S VM appliance. The basic information about downloading, setting up VM Player, and using the Hadoop VM is same as described in the tutorial module-3 — except the user has to use the following information and links to download the latest VM Player and Hadoop 0.20.S VM Image. You should also review the following information for security-specific commands that need to be performed before running M/R or Pig jobs.

For more details on deploying and configuring Yahoo! Hadoop 0.20.S security distribution, look for continuing announcements and details on Hadoop-YDN.

Installing and Running the Hadoop 0.20.S Virtual Machine:

Virtual Machine and Hadoop environment: See details here.

Install VMware Player: See details here. To download latest VMware Player for Windows/Linux, go to Vmware site

Setting up the Virtual Environment for Hadoop 0.20.S:

Copy the [Hadoop 0.20.S Virtual Machine] into a location on your hard drive.
It is a zipped vmware folder (hadoop-vm-appliance-0-20-S, appriox ~400MB), which includes a few files: a .vmdk file that is a snapshot of the virtual machine's hard drive, and a .vmx file that contains the configuration information to start the virtual machine. After unzipping the vmware folder zip file, to start the virtual machine, double-click on the hadoop-appliance-0.20.S.vmx file. Note: Uncompressed Size of hadoop-vm-appliance-0-20-S folder is ~2GB. Also, based on that data you upload for testing, VM disk is configured to grow up to 20GB).

When you start the virtual machine for the first time, VMware Player will recognize that the virtual machine image is not in the same location it used to be. You should inform VMware Player that you copied this virtual machine image (choose "I copied it"). VMware Player will then generate new session identifiers for this instance of the virtual machine. If you later move the VM image to a different location on your own hard drive, you should tell VMware Player that you have moved the image.

After you select this option and click OK, the virtual machine should begin booting normally. You will see it perform the standard boot procedure for a Linux system. It will bind itself to an IP address on an unused network segment, and then display a prompt allowing a user to log in.

Note: The IP address displayed on the login screen can be used to connect to VM instance over SSH. The Login screen also displays information about starting/stopping Hadoop daemons, users/passwords, and how to shutdown the VM.

Note: It is much more convenient to access the VM via SSH. See details here.

Virtual Machine User Accounts:

The virtual machine comes pre-configured with two user accounts: "root" and "hadoop-user". The hadoop-user account has sudo permissions to perform system-management functions, such as shutting down the virtual machine. The vast majority of your interaction with the virtual machine will be as hadoop-user. To log in as hadoop-user, first click inside the virtual machine's display. The virtual machine will take control of your keyboard and mouse. To escape back into Windows at any time, press CTRL+ALT at the same time. The hadoop-user user's password is hadoop. To log in as root, the password is root.

Hadoop Environment:

Linux : Ubuntu 8.04

Java : JRE 6 Update 7 (See License info @ /usr/jre16/)
Hadoop : 0.20.S (installed @ /usr/local/hadoop, /home/hadoop-user/hadoop is symlink to install directory)
Pig : 0.7.0 (pig jar is installed @ /usr/local/pig, /home/hadoop-user/pig-tutorial/pig.jar is symlink to one in install directory)

Login: hadoop-user, Passwd: hadoop (sudo privileges are granted for hadoop-user). The other usrers are hdfs and mapred (passwd: hadoop).

Hadoop VM starts all the required hadoop and Kerberos daemons during the boot-up process, but in case the user needs to stop/restart,

To start/stop/restart hadoop: login as hadoop-user and run 'sudo /etc/init.d/hadoop [start | stop | restart]' ('sudo /etc/init.d/hadoop' gives the usage)
To format the HDFS & clean all state/logs: login as hadoop-user and run 'sudo reinit-hadoop'
To start/stop/restart Kerberos KDC Server: login as hadoop-user and run 'sudo /etc/init.d/krb5-kdc [start | stop | restart]'
To start/stop/restart Kerberos ADMIN Server: login as hadoop-user and run 'sudo /etc/init.d/krb5-admin-server [start | stop | restart]'

To shut down the Virtual Machine: login as hadoop-user and run command 'sudo poweroff'

Environment for 'hadoop-user' (set in /home/hadoop-user/.profile)
$HADOOP_HOME=/usr/local/hadoop
$HADOOP_CONF_DIR=/usr/local/etc/hadoop-conf
$PATH=/usr/local/hadoop/bin:$PATH

Running M/R Jobs:

Running M/R jobs in Hadoop 0.20.S is pretty much same as running them in non-secure version of Hadoop. Except before running any Hadoop Jobs or HDFS commands, the hadoop-user needs to get the Kerberos authentication token using the command 'kinit'; the password is hadoopYahoo1234.

For example:
hadoop-user@hadoop-desk:~$ cd hadoop
hadoop-user@hadoop-desk:~$ kinit
Password for hadoop-user@LOCALDOMAIN: hadoopYahoo1234
hadoop-user@hadoop-desk:~/hadoop$ bin/hadoop jar hadoop-examples-0.20.104.1.1006042001.jar pi 10 1000000

For automated runs of hadoop jobs, a keytab file is created under the hadoop-user's home directory (/home/hadoop-user/hadoop-user.keytab). This will allow user to execute the "kinit" without having to manually enter the password. So for automated runs of hadoop commands or M/R, Pig jobs through the cron daemon, users can invoke the following command to get the Kerberos ticket. Use command 'klist' to view the Kerberos ticket and its validity.

For example:
hadoop-user@hadoop-desk:~$ cd hadoop
hadoop-user@hadoop-desk:~$ kinit -k -t /home/hadoop-user/hadoop-user.keytab hadoop-user/localhost@LOCALDOMAIN
hadoop-user@hadoop-desk:~/hadoop$ bin/hadoop jar hadoop-examples-0.20.104.1.1006042001.jar pi 10 1000000

Running Pig Tutorial:

The Pig tutorial is installed at "/home/hadoop-user/pig-tutorial". Example commands to run the Pig script are given in "example.run.cmd.sh". The Data needed for Pig scripts are already copied to HDFS. See more details about the Pig Tutorial at Pig@Apache

hadoop-user@hadoop-desk:~$ cd pig-tutorial
hadoop-user@hadoop-desk:~$ sh example.run.cmd.sh

Shutting down the VM:

When you are done with the virtual machine, you can turn it off by logging in as the hadoop-user and running the command 'sudo poweroff'. The virtual machine will shut itself down in an orderly fashion and the window it runs in will disappear.

Last but not least, I would like to thank Devaraj Das and Jianyong Dai from the Yahoo! Hadoop & Pig Develoment team for their help in setting up and configuring Hadoop 0.20.S and Pig respectively.

Notice: Yahoo! does not offer any support for the Hadoop Virtual Machine.
The software include cryptographic software that is subject to U.S. export control laws and applicable export and import laws of other countries. BEFORE using any software made available from this site, it is your responsibility to understand and comply with these laws. This software is being exported in accordance with the Export Administration Regulations. As of June 2009, you are prohibited from exporting and re-exporting this software to Cuba, Iran, North Korea, Sudan, Syria and any other countries specified by regulatory update to the U.S. export control laws and regulations. Diversion contrary to U.S. law is prohibited.

Suhas Gogate
Technical Yahoo!, Cloud Solutions Team, Yahoo!

Managing Big Data: Architectural Approaches for making batch data available online

Thu, 24 Jun 2010 10:05:16 -0800

This is the beginning of an ongoing series of blog posts on “Managing Big Data”. This series will focus on techniques that Yahoo uses to process large volumes of data, ranging from initial collection of data to the end usage of that data.

Introduction

Over the last several years there are two important trends that require additional thought when putting together an architecture for a hosted service. At Yahoo!, the ability to analyze and process enormous amounts of data is increasingly important. It’s a foundational layer for improving our consumer experiences and for sharing audience insights with advertisers.

From a technology perspective, the two trends I'd like to focus on are:

1. Batch processing -- the increasing awareness of batch processing and the recent uptick in use of the map/reduce paradigm for that purpose.

2. NoSQL stores – The rise of so called "NoSQL" stores and their use to serve up data to online users (typically inside of the user's request/response cycle).

Both of these trends represent significant advances in the way that hosted systems are developed. But in order to derive the most value for an entire system, developers must think about how these two areas will work together in some holistic manner.

Let's look at a specific scenario to make this more concrete:

Making batch data available to the online system

Let's assume for the moment that you're building a new e-commerce site. And let's assume that one of the significant features of this e-commerce site is to provide user recommendations for which items a user may be interested in purchasing. This overall feature will decompose into batch component (to determine the recommendations to give to a user) and an online component (that will present the recommendations to end user).

For the batch system we may choose to use a map/reduce framework like Hadoop . The batch recommendation component will rely on various types of raw event data. Some examples include: the products the end user has viewed, the products they have purchased and types of searches the user has performed, etc. We will create a model using this data; perhaps based on a simple behavior model (e.g. if the user looks at a significant amount of sports equipment than recommend products in the sports equipment category) or a collaborative filtering model (e.g. if other users purchased the same products as this end user, recommend other products that they purchased).

Once we have decided on the data inputs and the model for making recommendations, we'll like produce the output of the batch processing as a set of recommendations for each user on a Hadoop cluster.

There are two approaches we can then make for making the batch data available online in a NoSQL store.

1.Full updates

In this approach as part of the batch processing, we will recreate the set of recommendations for all users. It also may be possible to create a native version of the online store format in the batch system. This is generally only possible if there is no scenario where the data (user recommendations for products in this case) is updated in the online store. There must also exist a library to write the into the online store's file format (not all NoSQL stores have such as library).

This approach is attractive for a couple of reasons:

There should not be any consistency issues between your offline and online representation of the data as you are always creating an entirely new copy of the online data at some interval (for instance once a day).
This approach should have the least impact of your online store performance if you can perform a "swap" with new data set. How this would work is that the newly created set of recommendations are pushed in the native NoSQL store format to the online stores physical boxes. Once there each of the NoSQL stores are "upgraded" to start using the new copy of the recommendations and stop using the old version.

2. Incremental/Delta updates

Taking an incremental update approach between your offline and online store entails creating a new set incremental data changes in your batch system. For instance, creating a new set of user recommendations for newly joined users or users that have had some recent activity (which would change the recommendations). This incremental data must then be pushed to the online store.

This approach is attractive for a couple of reasons:

Latency. By processing just the incremental updates of your recommendations is possible to update the online stores on a frequent basis. For instance, it may be possible to run a batch job on Hadoop every 30 minutes that produces a new set of product recommendations. These product recommendations can then be pushed to the online store at a 30 minute interval. The full update approach to moving the offline data online will more likely have an update frequency of several hours or even once a day (due to the large amount of data that needs to be processed and transferred to the online stores).
Size of updates. Depending on the size of the data, it's also possible that incremental updates maybe the only viable solution. For instance if the number of users we would like to create recommendations for is in the 10s of millions and the number of recommendations for each user is large, the data set may be too large to recalculate and push the entire set.

One weakness of the incremental update approach is that it can have a performance impact on your online store. Therefore, when you apply updates to the online store you may need to consider some form of throttling of the updates as they are applied to the online store.

Key takeaways

Consider the type of data you have and whether pushing full updates or delta updates is more appropriate for your type of data as this fundamentally affects the architecture for making your batch data available online. At Yahoo we use both approaches depending on the specific scenario.
Throttling your updates to your online store is important consideration to maintain your online stores latency and availability.

In a future post I'll also address taking data that's available in an online NoSQL store and making that data available in batch system like Hadoop.

Dirk Reinshagen
Cloud Architect
Cloud Computing at Yahoo!

Hadoop and the fight against shape-shifting spam

Tue, 15 Jun 2010 13:58:40 -0800

At a recent Hadoop User Group meeting, I made a presentation on how we leverage hadoop for spam mitigation in Yahoo! Mail. A number of people followed up requesting additional details of our architecture and engineering strategy.
In this post, I am going to try and capture our antispam engineering story, how it came to be hadoop centric and how well the new architecture has worked. I will also highlight the results we have been able to achieve. Finally, I will provide an update on when we will be releasing these updates to wide production.

At the Hadoop User group presentation, I had delved into the details of two interesting antispam algorithms. The first was "frequent itemset mining", the second was what we called the "connected components" algorithm. Both these algorithms are implemented as part of our tools portfolio. They are used by engineers, product managers and operations analysts to get a compact summary of the major trends in spam. Both these tools were implemented as part of a new engineering strategy we put in place second quarter of last year.

Our new strategy called for pointed improvements in the ability of systems to digest massive amounts of data. The first portion of the strategy, implemented by the end of 2009, targeted our reporting systems, tool chains and existing abuse reputation algorithms. The proposal was to increase the granularity of the data being handled, increase the response time to detect an attack and do to do more early detection of spam attacks. In our analysis, we quickly realized that even the small changes we were proposing to our reporting systems, tools and algorithms required us to scale our existing systems well beyond the limits that they were meant to scale.

Also, we found that engineering, product and customer advocacy teams were all hungry for data and it would be great to support additional requirements around ad hoc joins across data streams and support a general "slice" and "dice" approach to data engineering. Our first revisions to the sender and content classifiers also made it abundantly clear that we needed massive storage and massive compute.

We took the simple approach of putting ALL the data that we would possibly want to query or develop algorithms on, on a hadoop grid and let the grid scale to the storage and compute requirements. To give you some idea of the scale involved here, let me provide some ball park metrics.
By the end of this quarter, we will be loading close to 4TB of antispam data on our hadoop grids every day and we will be querying several days of data for report generation and running automated classifiers and algorithms at a frequency of a few minutes. We have not run into any scalability problems so far. In general we have found that with proper data organization, hadoop is able to scale linearly with data and compute requirements.

I will complete this section by saying that this strategy has had a huge impact on spam complaints. See for you self; I am enclosing the graph of our spam complaints from last year. The big dip corresponds to when we shifted our reports, algorithms and filters to the hadoop grid. Need I say more?

While making changes to how an existing system works is interesting and clearly the first step, the second and more interesting step is the development of brand new, distributed reputation algorithms using hadoop. Once again, our new engineering strategy called for the rewiring of all algorithms to run in parallel and increase the level of feature engineering across heuristic, statistical and machine learning systems. We needed to do this across reputation algorithms for IPs, domains, from addresses(senders), receivers(users) and content. Once again, we realized that much of the complexity was in massive data engineering. We needed to ensure that we used every bit of data that would help us make a spam/notspam decision. We also had to choose an appropriate model that could interpret this vast amount of data without getting overwhelmed.

The term "massive feature engineering" should be familiar to people in the area of machine learning. In more common engineering terms, we needed to associate several pieces of meta data to every entity we needed to classify and we needed to choose algorithms that would parallelize well. We have been hard at work the last 5 months doing this new "hadoop engineering". By the end of this month, we hope to release our first hadoop based, massively feature engineered, distributed sender classification algorithm. Code named zeroB, our initial tests make this a very compelling replacement for our current sender management system. It is 25% more accurate while being faster and cheaper to run and maintain than the current version.

Indeed I have now come to believe that Hadoop has tremendous applicability to the abuse and security domains as a whole. Both these domains have the proverbial problem of finding the needle in the hay stack and hadoop is well equipped for this task. With the amount of spam that large mail systems like yahoo see, it is truly important to employ powerful frameworks like Hadoop to ensure the problem remains tractable.

Yahoo mail was recently voted by the renowned fraunhaufer institute as the best free mail service for spam management. This recognition clearly demonstrates that our new hadoop based strategy is working and working very well. This is just the tip of the success though. In the next 3 months, we are rolling out many of these new systems to our wide install base in the United States and I am eagerly waiting to see the effect this has on spam. It has been fun and exciting building these systems on top of hadoop but it has been even more exciting to see us winning the war on spam.

Join us next week for the Yahoo! keynote at Hadoop Summit 2010 to hear more about Hadoop and Mail Antispam.

Vishwanath Ramarao
Director of Anti-Spam Engineering
Yahoo! Mail

Enabling Hadoop Batch Processing Systems to Consume Streaming Data

Wed, 09 Jun 2010 12:40:47 -0800

At Yahoo!, the ability to analyze and process enormous amounts of data is increasingly important. It’s a foundational layer for improving our consumer experiences and for sharing audience insights with advertisers.

In the last few years, I have been a part of a project to design, build, and run a low-latency, large-scale, distributed event data collection system at Yahoo!. When we started off, the goal seemed relatively unambitious, to collect web-access event data across all of the web-servers across all the data centers and bring it to a central location for processing. This perception soon changed after we realized that this involved around 20000 machines and over 20 data centers across the world amounting to over 40 billion events per day that helped fill-up over 10 TB of disk space. To add to the mix, the data had to be available within 15 minutes with an expected completeness of 99% across trans-oceanic fiber optic cable.

We decided to collect the data in a streaming fashion. This enabled us to feed the data at very low latencies to stream processing applications. However, there was an existing batch processing application that required all the data for the entire day to be available with near 100% completeness.

In order to achieve this, the data was collected in a streaming fashion and put into files that contained events belonging to that particular time period, the default being a minute and hence called minute files. Once the data was collected for the minute, the minute files were closed and the data was made available to the consuming application using a queue. Now, this worked reasonably well when there was one application but had problems when the consuming application wanted to reprocess or perform partial updates. In addition to this, the queue essentially kept the consuming application state making the collection and processing systems tightly coupled. This made is increasingly hard when it came to supporting multiple applications because the state for each batch for each application needed to be kept.

This made it even more interesting when the Hadoop initiative at Yahoo! began. All batch processing application were now running on the grid while the data was still consumed by legacy applications. How would we feed the old and the new systems with the same data without duplication?

What we needed was a loosely coupled or completely decoupled method of communicating the files to be processed to downstream batch processing applications. The solution we came up with was a simple but elegant one called a List of Files (LoF) repository.

The LoF repository contains an entry for each minute file collected and its associated attributes such as the start time, the end time, the size in bytes, the number of events, the collection pipeline instance name, and other relevant data. An API to access this data called the LoF API was provided to be able to query the repository for a set of files that satisfied certain attribute constraints. For example, a query might request “all the files that belong to the period 12:00 to 12:05 collected from the sports web servers”. The repository did not need to keep state of which files this application had processed or maintain any queue. This allowed the application to maintain its state and multiple applications were as simple as the single application case. To simplify the usage the API was made available in the form of a RESTful web-service.

Different applications had different completeness requirements from the data collected. For example, a low-latency behavioral targeting application would typically be happy with 95% of the data within 1 minute of the data, while a revenue realization or tracking application would need 100% of the data within 15 minutes. In order to support this, the API returned a completeness metric along with the list of files returned to indicate the percentage of data the list represented. The application could use this information to commence processing based on its own completeness requirements.

Given the distributed nature of the web-servers, data was often delayed or unavailable due to network outages or temporary host unavailability. This meant that applications requiring higher levels of completeness were routinely delayed beyond their SLAs. To solve this we provided a simple timestamp based cursor facility to enable incremental processing. The cursor was essentially returned with the list to indicate the timestamp at which the list was generated. The subsequent query would provide the previously returned cursor along with the subsequent query to indicate the time of the last fetch and the query would return all the files later that that timestamp.

This is what the web-service request looks like:

The response to this is of the form:
collector1.yahoo.com 1234388520 1234 /col1/1200.gz
collector2.yahoo.com 1234388520 3232 /col2/1200.gz

A subsequent request to get incremental data would use the cursor timestamp returned to fetch additional files as follows:

which would get a response similar to:
collector1.yahoo.com 1234388580 3232 /col1/1201.gz

I would like to conclude by saying that the LoF API has enabled the same data to be made available to different application with varying completeness and latency requirements in a simple and elegant manner. Moreover, it has enabled the collection system, which uses a stream-based paradigm, to easily feed multiple largely batch-oriented systems in a relatively seamless manner. Keeping the design simple enabled us to solve a reasonably complex problem.

Akon Dey
Architect, Event Data Collection System at Yahoo!

Hadoop Summit 2010 - Agenda is available!

Thu, 27 May 2010 12:49:17 -0800

I’m happy to share the agenda for the upcoming Hadoop Summit – June 29th, Hyatt, Santa Clara.

We received over 70 great submissions for talks. It was a very impressive combination of development tool overviews, application case studies and innovative research.

We had the difficult task of selecting just a handful of presentations from this overwhelming collection of great quality abstracts and speakers. The variety of topics across numerous industry verticals, served as a clear evidence of how far this technology has evolved over the past year. Hadoop is really going mainstream!

Our goal was to create a diverse agenda that covers topics for experienced Hadoop users as well as people who recently began to explore this technology. We wanted to focus on the Hadoop eco-system of tools and solutions as well as “real life” users experience.

I want to thank all the people who submitted talks and encourage speakers that were not selected to submit their great presentations to our monthly Hadoop User Groups.

Detailed agenda and abstracts available at http://www.hadoopsummit.org/agenda.html

At Yahoo!, we are embracing Hadoop at the very core of our business. As you will hear at the summit, we are continuing to invest heavily in both the technology and the community to make it even better. We love being at the center of the discussion and debates around Hadoop and learning from other’s experiences.

We are planning to go bigger next year, with a broader event that will allow more opportunities for speakers as well as sponsors.

We hope you all join us at the summit – if you have not registered yet, please REGISTER TODAY!. Space is limited and we don’t want you to miss the opportunity to see the great variety of talks outlined above first-hand.

Special thanks to our Sponsors:

Evening reception sponsor: LightSpeed Venture Partners
Breakfast and breaks sponsor: IBM
Lunch co-sponsors: Amazon Web Services, Cloudera, Datameer, Karmasphere

Dekel Tankel
Director, Product Management
Cloud Computing at Yahoo!

Pig, Cascalog & HBase Among Highlights of May Hadoop Meet-Up

Fri, 21 May 2010 18:20:00 -0800

Hi Hadoopers

Thanks to close to 300 developers who came this week to Yahoo! for our monthly Hadoop User Group meeting. The energy in the packed room was phenomenal and conversations continued long after the formal sessions.

Hundreds of Hadoop Fans Flock to Yahoo! for the May Hadoop User Group" src="http://developer.yahoo.com/blogs/hadoop/pic1.jpg" width="425" height="355" />
Hundreds of Hadoop Fans Flock to Yahoo! for the May Hadoop User Group

A few lucky winners received free tickets to the upcoming Hadoop Summit 2010 (June 29th, at the Hyatt Regency, Santa Clara). Congratulations to those winners – everyone else please register here

The event started with Alan Gates from Yahoo! who described the new features and work done in Pig 0.6 and 0.7 including the Hadoop’s compatibility plan, described in more details in this post.

Nathan Marz from BackType presented a cool demo of how easy it is to query existing data stores using Cascalog, a query language for Hadoop. Nathan described how queries can be written as regular Clojure code and combined with Cascading. Be sure to watch the demo as part of the video below.

Next was Dmitriy Ryaboy, an engineer at Twitter and a Pig committer. Dmitriy walked us through the extensive use of Hadoop eco-system at Twitter. He explained what are the challenges they face in processing 55 million tweets a day and why they chose to use Hadoop, Pig and HBase. Dmitriy introduced the Elephant Bird libraries and shared interesting tips for dealing with Big Data.

We concluded with Tom White from Cloudera who walked us through the release plans for Apache Hadoop 0.21 including the Source Compatibility project described in the Yahoo! hadoop blog

We at Yahoo! are embracing Hadoop – we share the challenges presented by Twitter for processing massive data sets and continue to invest heavily in the technology and the community. We love to hear about the growing ecosystem and solutions like Cascalog.

Please join us at the Hadoop Summit to continue the conversation.

As always, we are looking for exciting technologies and experiences you want to share. Please contact me via the Hadoop Bay Area User Group Meetup page.

Note that we will not have a meetup in June due to the Hadoop Summit . See you all on July 21st, 2010. Registration is available here, agenda will be published soon

Dekel Tankel
Director, Product Management
Cloud Computing at Yahoo!

Towards Enterprise-Class Compatibility for Apache Hadoop

Wed, 19 May 2010 11:13:24 -0800

At Yahoo!, the first users of Apache Hadoop were researchers developing new algorithms or manually shifting through huge data sets. These users threw away most of their code after a few weeks or months, and the little code they carried forward was not subject to rigorous quality procedures. Thus, these early users cared more about new features and scalability improvements in Hadoop than they did about backward compatibility.

This early focus on bigger-and-better helped Hadoop become the powerful platform it is today. However, over the years, both inside and outside of Yahoo!, Hadoop is increasingly being used to run large, long-lived, enterprise-class applications. Porting these applications to non-compatible upgrades of Hadoop is an arduous, expensive task that distracts teams from finding new and better ways of using Hadoop to bring value to their companies. Today, Hadoop users are demanding backwards compatibility and interface stability; these features are necessary for the next growth phase of Hadoop, as it gains wider enterprise adoption.

Interface Classification

Over the last year, as part of our plan to provide stronger backward compatibility, we have tagged interfaces in Hadoop to denote their compatibility contract for future releases. An interface can be a Java API, a configuration variable, the parameters or output of a command, metrics variables, and so on. Java APIs are tagged using Java Annotations; other types of interfaces, such as configuration options and output formats, are tagged using informal documentation conventions. The upcoming release 0.21 of Hadoop will be the first to expose this classification.

The classification system is derived from OpenSolaris and from our own internal system at Yahoo. The system distinguishes two important aspects of an interface from the perspective of backward compatibility, the audience of the interface, and the stability of the interface:

The audience (or scope or visibility) of an interface denotes the potential customers of the interface. In addition to the more obvious public and private designations, the audience taxonomy also includes a limited-private category for hooks exposed to peer frameworks or systems.
The stability of an interface denotes when changes can be made to the interface that break compatibility. Again, a binary choice between stable (guaranteed not to change incompatibly) and unstable (guaranteed to change) is not sufficient. Interfaces may also be marked as evolving, intended for use by early adopters validating their suitability. An evolving interface is marked as public only after it’s been used internally in the system and is close to being stable. These dimensions allow early adopters to gauge the risk in relying on a new interface.

The following chart should help you understand what this tagging system means to you:

If you are an application developer, stick to public-stable interfaces. If you are early adopter, you may use a public-evolving interface, but be aware that the interface may change slightly in the future, forcing a change to your application.
If you are a framework developer on Hadoop, you can of course safely use any of the public interfaces, but can also use limited-private interfaces targeted to your framework. For example the Hadoop RPC layer provides limited-private interfaces for HDFS and MapReduce.
Stay away from the private interfaces unless you are an implementer of the actual sub-component. Pay attention to any private interfaces that are stable. While the vast majority of private interfaces are marked as unstable, a few are marked as stable to warn that breaking their compatibility should be done after only serious consideration (and community discussion). For example the internal HDFS and MapReduce protocols are marked stable to support rolling-upgrades (a feature we would like to add to Hadoop in the near future).

More details are provided in the Apache Hadoop API classification document.

The Larger Plan for Compatibility

This classification system for interfaces is part of a larger, multi-step plan for backwards compatibility, which has been in work for the last two years (HADOOP-5071).

The first step was to clean up the package structure of the Hadoop source base (HADOOP-2884). While the high level structure was fine, finer-grained packages better reflect the abstractions of the underlying system architecture. For example, the HDFS package structure did not distinguish the internal abstraction of block storage and namespace, and there were several cases of layer violations within and across packages. In cleaning-up the package structure, we not only provided a better foundation for future work, but also started the process of identifying and separating public interfaces from the private and limited-private ones. We also changed the documentation to reflect this separation.

At the same time, key Hadoop interfaces were redesigned to allow them to evolve compatibly without limiting innovation in the framework. These key interfaces, which also happened to be where Hadoop has faced its greatest compatibility issues, were the APIs to the HDFS file system (HADOOP-4952) ) and the MapReduce framework (HADOOP-1230).We will discuss the redesign of these interfaces in a future blog.

Finally, we have also started to address the wire compatibility problem, i.e. the compatibility of Hadoop components across RPC boundaries and versions. Hadoop, surprisingly, does not offer wire compatibility. To address this limitation, Yahoo started the Avro project (led by Doug Cutting), which was open-sourced in Apache last year and will be incorporated into Hadoop’s RPC system over time..

Acknowledgments

The source re-factoring was done by Sanjay Radia and Raghu Angadi. The interface classification project was initiated by Sanjay Radia and derived mostly from OpenSolaris’ classification scheme. Jakob Homan designed the use of annotations to tag the interfaces. Suresh Srinivas, Tom White, Arun Murthy, Owen O'Malley, and Sanjay Radia defined the annotations for the various parts of the Hadoop system. Tom White helped drive the labeling effort to its conclusion in 0.21, particularly its integration with Hadoop’s javadoc. Doug Cutting was the lead for the Avro project.

Sanjay Radia
Hadoop Team, Yahoo!

Scalability of the Hadoop Distributed File System

Wed, 05 May 2010 16:00:00 -0800

In his fictional story "The Library of Babel", Jorge Luis Borges describes a vast storage universe composed of all possible manuscripts uniformly formatted as 410-page books. Most are random meaningless sequences of symbols. But the rest excitingly forms a complete and an indestructible knowledge system, which stores any text written in the past or to be written in the future, thus providing solutions to all problems in the world. Just find the right book.

The same characteristic fascinates us in modern storage growth: The aggregation of information directly leads to proportional growth of new knowledge discovered out of it. A skeptic may doubt that further reward in knowledge mining will not justify the effort in information aggregation. What if by building sophisticated storage systems we are chasing the 19th century’s horse manure problem, when at the dawn of the automobile era the scientific world was preoccupied with the growth of the horse population that threatened to bury the streets of London nine feet deep in manure?

The historical judgment will turn one way or another. In the meantime, we want to know how far we can go with existing systems even if only out of pure curiosity. In my USENIX ;login: article “HDFS Scalability: the Limits to Growth,” I studied scalability and performance limitations imposed on a distributed file system by the single-node namespace server architecture. The study is based on experience with largest deployments of the Hadoop Distributed File System (HDFS) currently in production at Yahoo!.

The first part of the study focuses on how the single name-server architecture limits the number of namespace objects (files and blocks) and how this translates to the limitation of the physical storage capacity growth.

The second part explores the limits for linear performance growth of HDFS clusters bound by natural performance restrictions of a single namespace server. As the cluster grows, the linear increase in the demand for processing resources puts a higher workload on the single namespace server. When the workload exceeds a certain threshold, it saturates the server, turning it into a single-node bottleneck for linear scaling of the entire cluster.

In 2006, the Hadoop group at Yahoo! formulated long-term target requirements for HDFS and outlined a list of projects intended to bring the requirements to life. Table 1 summarizes the targets and compares them with the current achievements:

	Target	Deployed
Capacity	10 PB	14 PB
Nodes	10,000	4,000
Clients	100,000	15,000
Files	100,000,000	60,000,000

Table 1: Targets for HDFS vs. actual deployed values as of 2009

The bottom line is that we achieved the target in Petabytes and got close to the target in the number of files. But this is done with a smaller number of nodes and the need to support a workload close to 100,000 clients has not yet materialized. The question now is whether the goals are feasible with the current system architecture.

Namespace Limitations

HDFS is based on an architecture where the namespace is decoupled from the data. The namespace forms the file system metadata, which is maintained by a dedicated server called the name-node. The data itself resides on other servers called data-nodes.

The namespace consists of files and directories. Files are divided into large (128 MB) blocks. To provide data reliability, HDFS uses block replication. Each block by default is replicated to three data-nodes. Once the block is created its replication is maintained by the system automatically. The block copies are called replicas.

The name-node keeps the entire namespace in RAM. This architecture has a natural limiting factor: the memory size; that is, the number of namespace objects (files and blocks) the single namespace server can handle.

Estimates show that the name-node uses less than 200 bytes to store a single metadata object (a file inode or a block). According to statistics on Y! clusters, a file on average consists of 1.5 blocks. Which means that it takes 600 bytes (1 file object + 2 block objects) to store an average file in name-node’s RAM

To store 100 million files (referencing 200 million blocks), a name-node should have at least 60 GB of RAM.

Storage Capacity vs. Namespace Size

With 100 million files, each having an average of 1.5 blocks, we will have 200 million blocks in the file system. If the maximal block size is 128 MB and every block is replicated 3 times, then the total disk space required to store these blocks is about 60 PB.

100 million file namespace needs 60 PB of total storage capacity on the cluster

As a rule of thumb, the correlation between the representation of the metadata in RAM and physical storage space required to store data referenced by this namespace is:

1 GB metadata ≈ 1 PB physical storage

The rule should not be treated the same as, say, the Pythagorean Theorem, because the correlation depends on cluster parameters (the block to file ratio and the block size). But it can be used as a practical estimate for configuring cluster resources.

Cluster Size and Node Configuration

Next we can estimate the number of data-nodes the cluster should have to accommodate the namespace of a certain size. On Yahoo’s clusters, data-nodes are usually equipped with four disk drives of size 0.75 – 1 TB, and configured to use 2.5 – 3.5 TB of that space per node. The remaining space is allocated for MapReduce transient data, system logs, and the OS.

If we assume that an average data-node capacity is 3 TB, then we will need on the order of 20,000 nodes to store 60 PB of data. To keep it consistent with the target requirement of 10,000 nodes, each data-node should be configured with eight 1TB hard drives.

To accommodate data referenced by a 100 million file namespace, a HDFS cluster needs 10,000 nodes equipped with 8TB of total hard drive capacity per node

Note that these estimates are true under the assumption that the block per file ratio of 1.5 and the block size remain the same. If the ratio or the block size increases, a gigabyte of RAM will support more petabytes of physical storage and vice versa. Sadly, based on practical observations, the block to file ratio tends to decrease during the lifetime of a file system, meaning that the object count (and therefore the memory footprint) of a single namespace server grows faster than the physical data storage. That makes the object-count problem — which becomes file-count problem when the average file to block ratio is close to 1 — the real bottleneck for cluster scalability.

The Internal Load

Apart from the hierarchical namespace the name-node’s metadata includes a list of registered data-nodes, and a block to data-node mapping, which determines physical block locations.

A data-node identifies block replicas in its possession to the name-node by sending block reports. The first block report is sent at the startup. It reveals the block locations, which otherwise are not known to the name-node at startup time. Subsequently, block reports are sent periodically every 1 hour by default and serve as a sanity check providing that the name-node has an up-to-date view of block replica distribution on the cluster.

During normal operation, data-nodes periodically send heartbeats to the name-node to indicate that the data-node is alive. The default heartbeat interval is 3 seconds. If the name-node does not receive a heartbeat from a data-node in 10 minutes, it pronounces the data-node dead and schedules its blocks for replication on other nodes.

The block reports and heartbeats form the internal load of the cluster. If the internal load is too high, the cluster becomes dysfunctional, able to process only a few, if any, external client operations such as ls, read, or write. The internal load depends on the number of data-nodes. Assuming that the cluster is built of 10,000 data-nodes having 8 hard drives with 6 TB of effective storage capacity each, we estimate that the name-node will need to handle

3 block reports per second, each reporting 60,000 replicas
10,000 heartbeats per second

Using the standard HDFS benchmark called NNThroughputBenchmark, we measure the actual name-node performance with respect to the two internal operations. Table 2 summarizes the results. Note that the block report throughput is measured in the number of blocks processed by the name-node per second.

	Throughput
Number of blocks processed in block reports per second	639,713
Number of heartbeats per second	300,000

Table 2: Block report and heartbeat throughput

The implication of these results is:

The internal load for block reports and heartbeat processing on a 10,000 node HDFS cluster with the total storage capacity of 60 PB will consume 30% of the total name-node processing capacity.

The internal load is proportional to the number of nodes in the cluster and the average number of blocks on a node. Thus, if a node had only 30,000 blocks — half of the estimated amount — then the name-node will dedicate only 15% of its processing resources to the internal load. Vice versa, if the average number of blocks per node grows, then the internal load will grow proportionally. The latter means that the decrease in block to file ratio (more small files with the same file system size) increases the internal load and therefore negatively affects the performance of the system.

Reasonable Load Expectations

We have learned by now that the name-node can use 70% of its time to process external client requests. Even with a handful of clients one can saturate the name-node performance by letting the clients send requests to the name-node with very high frequency. The name-node most probably would become unresponsive, potentially sending the whole cluster into a tailspin because internal load requests do not have priority over regular client requests.

In practice, the extreme load bursts are uncommon. Regular Hadoop clusters run MapReduce jobs, and jobs perform conventional file reads or writes. To get or put data from or to HDFS, a client first accesses the name-node and receives block locations, and then directly talks to data-nodes to transfer file data. Thus the frequency of name-node requests is bound by the rate of data transfer from data-nodes.

Typically a map task of a MapReduce jobs reads one block of data. We estimate how much time it takes for a client to retrieve a block replica and, based on that, evaluate the expected read load on the name-node — namely, how many “get block location” requests the name-node should expect per second from 100,000 clients. We apply the same technique to evaluate the write load on the cluster.

Performance Limitations

The read and write throughputs have been measured by the DFSIO benchmark and are summarized in Table 3.

	Throughput
Average read throughput	66 MB/s
Average write throughput	40 MB/s

Table 3: HDFS read and write throughput

Another series of throughput results produced by NNThroughputBenchmark (Table 4) measures the number of open (the same as get block location) and create operations processed by the name-node per second:

	Throughput
Get block locations	126,119 ops/s
Create new block	5,600 ops/s

Table 4: Open and create throughput

The read throughput in Table 3 indicates that 100,000 readers are expected to produce 68,750 get-block-location requests to the name-node per second. And Table 4 confirms that the name-node is able to process that many requests even if only 70% of its processing capacity is dedicated to this task.

A 10,000 node HDFS cluster with the internal load at 30% will be able to handle an expected read-only load produced by 100,000 HDFS clients.

The write performance is not as optimistic. According to Table 3, 100,000 writers are expected to provide an average load of 41,667 create block requests per second on the name-node. This is way above 3,920 creates per second — 70% of possible processing capacity of the name-node (see Table 4).

A reasonably expected write-only load produced by 100,000 HDFS clients on a 10,000 node HDFS cluster will exceed the throughput capacity of a single name-node.

We have seen that a 10,000-node HDFS cluster with single name-node is expected to handle a workload of 100,000 readers well. However, even 10,000 writers can produce enough workload to saturate the name-node, making it a bottleneck for linear scaling.

Such a large difference in performance is attributed to get-block-locations (read workload) being a memory-only operation, while creates (write workload) requires journaling, which is bounded by the local hard drive performance.

There are ways to improve the single name-node performance. But any solution intended for single namespace server optimization lacks scalability.

Looking into the future — especially taking into account that the ratio of small files tend to grow — the most promising solutions seem to be based on distributing the namespace server itself both for workload balancing and for reducing the single server memory footprint.

I hope you will benefit from the information provided above. Please refer to the USENIX ;login: article for more details.

Konstantin V. Shvachko
Hadoop Distributed File System (HDFS) Team, Yahoo!

Hundreds of Hadoop Fans Flock to Yahoo! for the April Hadoop User Group Meet-Up

Wed, 21 Apr 2010 12:48:13 -0800

Hi Hadoopers

Thanks to more than 250 developers who came tonight to Yahoo! for our monthly Hadoop User Group meeting. With Facebook's F8 developer conference and the downpour of April showers it was nice to see such turnout.

The event started with Vishwanath Ramarao, Director of anti-spam engineering for Yahoo! Mail. Vish described the intricate cat-and-mouse games played with spammers, and how Yahoo! uses Hadoop to abstract away the complexity of large scale data analysis and provide deep insight into spammer campaigns.

Next was a presentation from John Sichi, lead engineer for Facebook's data infrastructure team. John provided an overview of Facebook's recent integration between Hadoop, HBase and Hive and the motivation for it - "Data, data, and more data".

We concluded with Ken Krugler, the founder of Bixo Labs. Ken described the Public Terabyte Dataset project - a large-scale web crawl that uses SimpleDB, Hadoop, Cascading and Bixo in the Amazon's EMR cloud.

We will publish shortly video recordings of the sessions on this blog. Stay tuned!

We at Yahoo! are embracing Hadoop – as illustrated by the Yahoo! Mail case study. The ability to process massive data sets is core to our business and we are continuing to invest heavily in the technology and the community. We love to hear about the growing ecosystem of solutions and frameworks built around Hadoop.

Please join us at the Hadoop Summit to continue the conversation.

As always, we are looking for exciting technologies and experiences you want to share. Please add presentation requests at the Hadoop Bay Area User Group Meetup page.

See you all on May 19th, 2010. Registration is available here, agenda will be published soon

Dekel Tankel
Director, Product Management
Cloud Computing at Yahoo!