Hadoop Bay Area User Group - Feb 17th at Yahoo!, Sunnyvale

Yahoo! is hosting the monthly Bay Area Hadoop User Group on Wednesday, February 17th 6PM at the Yahoo! Sunnyvale Campus. Whether you are an active submitter, developing using Hadoop related technologies or completely new to Hadoop -- we'd love to see you.

We are hosting Kevin Weil who leads the analytics team at Twitter. Kevin will provide an overview of Hadoop and Pig at Twitter. Kevin will cover LZO Compression and Protocol Buffers and how they are combined with PIG for flexible data storage and fast Map-Reduce jobs.

We are hosting Mathias Stearn from 10Gen who will provide introduction to MongoDB and Hadoop. Mathias will explain how to get started with CRUD and JavaScript, how to create schemas and how to scale. He will also describe the integration between MongoDB and Hadoop.

Registration and additional session information is available on the Bay Area HUG Meetup page

Looking forward to see you there and for those of you who can't attend in person, stay tune! we will publish the slides and video recording after the event.

Dekel Tankel
Director, Product Management
Cloud Computing at Yahoo!

Comparing Pig Latin and SQL for Constructing Data Processing Pipelines

I have been asked by users who are going to construct a data pipeline whether they should use Pig Latin or SQL.

For those of you who are not familiar with Pig, it is a platform for analyzing large data sets. It is built on Hadoop and provides ease of programming, optimization opportunities and extensibility. Pig Latin is the relational data-flow language and is one of the core aspects of Pig.

In this blog I refer to "data pipeline" as the means by which applications that take data from one or more sources, cleanse it, do some initial transformation on it that all the data readers will need, and then store it in a data warehouse . As SQL is known by almost everyone, it is often chosen as the language in which to write these data pipelines.

We are comparing Pig Latin over Hadoop to SQL over a relational database.

SQL's ubiquity is convenient. However, I believe that Pig Latin is a more natural choice for constructing data pipelines, for several reasons:

1. Pig Latin is procedural, where SQL is declarative.
2. Pig Latin allows pipeline developers to decide where to checkpoint data in the pipeline.
3. Pig Latin allows the developer to select specific operator implementations directly rather than relying on the optimizer.
4. Pig Latin supports splits in the pipeline.
5. Pig Latin allows developers to insert their own code almost anywhere in the data pipeline.

I will consider each of these points in turn.

Since Pig Latin is procedural, it fits very naturally in the pipeline paradigm. SQL on the other hand is declarative. Consider, for example, a simple pipeline, where data from sources users and clicks is to be joined and filtered, and then joined to data from a third source geoinfo and aggregated and finally stored into a table ValuableClicksPerDMA. In SQL this could be written as:

insert into ValuableClicksPerDMAselect dma, count(*)
from geoinfo join (
                select name, ipaddr
                from users join clicks on (users.name = clicks.user)
                where value > 0;
            ) using ipaddr
group by dma;

The Pig Latin for this will look like:

Users                = load 'users' as (name, age, ipaddr);Clicks               = load 'clicks' as (user, url, value);
ValuableClicks       = filter Clicks by value > 0;
UserClicks           = join Users by name, ValuableClicks by user;
Geoinfo              = load 'geoinfo' as (ipaddr, dma);
UserGeo              = join UserClicks by ipaddr, Geoinfo by ipaddr;
ByDMA                = group UserGeo by dma;
ValuableClicksPerDMA = foreach ByDMA generate group, COUNT(UserGeo);
store ValuableClicksPerDMA into 'ValuableClicksPerDMA';

Notice how SQL forces the pipeline to be written inside-out, with operations that need to happen first happening in the from clause sub-query. Of course this can be resolved with the use of intermediate or temporary tables. Then the pipeline becomes a disjointed set of SQL queries where ordering is only apparent by looking at a master script (written in some other language) that sews all the SQL together. Also, depending on how the database handles temporary tables, there may be cleanup issues to deal with. In contrast, Pig Latin shows users exactly the data flow, without forcing them to either think inside out or construct a set of temporary tables and manage how those tables are used between different SQL queries.

The pipeline given above is obviously simple and contrived. It consists of only two very simple steps. In practice data pipelines at large organizations are often quite complex, if each Pig Latin script spans ten steps then the number of scripts to manage in source control, code maintenance, and the workflow specification drops by an order of magnitude.

Experienced data pipeline developers will object to the point above about Pig Latin not needing temporary tables. They will note that storing data in between operations has the advantage of check pointing data in the pipeline. That way, when a failure occurs, the whole pipeline does not have to be rerun. This is true. Pig Latin allows users to store data at any point in the pipeline without disrupting the pipeline execution. The advantage that Pig Latin provides is that pipeline developers decide where appropriate checkpoints are in their pipeline rather than being forced to checkpoint wherever the semantics of SQL imposes it. So, if for the above pipeline there was a need to store data after the second join (UserGeo) and before the group by (ByDMA), the script could be changed to:

