Pig runs on Hadoop. It makes use of both the Hadoop Distributed File System, HDFS, and Hadoop's processing system MapReduce.

HDFS is a distributed file system that stores files across all of the nodes in a Hadoop cluster. It handles breaking the files into large blocks and distributing them across different machines, including making multiple copies of each block so that if any one machine fails no data is lost. It presents a POSIX like interface to users. By default, Pig reads input files from HDFS, uses HDFS to store intermediate data between MapReduce jobs, and writes its output to HDFS. As you will see later, it can also read input from and write output to sources other than HDFS, Chapter 11, Writing Load and Store Functions.

MapReduce is a simple but powerful parallel data processing paradigm. Every job in MapReduce consists of three main phases: map, shuffle, and reduce. In the map phase, the application has the opportunity to operate on each record in the input separately. Many maps are started at once, so that while the input may be gigabytes or terabytes in size, given enough machines the map phase can usually be completed in under one minute.

Part of the specification of a MapReduce job is the key on which data will be collected. For example, if you were processing web server logs for a website that required users to log in you might choose the userid to be your key so that you could see everything done by each user on your website. In the shuffle phase, which happens after the map phase, data is collected together by the key the user has chosen and distributed to different machines for the reduce phase. Every record for a given key will go to the same reducer.

In the reduce phase the application is presented each key, together with all of the records containing that key. Again this is done in parallel on many machines. After processing each group the reducer can write its output. See example Example 1.1, “MapReduce 1, hello 1, world 1” for a walk through of a simple MapReduce program. For more details on how MapReduce works see the section called “MapReduce”.

Figure 1.1. Map Reduce Illustration

Pig uses MapReduce to execute all of its data processing. It compiles the Pig Latin scripts that users write into a series of one or more MapReduce jobs that it then executes. See Example 1.2, “Pig counts Mary and her lamb” for a Pig Latin script that will do a word count of “Mary had a Little Lamb”.

-- Load input from the file named Mary, and call the single 
-- field in the record 'line'.
input = load 'mary' as (line);

-- TOKENIZE splits the line into a field for each word.
-- flatten will take the collection of records returned by
-- TOKENIZE and produce a separate record for each one, calling the single
-- field in the record word.
words = foreach input generate flatten(TOKENIZE(line)) as word;

-- Now group them together by each word.
grpd  = group words by word;

-- Count them
cntd  = foreach grpd generate group, COUNT(words);
-- Print out the results
dump cntd;

Pig Latin is a dataflow language. This means it allows users to describe how data from one or more inputs should be read, processed, and then stored to one or more outputs in parallel. These data flows can be simple linear flows like the word count example given above. They can also be complex workflows that include points where multiple inputs are joined and where data is split into multiple streams to be processed by different operators. To be mathematically precise, a Pig Latin script describes a directed acyclic graph (DAG) where the edges are data flows and the nodes are operators that process the data.

This means that Pig Latin looks different from many of the programming languages you have seen. There are no if statements or for loops in Pig Latin. This is because traditional procedural and object oriented programming languages describe control flow; data flow is a side effect of the program. Pig Latin instead focuses on data flow. For information on how to integrate the data flow described by a Pig Latin script with control flow see Chapter 9, Embedding Pig Latin in Python.

CREATE TEMP TABLE t1 AS
SELECT customer, sum(purchase) AS total_purchases
FROM transactions
GROUP BY customer;

SELECT customer, total_purchases, zipcode
FROM t1, customer_profile
WHERE t1.customer = customer_profile.customer;

-- Load the transactions file, group it by customer, and sum their total purchases
txns    = load 'transactions' as (customer, purchase);
grouped = group txns by customer;
total   = foreach grouped generate group, SUM(txns.purchase) as tp;
-- Load the customer_profile file
profile = load 'customer_profile' as (customer, zipcode);
-- join the grouped and summed transactions and customer_profile data
answer  = join total by group, profile by customer;
-- Write the results to the screen
dump answer;

Users = load ‘users’ as (name, age);
Fltrd = filter Users by age >= 18 and age <= 25;
Pages = load ‘pages’ as (user, url);
Jnd   = join Fltrd by name, Pages by user;
Grpd  = group Jnd by url;
Smmd  = foreach Grpd generate group, COUNT(Jnd) as clicks;
Srtd  = order Smmd by clicks desc;
Top5  = limit Srtd 5;
store Top5 into ‘top5sites’;

In my experience, Pig Latin use cases tend to fall into three separate categories: traditional extract transform load (ETL) data pipelines, research on raw data, and iterative processing.

The largest use case is data pipelines. A common example is web companies bringing in logs from their web servers, cleansing the data, and precomputing common aggregates before loading it into their data warehouse. In this case the data is loaded onto the grid and Pig is used to clean out records from bots and records with corrupt data. It is also used to join web event data against user databases so that user cookies can be connected with known user information.

Another example of data pipelines is using Pig offline to build behavior prediction models. Pig is used to scan through all the user interactions with a website and split the users into various segments. Then for each segment a mathematical model is produced that predicts how members of that segment will respond to types of advertisements or news articles. In this way the website can show ads that are more likely to get clicked on or news stories that are more likely to engage users and keep them coming back to the site.

Traditionally ad-hoc queries are done in languages like SQL which make it easy to quickly form a question to be answered by the data. However, for research on raw data some users prefer Pig Latin. Since Pig can operate in situations where the schema is unknown or incomplete or inconsistent and since it can easily manage nested data, researchers who want to work on data before it has been cleaned and loaded into the warehouse often prefer Pig. Researchers used to working with large data sets often are used to using scripting languages such as Perl or Python to do their processing. Users with these backgrounds often prefer the dataflow paradigm of Pig over the declarative query paradigm of SQL.

Users building iterative processing models are also starting to use Pig. Consider a news website that keeps a graph of all news stories on the web that it is tracking. In this graph each news story is a node, and edges indicate relationships between the stories. For example, all stories about an upcoming election are linked together. Every five minutes a new set of stories comes in, and the data processing engine must integrate them into the graph. Some of these stories are new, some are updates of existing stories, and some supersede existing stories. Some data processing steps need to operate on this entire graph of stories. For example, a process that builds a behavioral targeting model needs to join user data against this entire graph of stories. Re-running the entire join every five minutes is not feasible since it can not be completed in five minutes with a reasonable amount of hardware. But the model builders do not want to only update these models daily as that means an entire day of missed serving opportunities.

To cope with this problem it is possible to first do a join against the entire graph on a regular basis, for example daily. Then, as new data comes in every five minutes a join can be done with just the new incoming data and the results combined with the results of the join against the whole graph. This combination step takes some care as the five minute data contains the equivalent of inserts, updates, and deletes on the entire graph. It is possible and reasonably convenient to express this combination in Pig Latin.

One point that is implicit in everything that has been said so far is that Pig (like MapReduce) is oriented around batch processing of data. If you need to process gigabytes or terabytes of data Pig is a good choice. But it expects to read all the records of a file and write all of its output sequentially. For workloads that require writing single or small groups of records or looking up many different records in random order Pig (like MapReduce) is not a good choice. See the section called “NoSQL Databases” for a discussion of applications that are good for these use case.

Early on the Pig team discovered that people who came to the Pig project as potential contributors did not always understand what the project was about. This led to confusion on their part. They were not sure how to best contribute or what contributions would be accepted and what would not. So the team produced a statement of the project's philosophy that summarizes what Pig aspires to be: