Pig data model and operators

Tutorial 13 :Apache PigTechnoDJ · Tuple, bag, mapData Model and Execution Mode in Apache PigCode With Z · Data model and execution modesPig Latin Data Model and Data TypesCode With Z · Pig data model and typesApache Pig Tutorial 1edureka! · Pig overview

What Apache Pig isSearch on YouTubeBookmarkSaved

Apache Pig is an abstraction over MapReduce. It is a tool that analyses large data sets. Analysts express the work as data flows. The language for these flows is Pig Latin. Pig Latin gives a rich set of types and operators. Programmers can add their own functions (UDFs) to read, write and process data.

Pig converts every script into optimised MapReduce jobs. The jobs run across the Hadoop cluster. They read input from HDFS. They write results back to HDFS.

Why Pig existsSearch on YouTubeBookmarkSaved

Hand-written MapReduce jobs need long Java programs. The slide compares about 200 lines of Java with about 10 lines of Pig Latin. It reports development time falling by almost 16 times. Pig Latin looks like SQL, so it is quick to learn. Built-in operators cover join, filter, order and more. Nested types such as tuple, bag and map hold semi-structured data. The multi-query approach keeps scripts short.

Features of PigSearch on YouTubeBookmarkSaved

1. Rich set of operators for join, sort and filter.
2. Ease of programming: Pig Latin resembles SQL.
3. Optimisation: tasks optimise their own execution plans.
4. Extensibility: users build new functions from old operators.
5. UDFs: user-defined functions written in Java run inside scripts.
6. All kinds of data: structured and unstructured data, with results stored in HDFS.

Applications of PigSearch on YouTubeBookmarkSaved

Data model of PigSearch on YouTubeBookmarkSaved

Pig data nests in four levels. A field holds one value. A tuple holds an ordered set of fields. A bag holds a collection of tuples. A relation is a named bag. A map stands apart: it holds key-value pairs.

A field can itself hold a tuple, bag or map. This recursion lets Pig model JSON-like data with no fixed schema.

FieldSearch on YouTubeBookmarkSaved

A field is the smallest unit of data. It is also called an atom. Any single value is a field, whatever its type. Atomic types are int, long, float, double, chararray and bytearray. Fields in one tuple need not share a type. A field can also hold a complex value: a tuple, bag or map.

-- two fields: a name and an age
'rajash', 30

TupleSearch on YouTubeBookmarkSaved

A tuple is an ordered set of fields. It resembles a row of a relational table. Tuples use parentheses: (field1, field2, field3, ...). Fields are reached by position: `$0` is the first field, `$1` the second. A tuple has fixed length. Its fields may mix types.

students = LOAD 'students.txt' USING PigStorage(',')
    AS (id:int, name:chararray, age:int, city:chararray);
-- each input line becomes ONE tuple: (001, 'Rajiv', 21, 'Hyderabad')

BagSearch on YouTubeBookmarkSaved

A bag is a collection of tuples. Duplicates are allowed. Each tuple may hold any number of fields. Tuples in one bag may differ in size and type. Bags use curly braces: {tuple1, tuple2, ...}. There are two kinds. An outer bag is a relation. An inner bag is nested in a tuple and comes from GROUP.

grouped = GROUP students BY city;
-- each 'group' field of the result is an inner bag, e.g. {(Raja,30),(Mohammad,45)}

RelationSearch on YouTubeBookmarkSaved

A relation is a bag of tuples with a name. The name is the alias left of =. Every LOAD, FILTER, GROUP and JOIN makes a new relation. Relations are unordered. No order of tuples is promised. Chain aliases to build a pipeline.

A = LOAD 'students.txt' AS (id:int, name:chararray, age:int, city:chararray);
B = FILTER A BY age > 21;
C = GROUP B BY city;
DUMP C;

Here A, B and C are three relations. Each step feeds the next.

MapSearch on YouTubeBookmarkSaved