Users                = load 'users' as (name, age, ipaddr);Clicks               = load 'clicks' as (user, url, value);
ValuableClicks       = filter Clicks by value > 0;
UserClicks           = join Users by name, ValuableClicks by user;
Geoinfo              = load 'geoinfo' as (ipaddr, dma);
UserGeo              = join UserClicks by ipaddr, Geoinfo by ipaddr;
store UserGeo into 'UserGeoIntermediate';
ByDMA                = group UserGeo by dma;
ValuableClicksPerDMA = foreach ByDMA generate group, COUNT(UserGeo);
store ValuableClicksPerDMA into 'ValuableClicksPerDMA';

This would result in no additional Map Reduce jobs. Pig would store the intermediate data after the aggregation and continue with the join as before.

By definition, a declarative language allows the developer to specify what must be done, not how it is done. Thus in SQL users can specify that data from two tables must be joined, but not what join implementation to use. Developers are forced to have faith that the optimizer will make the right choice for them. Some databases work around this by allowing hints to be given to the optimizer, but even then the implementation is not required to follow those hints.

While for many SQL applications the query writer may not have enough knowledge of the data or enough expertise to specify an appropriate join algorithm, this is not usually the case for data pipelines. Data flowing through data pipelines does not tend to vary significantly from run to run, in terms either of volume or key distribution. In addition data pipeline developers are usually sophisticated enough to choose the correct algorithm. For these reasons allowing developers to explicitly choose an implementation, and be guaranteed that their choice will be honored, is quite useful in data pipelines.

Pig Latin allows users to specify an implementation or aspects of an implementation to be used in executing a script in several ways. For joins and grouping operations users can specify an implementation to use, and Pig guarantees that it will use that implementation. Currently Pig supports four different join implementations and two grouping implementations. It also allows users to specify parallelism of operations inside a Pig Latin script, and does not require that every operator in the script have the same parallelization factor. This is important because data sizes often grow and shrink as data flows through the pipeline.

Another common feature of data pipelines is that they are often graphs (DAGs) and not linear pipelines. SQL, however, is oriented around queries that produce a single result. Thus SQL handles trees (such as joins) naturally, but has no built in mechanism for splitting a data processing stream and applying different operators to each sub-stream. A very common use case we have seen in Yahoo! is a desire to read one data set in a pipeline and group it by multiple different grouping keys and store each as separate output. Since disk reads and writes (both scan time and intermediate results) usually dominate processing of large data sets, reducing the number of times data must be written to and read from disk is crucial to good performance.

Take for example a user data set where there is a desire to analyze the data set both in geographic and demographic dimensions. The Pig Latin to do this analysis looks like:

Users         = load 'users' as (name, age, gender, zip);Purchases     = load 'purchases' as (user, purchase_price);
UserPurchases = join Users by name, Purchases by user;
GeoGroup      = group UserPurchases by zip;
GeoPurchase   = foreach GeoGroup generate group, SUM(UserPurchases.purchase_price) as sum;
ValuableGeos  = filter GeoPurchase by sum > 1000000;
store ValuableGeos into 'byzip';
DemoGroup     = group UserPurchases by (age, gender);
DemoPurchases = foreach DemoGroup generate group, SUM(UserPurchases.purchase_price) as sum;
ValuableDemos = filter DemoPurchases by sum > 100000000;
store ValuableDemos into 'byagegender';

This Pig Latin script describes a DAG rather than a pipeline. It starts with two inputs which are brought into one stream (via join) which is then split into two streams. Pig will do this in two Map Reduce jobs (one for the join and one for both group bys and their filters) rather than requiring that the join be either run twice or materialized as an intermediate result as traditional SQL would.

Pig Latin's ability to include user code at any point in the pipeline is useful for pipeline development. This is accomplished through user defined functions (UDFs) and streaming. UDFs allow users to specify how data is loaded, how it is stored, and how it is processed. Streaming allows users to include executables at any point in the data flow.

Allowing developers to specify how data is loaded is useful because in most data pipelines data sources are not database tables. If SQL is used, data must first be imported into the database, and then the cleansing and transformation process can begin. There are many ETL tools on the market to handle this import process for databases. Pig allows developers to write a function in Java to read data directly from the source. This eliminates the need for a second tool which must be purchased, learned, and used and allows the data pipeline to combine the loading and initial cleansing and transformation steps.

