Aggregation functions in PIG

Aggregation functions in PIG

  4:12 PM    ssaikgame

---

Example1:

year	price
2015	60
2014	45
2016	34
2014	75
2015	45
2014	41

I would like to get the result as, total price of the item by year.

grunt> file = load '/tmp/yearprice.txt' using PigStorage(',') as (year:int,price:float);

grunt> data = foreach file generate year, price by $0 > 1;

grunt> group = group data by year;

grunt> sum = foreach group generate group as year, SUM(data.price);

grunt> dump sum;

2014 161
2015 105
2016 34


Example 2:

id1, 1,on,400 

id1, 2,off,100

id2, 3,on,200

i would like to get the result as "sum of $3 if $2 is 0, by ID $0"

grunt> file = load '/tmp/file' using PigStorage(',');

grunt> refineData = foreach file generate $0, $1, (($2 == 'on') ? $3 : 0);

grunt> grp = group refineData by $0;

grunt> sum = foreach grp generate group as id, SUM(refineData.$2);

grunt> dump sum;

id1, 500
id2, 200

Read More

PIG left, right outer joins error

PIG left, right outer joins error

  5:09 PM    ssaikgame

---

We know that, we can perform JOIN operation using pig, as well as left outer join, right outer join. sometimes you'll face error when working of left and right outer joins, but not to inner joins. that is because of schema.

Even if there is no schema can perform inner join, but not for left and right outer joins.

Left, right, outer joins will not be executed without well-defined bag schema. 

Left Join: Example, the below left join will not work, but

The following left join will work, because it has schema.

grunt> emp = load 'emp.txt' using PigStorage('|') as  (eid:int,name:chararray,sal:bytearray);

grunt> dept = load 'dept.txt' using PigStorage('|') as (did:int,eid:int,name:chararray);

grunt> ljoin = join emp by eid left,dept by eid;

grunt> dump ljoin;

Right Join:

grunt> rjoin = join emp by eid right,dept by eid;

grunt> dump rjoin;

Read More

Advanced scripts - PIG Part 3

Advanced scripts - PIG Part 3

  9:53 PM    ssaikgame

---

FILTER:
filter = FILTER climate by country MATCHES 'C.*a'; -> china
filter = FILTER climate by country MATCHES '.*(nda|hin).*; -> india, china

SPLIT:
SPLIT climate into B1992 if $1 == 1992, B2002 if $1 == 2002;
it split the climate bag into two records B1992 and B2002, contains 1992 and 2002 year data simuntaneously.
DUMP B1992;
DUMP B2002;

SAMPLE:
it generates sample data (random data) from the bag.
sample = SAMPLE climate 0.1; ->  it generates 1% of random data from climate.

ORDER:
order = ORDER climate BY year asc;

GROUP BY:
group = GROUP climate BY year;
group = GROUP climate BY (year, temp);

COGROUP:
it will combine the similar tuples into one group from 2 or more relations based on the grouping column.
cogroup = COGROUP a by a1, b by b1;
dump cogroup;

a1 a2 a3 b1 b2
1, 2, 3 4, 5
3, 5, 3 1, 3
2, 2, 7 5, 5
1, 4, 3 2, 5
2, 2, 3 1, 8

(1,3,2,1,2,4,1,5,2,1) --> (1,3,2,4,5)

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

cogroup_a_inner = COGROUP a BY a1 inner, b BY b1; --> atleast one tuple should be available in a.
(1,{(1,2,3),(1,4,3)},{(1,3),(1,8)})
(3,{(3,5,3)},{})
(2,{(2,2,7),(2,2,3)},{(2,5)})

cogroup_b_inner = COGROUP a BY a1, b BY b1 inner; --> atleast one tuple should be available in b.
(1,{(1,2,3),(1,4,3)},{(1,3),(1,8)})
(2,{(2,2,7),(2,2,3)},{(2,5)})
(4,{},{(4,5)})
(5,{},{(5,5)})

cogroup_a_b_inner = COGROUP a BY a1 inner, b BY b1 inner;
(1,{(1,2,3),(1,4,3)},{(1,3),(1,8)})
(2,{(2,2,7),(2,2,3)},{(2,5)})

TUPLE: 
store following data as tuple.
(1,2,3) (4,5,6)
(2,3,4) (5,6,7)

tuple = LOAD 'abc.txt' USING PigStorage(' ') AS (t1:tuple(t1a:int,t1b:int,t1c:int),t2:tuple(t2a:int,t2b:int,t2c:int));


BAG:
Store following data as Bag
{(1,2,3),(1,4,3)} {(1,3),(1,8)})
{(3,5,3)} {(3,5)})
{(2,2,7),(2,2,3)} {(2,5)})

bag1 = load 'data.txt' USING PigStorage(' ') AS (B1:Bag{t1:tuple(t1a:int,t1b:int,t1c:int)}, B2:Bag{t2:tuple(t2a:int,t2b:int)});

Introduction to PIG - Click here

Transformations or Operators in PIG - Click here

Read More