A map holds key-value pairs: [key#value, key#value, ...]. Each key is a chararray and is unique in the map. Each value may be any type, even a tuple, bag or map. Maps model semi-structured data such as JSON profiles.

records = LOAD 'profiles.json' AS (details:map[]);
name = FOREACH records GENERATE details#'name';
-- input map: [name#Raja, age#30]; output: (Raja)

A field holds one value. A map holds many values, each reached by its own key.

Architecture of PigSearch on YouTubeBookmarkSaved

A script passes through five stages. The Grunt shell or Pig Server takes the script. The parser builds a DAG. The optimiser trims the plan. The compiler emits MapReduce jobs. The execution engine runs them on Hadoop with HDFS.

Grunt Shell and Pig ServerSearch on YouTubeBookmarkSaved

The Grunt shell is the interactive command line. Analysts type one statement at a time and see results at once. It suits data exploration and debugging. Start it with pig. The Pig Server is a Java class. It submits Pig programs from inside a Java application with no shell. It runs whole .pig scripts, for example with pig script.pig.

pig
grunt> A = LOAD 'file.csv';
grunt> DUMP A;

pig script.pig

PigServer p = new PigServer(...);

Parser, optimiser and compilerSearch on YouTubeBookmarkSaved

The script turns into a plan in three steps.

1. Parse the script. The parser checks syntax and types. It builds a DAG. Nodes are operators such as LOAD, FILTER and GROUP. Edges are data flows.
2. Optimise the DAG. The optimiser applies logical rewrites. Examples are projection pushdown, filter pushdown and merged operations. These cut the work MapReduce must do.
3. Compile the plan. The compiler turns the logical plan into a physical plan. It then emits a series of MapReduce jobs for Hadoop.

DAG means Directed Acyclic Graph. It is the internal picture of the data flow.

Execution engine, MapReduce and HDFSSearch on YouTubeBookmarkSaved

1. The execution engine submits the compiled jobs to the cluster. It tracks progress until the jobs finish.
2. MapReduce processes input splits in parallel in map tasks. Reduce tasks aggregate results. GROUP, JOIN and ORDER use the reduce phase hard.
3. HDFS holds the input data and takes back the output. Data splits into blocks across the cluster.

Execution modesSearch on YouTubeBookmarkSaved

pig -x local

pig

Execution mechanismsSearch on YouTubeBookmarkSaved

Pig Latin operatorsSearch on YouTubeBookmarkSaved

This lecture covers LOAD through SPLIT in the list above. JOIN and RANK belong to l06.

LOADSearch on YouTubeBookmarkSaved

LOAD reads data from the file system into a relation. PigStorage is the default load function. Pass a delimiter with USING PigStorage(','). Give the schema with AS (field:type, ...).

A = LOAD '/path/file.txt' USING PigStorage(',') AS (name:chararray, age:int, marks:float);

A = LOAD 'data' USING PigStorage(',') AS (name:chararray, age:int, marks:float);
DUMP A;

John,18,4.0
Mary,19,3.8
Bill,20,3.9
Joe,18,3.8

(John,18,4.0)
(Mary,19,3.8)
(Bill,20,3.9)
(Joe,18,3.8)

STORESearch on YouTubeBookmarkSaved

STORE writes a relation to the file system. It writes a directory, not one file. Output rows use tabs between fields.

STORE A INTO 'output_latin';

-- relation A holds (John,18,4.0), (Mary,19,3.8), (Bill,20,3.9), (Joe,18,3.8)
STORE A INTO 'output_latin';
-- output directory holds: John 18 4.0 / Mary 19 3.8 / Bill 20 3.9 / Joe 18 3.8

CROSSSearch on YouTubeBookmarkSaved

CROSS builds the Cartesian product of two or more relations. Each tuple of A pairs with each tuple of B. Fields glue in order: all of A first, then all of B.

X = CROSS A, B;
DUMP X;

(1,2,3,2,4)
(1,2,3,8,9)
(1,2,3,1,3)
(4,2,1,2,4)
(4,2,1,8,9)
(4,2,1,1,3)

DISTINCTSearch on YouTubeBookmarkSaved

DISTINCT drops duplicate tuples. It compares whole tuples. Two rows count as duplicates only if every field matches.

X = DISTINCT A;

X = DISTINCT A;
DUMP X;

(8,3,4)
(1,2,3)
(4,3,3)
(4,3,3)
(1,2,3)

(1,2,3)
(4,3,3)
(8,3,4)

ORDER BYSearch on YouTubeBookmarkSaved

ORDER BY sorts a relation on one or more fields. Add ASC or DESC for direction. Ties may land in any order.

A = LOAD 'data' USING PigStorage(',') AS (a1:int, a2:int, a3:int);
X = ORDER A BY a3 DESC;

A = LOAD 'data' USING PigStorage(',') AS (a1:int, a2:int, a3:int);
X = ORDER A BY a3 DESC;
DUMP X;

(1,2,3)
(4,2,1)
(8,3,4)
(4,3,3)
(7,2,5)
(8,4,3)

(7,2,5)
(8,3,4)
(1,2,3)
(4,3,3)
(8,4,3)
(4,2,1)

LIMITSearch on YouTubeBookmarkSaved

LIMIT caps the number of output tuples. It promises nothing about which tuples survive. Order the relation first for a stable top-N.

X = LIMIT A 3;

X = LIMIT A 3;
DUMP X;

(1,2,3)
(4,2,1)
(8,3,4)

FILTERSearch on YouTubeBookmarkSaved

FILTER keeps tuples that pass a test. It resembles the WHERE clause of SQL. Dropped tuples leave the pipeline.

A = LOAD 'data' USING PigStorage(',') AS (f1:int, f2:int, f3:int);
X = FILTER A BY f3 == 3;

A = LOAD 'data' USING PigStorage(',') AS (f1:int, f2:int, f3:int);
X = FILTER A BY f3 == 3;
DUMP X;

(1,2,3)
(4,2,1)
(8,4,3)
(4,3,3)
(7,2,5)

(1,2,3)
(8,4,3)
(4,3,3)

FILTER deptdata BY state == 'Kerala';
FILTER studentdata BY name == 'ABC';

Kept rows are those with third field 3. Rows with third field 1 or 5 drop out.

FOREACH GENERATESearch on YouTubeBookmarkSaved

FOREACH ... GENERATE projects columns from a relation. It resembles the SELECT clause of SQL. It picks a subset of fields from each tuple.

X = FOREACH A GENERATE f1;

(1,2,3)
(4,2,1)
(8,4,3)
(4,3,3)
(7,2,5)

(1)
(4)
(8)
(4)
(7)

foreach_data = FOREACH student_details GENERATE id, age, city;
-- (1,21,Hyderabad), (2,22,Kolkata), (3,22,Delhi), (4,21,Pune)

GROUPSearch on YouTubeBookmarkSaved

GROUP collects rows by a key field. Each output tuple has two fields. The first is group, the key value. The second is a bag with all matching tuples. This bag is the inner bag.

A = LOAD 'data' USING PigStorage(',') AS (name:chararray, age:int, marks:float);
B = GROUP A BY age;

A = LOAD 'data' USING PigStorage(',') AS (name:chararray, age:int, marks:float);
B = GROUP A BY age;
DUMP B;

(John,18,4.0)
(Mary,19,3.8)
(Bill,20,3.9)
(Joe,18,3.8)

(18,{(John,18,4.0),(Joe,18,3.8)})
(19,{(Mary,19,3.8)})
(20,{(Bill,20,3.9)})

COUNTSearch on YouTubeBookmarkSaved

COUNT counts elements of a bag. It needs a prior GROUP. It skips NULL values.

A = LOAD 'data' USING PigStorage(',') AS (a1:int, a2:int, a3:int);
B = GROUP A BY a1;
Result = FOREACH B GENERATE COUNT(A);

(1,{(1,3,7),(1,3,6)})
(3,{(3,8,9)})
(5,{(5,1,2),(5,2,3)})
(7,{(7,2,5)})

(2)
(1)
(2)
(1)

MAXSearch on YouTubeBookmarkSaved

MAX takes the largest value in a single-column bag. It needs a prior GROUP. It works on numbers and on chararray fields. It skips NULL values.

A = LOAD 'data' USING PigStorage(',') AS (a1:int, a2:int, a3:int);
B = GROUP A BY a1;
Result = FOREACH B GENERATE group, MAX(A.a3);

(Sam,{(Sam,f,3.8),(Sam,e,3.4),(Sam,d,2.2)})
(Rohan,{(Rohan,c,3.6),(Rohan,b,2.8),(Rohan,a,2.4)})

(Sam,3.8)
(Rohan,3.6)

(101,Asha,Sales,40000)
(102,Manoj,Sales,50000)
(103,Divya,HR,45000)
(104,Kiran,HR,55000)
(105,Ravi,Sales,45000)

grpd = GROUP emps BY dept;
-- (Sales,{(101,Asha,Sales,40000),(102,Manoj,Sales,50000),(105,Ravi,Sales,45000)})
-- (HR,{(103,Divya,HR,45000),(104,Kiran,HR,55000)})

summary = FOREACH grpd GENERATE group, COUNT(emps), AVG(emps.sal), MAX(emps.sal);
DUMP summary;

(Sales,3,45000,50000)
(HR,2,50000,55000)

TOKENIZESearch on YouTubeBookmarkSaved

TOKENIZE splits a string into a bag of word-tuples. Each input line yields one bag. It is the first half of text pipelines.

Result = FOREACH A GENERATE TOKENIZE(a1);

Apache Pig is a high level data flow platform.
It executes Map Reduce programs of Hadoop.
The language for Pig is pig Latin.

{(Apache),(Pig),(is),(a),(high),(level),(data),(flow),(platform)}
{(It),(executes),(Map),(Reduce),(programs),(of),(Hadoop)}
{(The),(language),(for),(Pig),(is),(pig),(Latin)}

FLATTENSearch on YouTubeBookmarkSaved

FLATTEN un-nests a bag into plain tuples. It turns each element of the bag into its own row. TOKENIZE plus FLATTEN turns each text line into one row per word.

words = FOREACH input GENERATE FLATTEN(TOKENIZE(line, ' ')) AS word;

Word countSearch on YouTubeBookmarkSaved

Word count chains four steps. Load lines. Spread them into words. Group equal words. Count each group. This pattern fits many text pipelines.

input = LOAD '/path/to/file/' AS (line:chararray);
words = FOREACH input GENERATE FLATTEN(TOKENIZE(line, ' ')) AS word;
grouped = GROUP words BY word;
wordcount = FOREACH grouped GENERATE group, COUNT(words);

This is a hadoop class
hadoop is a bigdata technology

(This)(is)(a)(hadoop)(class)(hadoop)(is)(a)(bigdata)(technology)

(a,{(a),(a)})
(is,{(is),(is)})
(This,{(This)})
(class,{(class)})
(hadoop,{(hadoop),(hadoop)})
(bigdata,{(bigdata)})
(technology,{(technology)})

(a,2)
(is,2)
(This,1)
(class,1)
(hadoop,2)
(bigdata,1)
(technology,1)

The Hive version needs a nested query (Hive is the topic of l08):

SELECT word, COUNT(1) AS count FROM
  (SELECT explode(split(sentence, ' ')) AS word FROM texttable) tempTable
GROUP BY word;

Four Pig Latin lines replace one nested Hive query. Both stay far shorter than hand-written MapReduce Java.

UNIONSearch on YouTubeBookmarkSaved

UNION merges two or more relations into one. It stacks all tuples in one bag. Input schemas need not match. This leniency beats the strict union of SQL.

student = UNION student1, student2;

student1 = LOAD 'student_data1.txt' USING PigStorage(',')
    AS (id:int, firstname:chararray, lastname:chararray, phone:chararray);
student2 = LOAD 'student_data2.txt' USING PigStorage(',')
    AS (id:int, firstname:chararray, lastname:chararray, phone:chararray);
student = UNION student1, student2;

SPLITSearch on YouTubeBookmarkSaved

SPLIT routes one relation into several. Each branch keeps tuples that pass its own IF test. It splits one pipeline into many by threshold or category.

SPLIT student_details INTO student_details1 IF age < 23, student_details2 IF (22 < age AND age > 25);

student_details = LOAD 'student_details.txt' USING PigStorage(',')
    AS (id:int, firstname:chararray, lastname:chararray, age:int,
        phone:chararray, city:chararray);
SPLIT student_details INTO young IF age < 23, senior IF age > 25;

young:  2 tuples (ages 20, 22)
senior: 2 tuples (ages 26, 30)

End-to-end scriptSearch on YouTubeBookmarkSaved

Each line maps to one idea from this lecture. The engine path runs beside it.

students = LOAD 'students.txt' USING PigStorage(',')
    AS (id:int, name:chararray, age:int, city:chararray);
adults = FILTER students BY age > 21;
grouped = GROUP adults BY city;
DUMP grouped;

Path through the engine:

1. Grunt shell or Pig Server takes the script.
2. Parser builds the DAG and checks types.
3. Optimiser pushes the FILTER earlier where possible.
4. Compiler emits MapReduce jobs.
5. Execution engine submits the jobs. GROUP triggers a reduce phase.
6. Results read from and written to HDFS.

Quick self-testSearch on YouTubeBookmarkSaved