Pipelines also often include user defined column transformation functions and user defined aggregations. Pig Latin supports writing both of these types of functions in Java. We plan to extend that to a number of scripting languages in the near future, thus enabling users to easily write UDFs in the language of their choice.

If the user defined code will not fit well into a UDF, streaming allows pipelines to place an executable in the pipeline at any point. This can also be used to include legacy functionality that cannot be modified.

To conclude, I hope you will agree with me that these advantages of an intuitive, procedural programming model, control of where data is check pointed in the pipeline, the ability to completely control how data is processed, support for general DAGs, and the ability to include user code wherever necessary make Pig Latin a better choice for developing data pipelines on Hadoop.

Video from Jan. 20, 2010 Hadoop Bay Area User Group now online

Stomping out Java "concurrency cockroaches" with SureLogic's Flashlight and JSure tools

Concurrency errors are the cockroaches of the software bug ecosystem. They are difficult to detect and tend appear at the most inopportune times. And, when you try to shine a light on them, they go into hiding. When supplying widely-used infrastructure software, like Hadoop, it is doubly important to stomp out them out and keep them from coming back. To date, there has been a shortage of tools focused on helping programmers with this daunting task. A small Pittsburgh company, SureLogic, is trying to address this tool shortage with a suite of Java-based concurrency-focused static and dynamic analysis tools. SureLogic has extensive experience in Java concurrency and in analysis of large and complex Java code bases.

SureLogic JSure is a model-based static analysis tool that helps developers gain confidence in their multi-threaded Java code, regardless of scale or complexity. JSure provides positive assurance (sound analysis, not rule-based) that correct locks are held when shared state is accessed. This tool helps the programmer answer the question "Are my threads accessing shared state in a safe way?" The SureLogic JSure modeling language is open source under the Apache license. Even when the JSure tool is not used, developers have found the annotation-based models very useful for documentation purposes.

SureLogic Flashlight is a dynamic analysis tool that acts as a concurrency-focused runtime profiler that illuminates threading behavior and access to shared state. When developers are in the dark about why their application is experiencing intermittent failures, poor performance, or data corruption, Flashlight provides visibility. SureLogic is working with Carnegie Mellon University to enhance Flashlight to bridge from development into distributed monitoring.

Four SureLogic engineers visited Yahoo! Sunnydale on October 28, 29, and 30 and worked with members of the MapReduce, Zookeeper, and HDFS teams. The SureLogic engineers were Tim Halloran, Aaron Greenhouse, Edwin Chan, and Nathan Boy. On the Yahoo! side, Konstantin (Cos) Boudnik hosted the SureLogic team. Cos works on HDFS. The other HDFS engineers who participated were Konstantin Shvachko, Hairong Kuang, Jakob Homan, and Boris Shkolnik. The Zookeeper engineers who participated were Pat Hunt, Mahadev Konar, and Ben Reed. The MapReduce engineers who participated were Dick King, Chris Douglas, Owen O'Malley, and Hong Tang. Nigel Daley also participated.

Above:SureLogic engineers Edwin Chan, Aaron Greenhouse, and Nathan Boy in front of Building E in Sunnyvale.

During the visit Yahoo! and SureLogic engineers worked side-by-side in several conference rooms. JSure and Flashlight were run on the Hadoop code with the SureLogic team providing expertise and instruction on JSure and Flashlight tools and the Yahoo! engineers providing a deep understanding about the code they work on and the environment it is developed within. Typically using a projector was used to allow everyone in the room to see the tool results.

Flashlight

To work with Flashlight programmers run an instrumented version of their program automatically created by Flashlight. The data collected from the program run can be queried in a very general way. Flashlight currently supports 47 concurrency-focused queries as well as custom queries created by users. These range from informational queries such as "What fields were observed to be shared between threads?" "Where were two or more locks held at the same time?" to focused queries to uncover race conditions and the potential for the program to deadlock such as "What locks could potentially deadlock?" and "What shared fields are not protected by a consistent lock?" The Flashlight tool, which is hosted within the Ecilpse Java IDE, is shown in the screenshot below.

Above:Sample Flashlight tool output on a run of HDFS TestFileAppend2. The query shows non-final non-static fields that were observed to be shared between one or more threads during the run. A lock contention query (highly contested locks can impact program performance) and a potential deadlock query have been run on the data.

The Yahoo! engineers thought that Flashlight was "cool/powerful/useful" but also provided a list of ideas to improve the tool and make it more useful on Hadoop. In particular the ability to focus on "what changed" from one run to another (e.g., deadlocks or race conditions) would help make the tool more useful. SureLogic is working on these improvements to Flashlight.

JSure