PIG Transformations or Operators - Part 2

PIG Transformations or Operators - Part 2

  2:51 AM    ssaikgame

---

LOAD:

Loading the data into “base”

grunt> base = load 'filepig.txt' using PigStorage('|') as (empid:int,ename:chararray,salary:int);

PigStorage(‘|’) è is used to separate the columns using delimiter ‘|’

The filepig.txt contains 3columns separated with delimiter “|”, that was done by PigStorage(‘|’) and name each column with title by using as (empid:int,ename:chararray,salary:int); ,so empid represents 1^st column, ename represents 2^nd column, salary represents 3^rd column.

The above line load the data from filepig.txt into base operator.

We can check the operator description by using describe base;

foreach and generate: goes through the each column and generates results.

grunt> eid = foreach base generate empid;

DUMP: dump is used to get the result from the operators/variables.

grunt> dump base;

Result:

FILTER: filter is used to filter data, example filter the employees who has salary more than 4000.

grunt> moresal = filter base by salary==4000;

            

dump moresal;

ILLUSTRATE: shows schema how the data has been stored internally.

grunt> illustrate base;

EXPLAIN:

Used to explain logical plan, map and reducer plan and scope of the operands/variables.

grunt> explain emp;

2016-06-17 02:35:55,746 [main] INFO  org.apache.hadoop.conf.Configuration.deprecation - fs.default.name is deprecated. Instead, use fs.defaultFS

2016-06-17 02:35:55,752 [main] WARN  org.apache.pig.data.SchemaTupleBackend - SchemaTupleBackend has already been initialized

2016-06-17 02:35:55,752 [main] INFO  org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer - {RULES_ENABLED=[AddForEach, ColumnMapKeyPrune, ConstantCalculator, GroupByConstParallelSetter, LimitOptimizer, LoadTypeCastInserter, MergeFilter, MergeForEach, PartitionFilterOptimizer, PredicatePushdownOptimizer, PushDownForEachFlatten, PushUpFilter, SplitFilter, StreamTypeCastInserter]}

#-----------------------------------------------

# New Logical Plan:

#-----------------------------------------------

