APACHE PIG
Definition en Francais
u Pig
la
est
une
haut
pour
niveau
plateforme
de
programme MapReduce utilis avec Hadoop. Le langage de cette
plateforme est appel le Pig Latin3. Pig Latin s'abstrait du langage de
programmation Java MapReduce et se place un niveau d'abstraction
sup rieur, similaire celle de SQL pour syst mes SGBDR. Pig Latin peut
tre tendue en utilisant UDF (User Defined Functions) que l'utilisateur
peut crire en Java, en Python, en JavaScript, en Ruby ou en Groovy4 et
ensuite tre utilis directement au sein du langage.
cr ation
un exemple d'un programme "Word Count" en Pig Latin:
input_lines = LOAD '/tmp/my-copy-of-all-pages-on-internet' AS (line:chararray);
Exemple
-- Extract words from each line and put them into a pig bag
-- datatype, then flatten the bag to get one word on each row
words = FOREACH input_lines GENERATE FLATTEN((line)) AS word;
-- filter out any words that are just white spaces
filtered_words = FILTER words BY word MATCHES '\\w+';
-- create a group for each word
word_groups = GROUP filtered_words BY word;
-- count the entries in each group
word_count = FOREACH word_groups GENERATE (filtered_words) AS , group AS word;
-- order the records by count
ordered_word_count = ORDER word_count BY DESC; STORE ordered_word_count INTO
'/tmp/number-of-words-on-internet';
Le programme ci-dessus va g n rer des t ches ex cutables parall les qui peuvent tre distribu s sur plusieurs
machines dans un cluster Hadoop pour compter le nombre de mots dans un ensemble de donn es telles que les
pages Web sur Internet.
Par rapport au SQL, Pig: Fr
u Utilise l' valuation paresseuse,
u utilise des extract, transform, load (ETL),
u est capable de stocker des donn es tout moment pendant un pipeline,
u d clare le plan d'ex cution,
u ex cute le workflow subdivis selon un graphe, au lieu d'une ex cution
purement s quentielle.
Avantage de Pig& Fr
u Pig Latin est un langage proc dural et s'inscrit tout naturellement dans le
paradigme du pipeline tandis que SQL est plut t d claratif.
u Pig Latin permet aux utilisateurs de sp cifier une impl mentation ou des
aspects de l'impl mentation utiliser dans l'ex cution d'un script de
plusieurs fa ons
u La programmation Pig Latin est similaire la sp cification d'un plan
d'ex cution de la requ te, qui rend plus facile pour les programmeurs de
contr ler explicitement le flux de leur t che de traitement de donn es
u SQL est orient autour de requ tes qui produisent un r sultat unique. Il g re
galement les arbres, mais n'a aucun m canisme int gr pour diviser un flux de
traitement de donn es et appliquer les diff rents op rateurs chaque sous-flux. Pig
Latin d crit un Graphe orient acyclique (DAG) plut t qu'un pipeline
u Pig Latin est capable d'inclure du code utilisateur n'importe quel point
dans le pipeline. Avec SQL, les donn es doivent d'abord tre import es
dans la base de donn es, avant que l'on puisse lancer un processus de
nettoyage et de transformation de celles-ci
Motivation
} Youre a procedural programmer
} You have huge data
} You want to analyze it
2
Motivation
} As a procedural programmer&
} May find writing queries in SQL unnatural and too restrictive
} More comfortable with writing code; a series of statements as
opposed to a long query. (Ex: MapReduce is so successful).
3
Motivation
} Data analysis goals
} Quick
} Exploit parallel processing power of a distributed system
} Easy
} Be able to write a program or query without a huge learning curve
} Have some common analysis tasks predefined
} Flexible
} Transform a data set(s) into a workable structure without much
overhead
} Perform customized processing
} Transparent
} Have a say in how the data processing is executed on the system
5
Motivation
} Relational Distributed Databases
} Parallel database products expensive
} Rigid schemas
} Processing requires declarative SQL query construction
} Map-Reduce
} Relies on custom code for even common operations
} Need to do workarounds for tasks that have different data
flows other than the expected Map Combine Reduce
6
Motivation
} Relational Distributed Databases
} Sweet Spot: Take the best of both SQL and Map-Reduce;
combine high-level declarative querying with low-level
procedural programming&Pig Latin!
} Map-Reduce
7
Pig Latin Example
Table urls: (url,category, pagerank)
Find for each suffciently large category, the average pagerank of high-pagerank urls in that
category
SQL:
SELECT category, AVG(pagerank)
FROM urls WHERE pagerank > 0.2
GROUP BY category HAVING COUNT(*) > 10^6
Pig Latin:
good_urls = FILTER urls BY pagerank > 0.2;
groups = GROUP good_urls BY category;
big_groups = FILTER groups BY COUNT(good_urls)>10^6;
Publicité
output = FOREACH big_groups GENERATE category, AVG(good_urls.pagerank);
Why Pig?
Writing mappers and reducers by hand
takes a long time.
Pig introduces Pig Latin, a scripting
language that lets you use SQL-like
syntax to define your map and reduce
steps.
Highly extensible with user-defined
functions (UDFs)
Big Picture
Pig Latin
Script
User-
Defined
Functions
10
Map-Reduce
Statements
Pig
Compile
Optimize
Write Results
Read Data
MapReduce
YARN
HDFS
Running Pig
Grunt
Script
Ambari / Hue
An example
Find the oldest 5-star movies
ratings = LOAD '/user/maria_dev/ml-100k/u.data' AS
(userID:int, movieID:int, rating:int, ratingTime:int);
This creates a relation named ratings with a given schema.
(660,229,2,891406212)
(421,498,4,892241344)
(495,1091,4,888637503)
(806,421,4,882388897)
(676,538,4,892685437)
(721,262,3,877137285)
Use PigStorage if you need a different
delimiter.
metadata = LOAD '/user/maria_dev/ml-100k/u.item' USING
PigStorage('|')AS (movieID:int, movieTitle:chararray,
releaseDate:chararray, videoRelease:chararray,
imdbLink:chararray);
DUMP metadata;
(1,Toy Story (1995),01-Jan-1995,,http://us.imdb.com/M/title-exact?Toy%20Story%20(1995))
(2,GoldenEye (1995),01-Jan-1995,,http://us.imdb.com/M/title-exact?GoldenEye%20(1995))
(3,Four Rooms (1995),01-Jan-1995,,http://us.imdb.com/M/title-exact?Four%20Rooms%20(1995))
(4,Get Shorty (1995),01-Jan-1995,,http://us.imdb.com/M/title-exact?Get%20Shorty%20(1995))
(5,Copycat (1995),01-Jan-1995,,http://us.imdb.com/M/title-exact?Copycat%20(1995))
Creating a relation from another
relation; FOREACH / GENERATE
metadata = LOAD '/user/maria_dev/ml-100k/u.item' USING PigStorage('|')
AS (movieID:int, movieTitle:chararray, releaseDate:chararray,
videoRelease:chararray, imdbLink:chararray);
nameLookup = FOREACH metadata GENERATE movieID, movieTitle,
ToUnixTime(ToDate(releaseDate, 'dd-MMM-yyyy')) AS releaseTime;
(1,Toy Story (1995),01-Jan-1995,,http://us.imdb.com/M/title-exact?Toy%20Story%20(1995))
(1,Toy Story (1995),788918400)
Group By
ratingsByMovie = GROUP ratings BY movieID;
DUMP ratingsByMovie;
(1,{(807,1,4,892528231),(554,1,3,876231938),(49,1,2,888068651), & }
(2,{(429,2,3,882387599),(551,2,2,892784780),(774,2,1,888557383), & }
ratingsByMovie: {group: int,ratings: {(userID: int,movieID: int,rating: int,ratingTime: int)}}
avgRatings = FOREACH ratingsByMovie GENERATE group AS movieID,
AVG(ratings.rating) AS avgRating;
DUMP avgRatings;
(1,3.8783185840707963)
(2,3.2061068702290076)
(3,3.033333333333333)
(4,3.550239234449761)
(5,3.302325581395349)
DESCRIBE ratings;
DESCRIBE ratingsByMovie;
DESCRIBE avgRatings;
ratings: {userID: int,movieID: int,rating: int,ratingTime: int}
ratingsByMovie: {group: int,ratings: {(userID: int,movieID: int,rating: int,ratingTime: int)}}
avgRatings: {movieID: int,avgRating: double}
FILTER
fiveStarMovies = FILTER avgRatings BY avgRating > 4.0;
(12,4.385767790262173)
(22,4.151515151515151)
(23,4.1208791208791204)
(45,4.05)
JOIN
DESCRIBE fiveStarMovies;
DESCRIBE nameLookup;
fiveStarsWithData = JOIN fiveStarMovies BY movieID, nameLookup BY movieID;
DESCRIBE fiveStarsWithData;
DUMP fiveStarsWithData;
fiveStarMovies: {movieID: int,avgRating: double}
nameLookup: {movieID: int,movieTitle: chararray,releaseTime: long}
fiveStarsWithData: {fiveStarMovies::movieID: int,fiveStarMovies::avgRating: double,
nameLookup::movieID: int,nameLookup::movieTitle: chararray,nameLookup::releaseTime: long}
(12,4.385767790262173,12,Usual Suspects, The (1995),808358400)
(22,4.151515151515151,22,Braveheart (1995),824428800)
(23,4.1208791208791204,23,Taxi Driver (1976),824428800)
ORDER BY
oldestFiveStarMovies = ORDER fiveStarsWithData BY
nameLookup::releaseTime;
DUMP oldestFiveStarMovies;
(493,4.15,493,Thin Man, The (1934),-1136073600)
(604,4.012345679012346,604,It Happened One Night (1934),-1136073600)
(615,4.0508474576271185,615,39 Steps, The (1935),-1104537600)
(1203,4.0476190476190474,1203,Top Hat (1935),-1104537600)
Putting it all together
Lets run it
Pig Latin: Diving Deeper Things you
can do to a relation
Publicité
LOAD
STORE
DUMP
u
STORE ratings INTO outRatings USING PigStorage(:);
FILTER
DISTINCT
FOREACH/GENERATE
MAPREDUCE
STREAM SAMPLE
JOIN COGROUP GROUP
CROSS CUBE
ORDER RANK LIMIT
UNION SPLIT
Diagnostics
DESCRIBE
EXPLAIN
ILLUSTRATE
UDFs
REGISTER
DEFINE
IMPORT
Some other functions and loaders
AVG CONCAT
COUNT MAX MIN SIZE SUM
PigStorage
TextLoader
JsonLoader
AvroStorage
ParquetLoader
OrcStorage
HBaseStorage
Learning more
Data Model
} Atom - simple atomic value (ie: number or string)
} Tuple
} Bag
} Map
11
Data Model
} Atom
} Tuple - sequence of fields; each field any type
} Bag
} Map
12
Data Model
} Atom
} Tuple
} Bag - collection of tuples
} Duplicates possible
} Tuples in a bag can have different field lengths and field types
} Map
13
Data Model
} Atom
} Tuple
} Bag
} Map - collection of key-value pairs
} Key is an atom; value can be any type
14
Data Model
} Control over dataflow
Ex 1 (less efficient)
spam_urls = FILTER urls BY isSpam(url);
culprit_urls = FILTER spam_urls BY pagerank > 0.8;
Ex 2 (most efficient)
highpgr_urls = FILTER urls BY pagerank > 0.8;
spam_urls = FILTER highpgr_urls BY isSpam(url);
} Fully nested
} More natural for procedural programmers (target user) than
normalization
} Data is often stored on disk in a nested fashion
} Facilitates ease of writing user-defined functions
} No schema required
15
Data Model
} User-Defined Functions (UDFs)
} Ex: spam_urls = FILTER urls BY isSpam(url);
} Can be used in many Pig Latin statements
} Useful for custom processing tasks
} Can use non-atomic values for input and output
} Currently must be written in Java
16
Speaking Pig Latin
} LOAD
} Input is assumed to be a bag (sequence of tuples)
} Can specify a deserializer with USING
} Can provide a schema with AS
newBag = LOAD filename
<USING functionName() >
<AS (fieldName1, fieldName2,&)>;
Queries = LOAD query_log.txt
USING myLoad()
AS (userID,queryString, timeStamp)
17
Speaking Pig Latin
} FOREACH
} Apply some processing to each tuple in a bag
} Each field can be:
} A fieldname of the bag
} A constant
} A simple expression (ie: f1+f2)
} A predefined function (ie: SUM, AVG, COUNT, FLATTEN)
} A UDF (ie: sumTaxes(gst, pst) )
newBag =
FOREACH bagName
GENERATE field1, field2, &;
18
Speaking Pig Latin
} FILTER
} Select a subset of the tuples in a bag
newBag = FILTER bagName
BY expression;
Publicité
} Expression uses simple comparison operators (==, !=, <, >, &)
and Logical connectors (AND, NOT, OR)
some_apples =
FILTER apples BY colour != red;
} Can use UDFs
some_apples =
FILTER apples BY NOT isRed(colour);
19
Speaking Pig Latin
} COGROUP
} Group two datasets together by a common attribute
} Groups data into nested bags
grouped_data = COGROUP results BY queryString,
revenue BY queryString;
20
Speaking Pig Latin
} Why COGROUP and not JOIN?
url_revenues =
FOREACH grouped_data GENERATE
FLATTEN(distributeRev(results, revenue));
21
Speaking Pig Latin
} Why COGROUP and not JOIN?
} May want to process nested bags of tuples before taking the
cross product.
} Keeps to the goal of a single high-level data transformation per
pig-latin statement.
} However, JOIN keyword is still available:
JOIN results BY queryString,
revenue BY queryString;
Equivalent
temp = COGROUP results BY queryString,
revenue BY queryString;
join_result = FOREACH temp GENERATE
FLATTEN(results), FLATTEN(revenue);
22
Speaking Pig Latin
} STORE (& DUMP)
} Output data to a file (or screen)
STORE bagName INTO filename
<USING deserializer ()>;
} Other Commands (incomplete)
} UNION - return the union of two or more bags
} CROSS - take the cross product of two or more bags
} ORDER - order tuples by a specified field(s)
} DISTINCT - eliminate duplicate tuples in a bag
} LIMIT - Limit results to a subset
23
Compilation
} Pig system does two tasks:
} Builds a Logical Plan from a Pig Latin script
} Supports execution platform independence
} No processing of data performed at this stage
} Compiles the Logical Plan to a Physical Plan and Executes
} Convert the Logical Plan into a series of Map-Reduce statements to
be executed (in this case) by Hadoop Map-Reduce
24
Compilation
} Building a Logical Plan
} Verify input files and bags referred to are valid
} Create a logical plan for each bag(variable) defined
25
Compilation
} Building a Logical Plan Example
A = LOAD user.dat AS (name, age, city);
B = GROUP A BY city;
C = FOREACH B GENERATE group AS city,
COUNT(A);
D = FILTER C BY city IS kitchener
OR city IS waterloo;
STORE D INTO local_user_count.dat;
Load(user.dat)
26
Compilation
} Building a Logical Plan Example
A = LOAD user.dat AS (name, age, city);
B = GROUP A BY city;
C = FOREACH B GENERATE group AS city,
COUNT(A);
D = FILTER C BY city IS kitchener
OR city IS waterloo;
STORE D INTO local_user_count.dat;
Load(user.dat)
Group
27
Compilation
} Building a Logical Plan Example
A = LOAD user.dat AS (name, age, city);
B = GROUP A BY city;
C = FOREACH B GENERATE group AS city,
COUNT(A);
D = FILTER C BY city IS kitchener
OR city IS waterloo;
STORE D INTO local_user_count.dat;
Load(user.dat)
Group
Foreach
28
Compilation
} Building a Logical Plan Example
A = LOAD user.dat AS (name, age, city);
B = GROUP A BY city;
C = FOREACH B GENERATE group AS city,
COUNT(A);
D = FILTER C BY city IS kitchener
OR city IS waterloo;
STORE D INTO local_user_count.dat;
Load(user.dat)
Group
Foreach
Filter
29
Compilation
} Building a Logical Plan Example
A = LOAD user.dat AS (name, age, city);
Publicité
B = GROUP A BY city;
C = FOREACH B GENERATE group AS city,
COUNT(A);
D = FILTER C BY city IS kitchener
OR city IS waterloo;
STORE D INTO local_user_count.dat;
Load(user.dat)
Filter
Group
Foreach
30
Compilation
} Building a Physical Plan
A = LOAD user.dat AS (name, age, city);
B = GROUP A BY city;
C = FOREACH B GENERATE group AS city,
COUNT(A);
D = FILTER C BY city IS kitchener
OR city IS waterloo;
STORE D INTO local_user_count.dat;
Only happens when output is
specified by STORE or DUMP
32
Load(user.dat)
Filter
Group
Foreach
Compilation
} Building a Physical Plan
} Step 1: Create a map-reduce job for each
Load(user.dat)
COGROUP
Map
Reduce
Filter
Group
Foreach
33
Compilation
} Building a Physical Plan
} Step 1: Create a map-reduce job for each
Load(user.dat)
COGROUP
} Step 2: Push other commands into the
map and reduce functions where
possible
Map
Filter
} May be the case certain commands
require their own map-reduce
job (ie: ORDER needs separate map-
reduce jobs)
Reduce
Group
Foreach
34
Compilation
} Efficiency in Execution
} Parallelism
} Loading data - Files are loaded from HDFS
} Statements are compiled into map-reduce jobs
35
Compilation
} Efficiency with Nested Bags
} In many cases, the nested bags created in each tuple of a COGROUP
statement never need to physically materialize
} Generally perform aggregation after a COGROUP and the
statements for said aggregation are pushed into the reduce function
} Applies to algebraic functions (ie: COUNT, MAX, MIN, SUM, AVG)
36
Compilation
} Efficiency with Nested Bags
Load(user.dat)
Map
Filter
Group
Foreach
37
Compilation
} Efficiency with Nested Bags
Load(user.dat)
Filter
Group
Foreach
Combine
38
Compilation
} Efficiency with Nested Bags
Load(user.dat)
Filter
Group
Reduce
Foreach
39
PIG CHALLENGE
Find the most popular bad movies
Defining the problem
Find all movies with an average rating less than 2.0
Sort them by the total number of ratings
Hint
We used everything you need in our earlier example of finding old movies with
ratings greater than 4.0
Only new thing you need is COUNT(). This lets you count up the number of
items in a bag.
So just like you can say AVG(ratings.rating) to get the average rating
from a bag of ratings,
You can say COUNT(ratings.rating) to get the total number of ratings for
a given groups bag.