To work with JSure programmers place a few Java 5 annotations (or Javadoc if 1.4 code) in their code that are then checked by the tool. These annotations represent programmer design intent about the program that cannot be readily inferred from the program's code. JSure checks that the annotations are consistent with the code. A few sample JSure annotations are shown below.

@Regions({   @Region("HeartbeatState"),
   @Region("DatanodeState")
})
@RegionLocks({
   @RegionLock("HeartbeatLock is heartbeats protects HeartbeatState"),
   @RegionLock("DatanodeLock is datanodeMap protects DatanodeState")
})
public class FSNamesystem implements FSConstants, FSNamesystemMBean,
FSClusterStats { ... }

These annotations define two regions of the HDFS program's state, HeartbeatState and DatanodeState, and two locking models, HeartbeatLock and DatanodeLock. Fields are added to a declared regions using annotations at the field declarations. For example,

@InRegion("HeartbeatState") private int totalLoad = 0;

adds the field totalLoad to the region HeartbeatState.

The @RegionLock annotation defines a lock that is intended to protect a region of the program's state. For example,

@RegionLock("DatanodeLock is datanodeMap protects DatanodeState")

states that a lock on the object referenced by the field datanodeMap protects the program's state defined by the DatanodeState region. JSure checks that the annotations, which define a model of programmer design intent, are consistent with the code. An example of the tool output for the DatanodeLock model is shown in the screenshot below.

Above: Sample JSure tool output for the DatanodeLock model annotated in FSNamesystem (part of HDFS). The green results indicate areas of model-code consistency the red results indicate inconsistency. The goal is to keep the code and the annotated model consistent.

JSure supports 22 kinds of annotations that help a programmer precisely document concurrency design intent about his or her code. These annotations are useful even without the tool and are described at http://www.surelogic.com/promises/apidocs/index.html

The Yahoo! engineers liked the annotations and their ability to formally express concurrency design intent. They had suggestions to improve the tool experience and make it more useful to them. SureLogic is working on these improvements to JSure. In particular the ZooKeeper team found JSure to be extremely impressive. "We found a number of significant issues with just a few hours of work. We really like the iterative approach. We really like the start with nothing approach (We hate tools that spew thousands of problems that are not actionable). We like the idea that JSure can be integrated into our build and run as part of the patch process." They also noted that, "The annotations need to be standardized." SureLogic has open sourced the JSure annotations and is involved with the JSR process.

Since the SureLogic visit to Sunnyvale, SureLogic has released the JSure annotations into the Maven global repository and JIRA requests are pending to introduce the annotations into the Hadoop build process. Cos and Edwin Chan are working on including JSure into the patch system used by Hadoop (similar to FindBugs).

SureLogic is supporting the Hadoop community by providing licenses and support for their tools and has open sourced the JSure annotations. The tools can be downloaded from http://www.surelogic.com/static/eclipse/install.html and Hadoop community members read the post at http://wiki.apache.org/hadoop/HowToUseConcurrencyAnalysisTools about the tools and obtain the license from there. The SureLogic tools have several tutorials that should get you up and running quickly. You can also get help directly from SureLogic if you have any problems. SureLogic has a public Bugzilla at http://www.surelogic.com/bugzilla/index.cgi for tracking user issues (focused on the annotations and their effective use as well as the tools).

A great presentation "Racing Toward Disaster" discussing concurrency traps and pitfalls given by Tim during SureLogic visit to Sunnyvale can be found at http://www.scribd.com/doc/25448092

SureLogic would like to thank Cos for hosting us and Don McGillen for his help in getting us to Sunnyvale. Thanks!

Hadoop Bay Area January 2010 User Group - Recap

Hi Hadoopers

Thanks everyone for joining us last night at the Yahoo!’s Sunnyvale campus. There were close to 150 attendees, a nice way to start the meetings for 2010. I was happy to see familiar and many new faces. It was also great to see the thriving conversations and solution sharing..

For those of you who were unable to attend in person the session's details, slides and video recordings are posted below

Bhupesh Bansal, Senior Engineer at LinkedIn shared the details behind Project-Voldemort (distributed key-value storage system based on the Amazon Dynamo project), challenges, performance, features and more. Bhupesh reviewed the growing use of Hadoop for Batch Computing at Linkedin - data store, workflows, ETL, prototyping and more.
Bhupesh is a member of LinkedIn's search and data platform team and an active commiter for Project-Voldemort

Slides:

Video:

Cloudera speaks VMware vCloud API, too.

Hadoop World: Building Data Intensive Apps with Hadoop and EC2

Hadoop World: Making Hadoop Easy on Amazon Web Services

Hadoop World: Hadoop Applications at Yahoo!

7 Tips for Improving MapReduce Performance