emp: (Name: LOStore Schema: eid#10443:int,name#10444:chararray,sal#10445:chararray)

|

|---emp: (Name: LOForEach Schema: eid#10443:int,name#10444:chararray,sal#10445:chararray)

    |   |

    |   (Name: LOGenerate[false,false,false] Schema: eid#10443:int,name#10444:chararray,sal#10445:chararray)ColumnPrune:OutputUids=[10443, 10444, 10445]ColumnPrune:InputUids=[10443, 10444, 10445]

    |   |   |

    |   |   (Name: Cast Type: int Uid: 10443)

    |   |   |

    |   |   |---eid:(Name: Project Type: bytearray Uid: 10443 Input: 0 Column: (*))

    |   |   |

    |   |   (Name: Cast Type: chararray Uid: 10444)

    |   |   |

    |   |   |---name:(Name: Project Type: bytearray Uid: 10444 Input: 1 Column: (*))

    |   |   |

    |   |   (Name: Cast Type: chararray Uid: 10445)

    |   |   |

    |   |   |---sal:(Name: Project Type: bytearray Uid: 10445 Input: 2 Column: (*))

    |   |

    |   |---(Name: LOInnerLoad[0] Schema: eid#10443:bytearray)

    |   |

    |   |---(Name: LOInnerLoad[1] Schema: name#10444:bytearray)

    |   |

    |   |---(Name: LOInnerLoad[2] Schema: sal#10445:bytearray)

    |

    |---emp: (Name: LOLoad Schema: eid#10443:bytearray,name#10444:bytearray,sal#10445:bytearray)RequiredFields:null

#-----------------------------------------------

# Physical Plan:

#-----------------------------------------------

emp: Store(fakefile:org.apache.pig.builtin.PigStorage) - scope-621

|

|---emp: New For Each(false,false,false)[bag] - scope-620

    |   |

    |   Cast[int] - scope-612

    |   |

    |   |---Project[bytearray][0] - scope-611

    |   |

    |   Cast[chararray] - scope-615

    |   |

    |   |---Project[bytearray][1] - scope-614

    |   |

    |   Cast[chararray] - scope-618

    |   |

    |   |---Project[bytearray][2] - scope-617

    |

    |---emp: Load(hdfs://localhost:9000/user/hadoop/emp.txt:PigStorage('|')) - scope-610

2016-06-17 02:35:55,757 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRCompiler - File concatenation threshold: 100 optimistic? false

2016-06-17 02:35:55,759 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MultiQueryOptimizer - MR plan size before optimization: 1

2016-06-17 02:35:55,759 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MultiQueryOptimizer - MR plan size after optimization: 1

#--------------------------------------------------

# Map Reduce Plan

#--------------------------------------------------

MapReduce node scope-622

Map Plan

emp: Store(fakefile:org.apache.pig.builtin.PigStorage) - scope-621

|

|---emp: New For Each(false,false,false)[bag] - scope-620

    |   |

    |   Cast[int] - scope-612

    |   |

    |   |---Project[bytearray][0] - scope-611

    |   |

    |   Cast[chararray] - scope-615

    |   |

    |   |---Project[bytearray][1] - scope-614

    |   |

    |   Cast[chararray] - scope-618

    |   |

    |   |---Project[bytearray][2] - scope-617

    |

    |---emp: Load(hdfs://localhost:9000/user/hadoop/emp.txt:PigStorage('|')) - scope-610--------

Global sort: false

----------------

Order By:

grunt> ordersalary = order base by salary;

grunt> dump ordersalary;

Group BY:

grunt> group_data = group emp by eid;

grunt> dump group_data;

grunt> describe group_data;

 

JOIN:

Let’s take another table, name it as dept.txt, and rename the old file as emp.txt, so it doesn’t confuse.

Move the dept and emp files to HDFS.

$ hadoop fs -put dept.txt

$ hadoop fs -put emp.txt

grunt> emp = load 'emp.txt' using PigStorage('|');

grunt> dept = load dept.txt' using PigStorage('|');

grunt> joindata = join emp by $0, dept by $1;

grunt> dump joindata;

here $0 represents, the first column in the emp Bag and $1 represents second column of Dept Bag.

Left Join ->

Left, right, outer joins will not be executed without well-defined bag schema. Example, the below left join will not work, but

The following left join will work, because it has schema.

grunt> emp = load 'emp.txt' using PigStorage('|') as (eid:int,name:chararray,sal:bytearray);

grunt> dept = load 'dept.txt' using PigStorage('|') as (did:int,eid:int,name:chararray);

grunt> ljoin = join emp by eid left,dept by eid;

grunt> dump ljoin;

Right Join:

grunt> rjoin = join emp by eid right,dept by eid;

grunt> dump rjoin;

STORE:

grunt> store rjoin into '/user/hadoop/rjoin_putput'

[hadoop@localhost blog]$ hadoop fs -cat /user/hadoop/rjoin_putput/part-r-00000

UNION:

Union is used to combine two similar bags.

Union_data = union emp,dept;

CROSS:

Is used to multiple the second operator/variable data with first operator/variable.

grunt> cross_data = cross emp,dept;

it multiple dept data by emp data.

grunt>dump cross_data;

LIMIT:

Is used to limit the output by number lines.

grunt> limit_data = limit cross_data 3;

grunt> dump limit_data;

it limit the output by 3 lines.

Flatten, AggFunctions, Distinct, Cogroup, Tokenizer

click here for PART 1 - What is PIG
click here for PART 3 - Advanced scripts

Read More

What is PIG? - Hadoop

What is PIG? - Hadoop

  3:08 PM    ssaikgame

---

Apache PIG is meant for process the data i.e., for data summarization, querying and advanced querying.

Pig has its own scripting language called, pig Latin scripting. Whenever we are executing PIG actions, internally Map Reduce jobs are getting triggered by the framework.

Yahoo introduced PIG, and it is one of the great Apache project now. Pig has built on top of MapReduce.

PIG has BAG is can be ordered or disordered table and each bag has two attributes, called tuples and atoms.

NOTE: PIG will never has Metadata and Warehouse.PIG can work with 3 modes, they are LOCAL, HDFS and EMBEDDED MODE.

LOCAL mode: Input data will be taken from LFS path (Not from HDFS) and once the processing is completed the generated output will also be part of LFS path, means in the Local mode of PIG interacting, there is no intervention of HDFS.

So, here frame will be copy this data into temporary HDFS path and initiate MapReduce jobs and then process the data. Once data has been processed, again framework will copy that data into LFS path.

$> pig -x local
If there is script, use $> pig -x local <<script-name>>

HDFS mode: Input is taken from HDFS, and output is into HDFS.

$> pig
if there is a script, use $> pig <<script-name>>

EMBEDDED mode: if at all we are not able to achieve desired functionality using existing commands or operations or actions we can choose embedded mode to develop customized application.

I.e. UDF's - User Defined Functions

Following are the Default Transformations or operators in PIG:

1) LOAD                                             2) FOREACH                                      3) GENERATE 4) FILTER                                           5) DUMP                                           6) STORE 7) DESCRIBE                                      8) SPLIT                                             9) ORDER BY 10) GROUP BY                                 11) JOIN                                            12) UNION 13) CROSS                                         14) LIMIT                                          15) TOKENIZE 16) EXPLAIN                                     17) ILLUSTRATE                               18) FLATTEN 19) AGGFUNCTIONS                       20) DISTINCT                                    21) COGROUP

Following Datatypes are used in PIG

DATA TYPES                       PIG LATIN DATATYPES Int                                                                      int    String                                                              chararray float                                                                   float long                                                                   long boolean                                                             boolean byte                                                 (Default) bytearray

TIP: Execution of PIG can be done in 2 flavors, they are grunt shell (i.e. line by line) and script mode (i.e. Group of commands)

Next Lesson: Transformations or Operators in PIG

Read More