Hive Lab1 : Inner tables and External Tables

hive> show databases;

-- lists existed databases;

hive> create table mysamp(line string);
OK
Time taken: 0.543 seconds

hive> load data local inpath 'file1'             
    > into table mysamp;

[training@localhost ~]$ hadoop fs -ls /user/hive/warehouse/mysamp
Found 1 items
-rw-r--r--   1 training supergroup         50 2016-06-15 07:21 /user/hive/warehouse/mysamp/file1

when ever a table is created under default database, in hdfs one directory will created with table name.

 when you load a file into table,
 the file will be copied into backend table's directory.

 location of default database:
  /user/hive/warehouse 

hive> select * from mysamp;
OK
aaaaaaaaaaa
aaaaaaaaaaaaaaaa
aaaaaaaaaaaaaaaaaaaa
Time taken: 0.166 seconds
hive> 

when select statement is submitted hive starts reading from table's backend directory(all files of directory).

[training@localhost ~]$ cat > file2
bbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbb

hive> load data local inpath 'file2'
    >  into table mysamp;


[training@localhost ~]$ hadoop fs -ls /user/hive/warehouse/mysamp
Found 2 items
-rw-r--r--   1 training supergroup         50 2016-06-15 07:21 /user/hive/warehouse/mysamp/file1
-rw-r--r--   1 training supergroup         55 2016-06-15 07:28 /user/hive/warehouse/mysamp/file2
[training@localhost ~]$ 

hive> select * from mysamp;
OK
aaaaaaaaaaa
aaaaaaaaaaaaaaaa
aaaaaaaaaaaaaaaaaaaa
bbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbb
Time taken: 0.065 seconds
hive> 

-- here, hive reading from all files of backend directory.

[training@localhost ~]$ hadoop fs -ls /user/hive/warehouse/mysamp
ls: Cannot access /user/hive/warehouse/mysamp: No such file or directory.

-- as above operation, 
  when table is dropped, 
    in hdfs, the backend directory will also be deleted.

--- means, metadata and data both are dropped.
  this is behavior of "inner table" 

two types of tables.
____________________
 i) inner table. (default)
  --if table is dropped, both metadata and data will be deleted 
  from hive table will be deleted.
  from hdfs, backend directory will be deleted.

 ii) external table.
  -- if table is dropped, only metadata will be deleted.

   from hive table only will be deleted. but still backend directory is available with data files.

  so that we can re use.
creating external table:


hive> create external table ursamp(line  string);

hive> load data local inpath 'file1' 
    >  into table ursamp;

hive> load data local inpath 'file2'
    >  into table ursamp;

hive> 

[training@localhost ~]$ hadoop  fs -ls /user/hive/warehouse/ursamp
Found 2 items
-rw-r--r--   1 training supergroup         50 2016-06-15 07:40 /user/hive/warehouse/ursamp/file1
-rw-r--r--   1 training supergroup         55 2016-06-15 07:40 /user/hive/warehouse/ursamp/file2
[training@localhost ~]$ 

hive> select * from ursamp;
OK
aaaaaaaaaaa
aaaaaaaaaaaaaaaa
aaaaaaaaaaaaaaaaaaaa
bbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbb
Time taken: 0.081 seconds
hive> 

hive> drop table ursamp;


[training@localhost ~]$ hadoop  fs -ls /user/hive/warehouse/ursamp
Found 2 items
-rw-r--r--   1 training supergroup         50 2016-06-15 07:40 /user/hive/warehouse/ursamp/file1
-rw-r--r--   1 training supergroup         55 2016-06-15 07:40 /user/hive/warehouse/ursamp/file2
[training@localhost ~]$ 


from hive ursamp is deleted,
  but still ursamp directory with data files safely available in warehouse location.

adv of exteral table reuse of the data.

___________________________________

how to reuse.?

hive> create table ursamp(line string);

-- create behavior,

   if table's directory (ursamp) is already available , it will use it.
 if not available, it will create directory.

_____________________________

creating databases.

hive> create database mydb;
OK
Time taken: 0.029 seconds
hive> use mydb;                       
OK
Time taken: 0.013 seconds
hive> create table urtab(line string);
OK
Time taken: 0.04 seconds
hive> 

[training@localhost ~]$ hadoop fs -ls /user/hive/warehouse/mydb.db
Found 1 items
drwxr-xr-x   - training supergroup          0 2016-06-15 07:56 /user/hive/warehouse/mydb.db/urtab
[training@localhost ~]$ 

in hdfs, under warehouse location with name database and with .db extension one directory will be created.

    mydb.db

 hive> load data local inpath 'file1' 
    >   into table urtab;

[training@localhost ~]$ hadoop fs -ls /user/hive/warehouse/mydb.db/urtab
Found 1 items
-rw-r--r--   1 training supergroup         50 2016-06-15 07:59 /user/hive/warehouse/mydb.db/urtab/file1

________________________________

create table's directory under custom location.

hive> create table mytab(line string) 
    >  location '/user/myloc';
OK
Time taken: 0.034 seconds
hive> load data local inpath 'file1' into table mytab;
Copying data from file:/home/training/file1
Copying file: file:/home/training/file1
Loading data to table default.mytab
OK
Time taken: 0.131 seconds
hive> 

[training@localhost ~]$ hadoop fs -ls /user/myloc
Found 1 items
-rw-r--r--   1 training supergroup         50 2016-06-15 08:02 /user/myloc/file1
[training@localhost ~]$ 


--- with name /user/myloc one directory is created for mytab table.

______________________________________



  

Pig Lab8

grunt> cat pdemo/profiles
ravi,90000
rani,90000
giri,10000
mani,50000
siva,50000
hari,50000
mani,50000
vani,10000
veni,5000
grunt> 
package pig.udfs;

import java.io.IOException;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;

public class MathRank extends EvalFunc<Integer>
{
    int prev = 0;
    int cnt = 0;
public Integer  exec(Tuple v)
    throws IOException
    {
  int cv = (Integer) v.get(0);
  if (cv!=prev) cnt++;

prev = cv;
  return new Integer(cnt);
    }
}

export into --> /home/training/Desktop/pjars.jar

grunt> emp = load 'pdemo/profiles'
>>    using PigStorage(',')
>>  as (name:chararray, sal:int);
grunt> e = order emp by sal desc;

grunt> register  Desktop/pjars.jar;
grunt> define rank  pig.udfs.MathRank();
grunt> ee = foreach e generate *, rank(sal) as rank;
grunt> dump ee
(ravi,90000,1)
(rani,90000,1)
(mani,50000,2)
(siva,50000,2)
(hari,50000,2)
(mani,50000,2)
(giri,10000,3)
(vani,10000,3)
(veni,5000,4)
_________________________________________

[training@localhost ~]$ cat > test1
100     345     890
123     346     340
340     240     140
[training@localhost ~]$ hadoop fs -copyFromLocal test1  pdemo
copyFromLocal: Target pdemo/test1 already exists
[training@localhost ~]$ hadoop fs -copyFromLocal test1  pdemo/t1
[training@localhost ~]$ 


Udf to find row level max.

package pig.udfs;

import java.io.IOException;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;

public class RowMax extends EvalFunc<Integer> 
{
  public Integer exec(Tuple v) throws IOException
  {
  int a = (Integer)v.get(0);
  int b = (Integer)v.get(1);
  int c = (Integer)v.get(2);
  int big=0;   // 10 34 5
  if(a>big) big=a;
  if(b>big) big=b;
  if(c>big) big=c;
 return new Integer(big);
  }
}

export into -->   Desktop/pjars.jar




grunt> t1 = load 'pdemo/t1' 
>>      as (a:int, b:int, c:int);
grunt> register Desktop/pjars.jar;
grunt> define rmax pig.udfs.RowMax();
grunt> r1 = foreach t1 generate *, rmax(*) as rm;
grunt> dump r1


 the above udf fits only for 3 numeric fields.
________________________

    example of   v.getAll()


[training@localhost ~]$ cat > t2
100     200     400     600     120
300     450     123     567     678
[training@localhost ~]$ hadoop fs -copyFromLocal  t2 pdemo
[training@localhost ~]$ 

package pig.udfs;

import java.io.IOException;
import java.util.List;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;

public class RMax extends EvalFunc<Integer> 
{
  public Integer exec(Tuple v) throws IOException
  {  
List<Object> lobs  =    v.getAll();
int max = 0;
for(Object  o : lobs)  //  120,456,345,654,600
{
    int val = (Integer)o;
    max = Math.max(val, max);
 
}
 
    
 return new Integer(max);
  }
}

grunt> t2 = load 'pdemo/t2'       
>>   as (a:int, b:int, c:int, d:int, e:int);
grunt> register Desktop/pjars.jar;
grunt> define rmax2 pig.udfs.RMax();
grunt> r2 = foreach t2 generate * , rmax2(*) as max;
grunt> dump r2

(100,200,400,600,120,600)
(300,450,123,567,678,678)

_______________________________________

Handling Unstructured Text.


Filtering Dell news.

[training@localhost ~]$ cat > news
Modi met obama           
Sonia  went to Usa
Dell implement big data
Modi has order 1 lakh Dell laptops for opts
[training@localhost ~]$ hadoop fs -copyFromLocal  news pdemo
[training@localhost ~]$ 

grunt> news = load 'pdemo/news'   
>>    as (line:chararray);
grunt> dump news

grunt> news2 = foreach news generate *,
>>     INDEXOF(line,'Dell') as idx;
grunt> dump news2

(Modi met obama,-1)
(Sonia  went to Usa,-1)
(Dell implement big data,0)
(Modi has order 1 lakh Dell laptops for opts,22)
grunt> 


no we can apply filter to fetch out Dell news.
 if line contain Dell word,  idx value will be >= 0.
if not, idx = -1

Pig Lab7

package pig.udfs;
---
----
public  class Concat
   extends EvalFunc<>
{
  public  String  exec(Tuple v)  
    throws IOException
  {
     String fn =(String) v.get(0);
     String sn =(String) v.get(1);
    String name = fn+" "+sn;  
     return name;
  }
}
_____________________________

when java class extended EvalFunc ,. 

the class will get udf functionality.

  the udf code(logic) to be written in 

exec method(function).

 exec has  Tuple variable, to capture 

passed values from function call of 

foreach statement.

ex:

x = foreach y generate fn, sn, 
   concat(fn,sn) as name;

Tuple variable can hold multiple 

values.

    to get first value.

     v.get(0)
2nd value ---  v.get(1)...
______________________________
 to get all values,
             v.getAll()
__________________________________

exec() function is executed for n 

times. n is number of tuples in the 

relation.

_________________________________

steps of udf
________________

step1) Develop UDF class 
step2) export class(es) into jar file.
step3) register jar file in pig.
step4) create temporary function for 

Pig Udf class.
step5) call the function.
_________________________________
grunt> emp = load 'pdemo/emp'        
>>    using PigStorage(',') 
>>   as (id:int, name:chararray, 

sal:int,
>>     sex:chararray, dno:int);
grunt> cat pdemo/emp
101,vino,26000,m,11
102,Sri,25000,f,11
103,mohan,13000,m,13
104,lokitha,8000,f,12
105,naga,6000,m,13
101,janaki,10000,f,12
201,aaa,30000,m,12
202,bbbb,50000,f,13
203,ccc,10000,f,13
204,ddddd,50000,m,13
grunt> 

task:
 design udf to convert name into 

uppercase.


step1>

 Design udf class.

elipse steps.

create project:

file ---> new  ---> java project.
    MyUdfs

configure pig jar file:

src ---> build path ---> configure 

build path ---> libraries --> add 

external jar .

   /usr/lib/pig/pig-core.jar 

create package

 src ---> new --> package 

ex:      pig.udfs

create java class.
  
 package(pig.udfs) ---> new ---> class

 ex:   ToBigCase

package pig.udfs;

import java.io.IOException;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;

public class ToBigCase extends 

EvalFunc<String>
{
    public   String exec(Tuple v)
     throws IOException
     {
    String val =(String) v.get(0);
    String bname = 

val.toUpperCase();
    return  bname;
     }
}

Step2.
 Export into jar file.

  project(MyUdfs) --->
     export --->
       java --->
          jar file

    ex:  

/home/training/Desktop/pjars.jar

Step3:
 Register jar file into pig.

grunt> register Desktop/pjars.jar;

step4: create temporary function.

grunt> define tobig  

pig.udfs.ToBigCase();

step5: call the function.

grunt> 

grunt> describe emp;
emp: {id: int,name: chararray,sal: 

int,sex: chararray,dno: int}
grunt> e = foreach emp generate 
>>    id, tobig(name) as name, sal ,   


>>      tobig(sex) as sex, dno;
grunt> dump e

(101,VINO,26000,M,11)
(102,SRI,25000,F,11)
(103,MOHAN,13000,M,13)
(104,LOKITHA,8000,F,12)
(105,NAGA,6000,M,13)
(101,JANAKI,10000,F,12)
(201,AAA,30000,M,12)
(202,BBBB,50000,F,13)
(203,CCC,10000,F,13)
(204,DDDDD,50000,M,13)
_____________________________
Assignment:

comment file:

 hadoop       is    going to be 

platform of all bigdata frames.





lines = load 'pdemo/comment' 
    as (line:chararray);

nlines = foreach lines generate 
     removeWhiteSp(line) as line;

 expected output:
   
  hadoop is going to be .....
___________________________-
Assignment 2.
 top 3 salaries.

[training@localhost ~]$ cat > profiles
ravi,90000
rani,90000
giri,10000
mani,50000
siva,50000
hari,50000
mani,50000 
vani,10000
veni,5000 
[training@localhost ~]$ hadoop fs -

copyFromLocal profiles pdemo
[training@localhost ~]$ 

_______________________

Assignment 3.

 make all hyphons  to comma.



101,aaa,30000-11-hyd,m
   :
   :
109,bbbb,40000-12-del,f
______________________________







 ex udfs ...  neutral()

public class Neutral extends
    EvalFunc<String>
{
  public String exec(Tuple v) 
   throws IOException
  {
     String val =(String) v.get(0);
   String nval =   val.replaceAll

("-",",");
    return nval
  }
}

export into -->  pjars.jar.

 register Desktop/pjars.jar;

define  neutral  pig.udfs.Nuetral();

 ds = load 'pdemo/file1' 
   as (line:chararray);

 nds = foreach ds generate   
         neutral(line) as line;

     line ---> 101,aaa,3000,11,hyd,m
     
 store nds into 'myloc';

 x = load 'myloc/part-m-00000'
   using PigStorage(",")
  AS (id:int, name:chararray, sal:int, 
  dno:int, city:chararray, 

sex:chararray);

______________________________Assignment2 solution is given next post.-----------------



























































































































































  

Pig Lab6

task:
  getting top3 salaried list.

 (case: a salary can be taken by multiple people).

[training@localhost ~]$ cat > samps
aaa,10000
bbb,80000
ccc,90000
ddd,90000
eeee,90000
ffff,80000
mmmmm,80000
nnnnn,70000
nnnn,70000
nn,60000 
m,65000 
xx,10000
[training@localhost ~]$ hadoop fs -copyFromLocal samps pdemo
[training@localhost ~]$ 

grunt> e = load 'pdemo/samps'     
>>     using PigStorage(',')
>>   as (name:chararray, sal:int);
grunt> sals = foreach e generate sal;
grunt> sals = distinct sals;
grunt> sals2 = order sals by sal desc;
grunt> top3 = limit sals2 3;
grunt> dump top3

grunt> describe top3
top3: {sal: int}
grunt> describe e
e: {name: chararray,sal: int}
grunt> res = join e by sal , top3 by sal;
grunt> describe res;
res: {e::name: chararray,e::sal: int,top3::sal: int}
grunt> res = foreach res generate e::name as name,
>>         e::sal as sal;
grunt> dump res
(nnnnn,70000)
(nnnn,70000)
(bbb,80000)
(ffff,80000)
(mmmmm,80000)
(ccc,90000)
(ddd,90000)
(eeee,90000)
_____________________________________
Cross:
 gives cartisian product.

used for non-equi functionalities of joins.

[training@localhost ~]$ cat > matrimony
Ravi,25,m
Rani,24,f
Ilean,23,f
trisha,27,f
Kiran,29,m
madhu,22,m
avi,26,m
srithi,21,f
[training@localhost ~]$ hadoop fs -copyFromLocal matrimony pdemo
[training@localhost ~]$ 
grunt> matri = load 'pdemo/matrimony' 
>>    using PigStorage(',')
>>   as (name:chararray, age:int, sex:chararray);
grunt> males = filter matri by (sex=='m');
grunt> fems = filter matri by (sex=='f');
grunt> cr = cross males, fems;
grunt> describe cr
cr: {males::name: chararray,males::age: int,males::sex: chararray,fems::name: chararray,fems::age: int,fems::sex: chararray}
grunt> mf = foreach cr generate males::name as mname, fems::name as fname , males::age as mage,
>>  fems::age as fage;
grunt> 
grunt> describe mf
mf: {mname: chararray,fname: chararray,mage: int,fage: int}
grunt> mlist = filter mf by                
>>   (mage>fage  and (mage-fage)<4);

grunt> dump mlist;
(madhu,srithi,22,21)
(avi,Rani,26,24)
(avi,Ilean,26,23)
(Kiran,trisha,29,27)
(Ravi,Rani,25,24)
(Ravi,Ilean,25,23)
_________________________________

to submit scripts

3 commands:
 i) Pig
 ii) exec
 iii) run 

 pig to submit from command prompt.
  aliases will not be available in grunt.

 exec- to submit script from grunt shell. aliases will not be available.

 run- to submit script from grunt,
 aliases will be available.
 so that we reuse them.

[training@localhost ~]$ cat script1.pig
emp = load 'pdemo/emp' using PigStorage(',')
    as (id:int, name:chararray, sal:int, sex:chararray, dno:int);
e = foreach emp generate sex, sal;
bySex = group e by sex;
res = foreach bySex generate group as sex, SUM(e.sal) as tot;
dump res

$ pig script1.pig

grunt> exec script1.pig

grunt> run script1.pig

______________________________

register, define.






_____________________




___________________________________

Pig Joins

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

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

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

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

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

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

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

Pig Lab5

grunt> ls pdemo/e1
hdfs://localhost/user/training/pdemo/e1<r 1>    39
grunt> cat pdemo/e1
ravi,30000,m
rani,40000,f
giri,50000,m
grunt> cat pdemo/e2
hari,60000,11
hara,70000,12
grunt> e1 = load 'pdemo/e1' using PigStorage(',')
>>   as (name:chararray, sal:int, sex:chararray);
grunt> cat pdemo/e2
hari,60000,11
hara,70000,12
grunt> e2 = load 'pdemo/e2' using PigStorage(',')
>>   as (name:chararray, sal:int ,
>>    dno:int);
grunt> ee1 = foreach e1 generate *,0 as dno;
grunt> describe ee1;
ee1: {name: chararray,sal: int,sex: chararray,dno: int}
grunt> 
grunt> ee2 = foreach e2 generate name,sal,
>>        '*' as sex, dno;
grunt> describe ee2
ee2: {name: chararray,sal: int,sex: chararray,dno: int}
grunt> e = union ee1, ee2;
grunt> dump e;

(ravi,30000,m,0)
(rani,40000,f,0)
(giri,50000,m,0)
(hari,60000,*,11)
(hara,70000,*,12)
________________________________________
sql-->
  select dno , sum(sal) as tot from emp
     group by dno
    order by tot desc
    limit 3;

   ______________________________
   emp = load 'pdemo/emp' using PigStorage(',')
    as (id:int, name:chararray, sal:int, 
     sex:chararray, dno:int);

   e = foreach emp generate dno, sal;
   grp = group e by dno;
   res = foreach grp generate 
       group as dno, SUM(e.sal) as tot;
   ores = order res by tot desc;
   top3 = limit ores 3;

     above model is correct , if each dno total salary is unique.
   if total sal is taken by multiple dno s.
  you get wrong results.

  solution:  joins/ udfs. 

 ex:
      15   9L
      18   9L
      25   9L
      30   8L
      40   7L

    top3;
      15   9L
      18   9L
      25   9L
    2nd top, 3rd top were missed.
_________________________________________

Joins examples;


[training@localhost ~]$ cat emp
101,vino,26000,m,11
102,Sri,25000,f,11
103,mohan,13000,m,13
104,lokitha,8000,f,12
105,naga,6000,m,13
101,janaki,10000,f,12
201,aaa,30000,m,12
202,bbbb,50000,f,13
203,ccc,10000,f,13
204,ddddd,50000,m,13
304,xxx,70000,m,14
305,yyy,80000,f,15
[training@localhost ~]$ 

[training@localhost ~]$ cat > dept
11,marketing,hyd
12,hr,del
13,finance,hyd
20,prod,hyd
21,admin,chennai
[training@localhost ~]$ 

[training@localhost ~]$ hadoop fs -copyFromLocal emp pdemo/empl
[training@localhost ~]$ hadoop fs -copyFromLocal dept  pdemo
[training@localhost ~]$ 



 emp = load ' ... '   ----
 dept = load ' ... '  ------

 ij = join emp by dno, dept by dno;
 lj = join emp by dno left outer, dept by dno;
 rj = join emp by dno right outer, dept by dno;
 fj = join emp by dno full outer, dept by dno;

 dump fj;
(101,vino,26000,m,11,11,marketing,hyd)
(102,Sri,25000,f,11,11,marketing,hyd)
(104,lokitha,8000,f,12,12,hr,del)
(101,janaki,10000,f,12,12,hr,del)
(201,aaa,30000,m,12,12,hr,del)
(103,mohan,13000,m,13,13,finance,hyd)
(105,naga,6000,m,13,13,finance,hyd)
(202,bbbb,50000,f,13,13,finance,hyd)
(203,ccc,10000,f,13,13,finance,hyd)
(204,ddddd,50000,m,13,13,finance,hyd)
(304,xxx,70000,m,14,,,)
(305,yyy,80000,f,15,,,)
(,,,,,20,prod,hyd)
(,,,,,21,admin,chennai)

grunt> describe fj
fj: {emp::id: int,emp::name: chararray,emp::sal: int,emp::sex: chararray,emp::dno: int,dept::dno: int,dept::dname: chararray,dept::loc: chararray}
grunt> fj2 = foreach fj generate 
>>         emp::dno as dno1 ,
>>        dept::dno as dno2,
>>         emp::sal as sal;
grunt> describe fj2;
fj2: {dno1: int,dno2: int,sal: int}
grunt> 

(11,11,26000)
(11,11,25000)
(12,12,8000)
(12,12,10000)
(12,12,30000)
(13,13,13000)
(13,13,6000)
(13,13,50000)
(13,13,10000)
(13,13,50000)
(14,,70000)
(15,,80000)
(,20,)
(,21,)

grunt> fj3 = foreach fj2 generate 
>>      (dno1 is not null and dno2 is not null ?    'Working':(dno1 is not null and dno2 is null ?     'BenchTeam':'BenchProj')) as stat, sal;
grunt> describe fj3;
fj3: {stat: chararray,sal: int}
grunt> dump fj3

(Working,26000)
(Working,25000)
(Working,8000)
(Working,10000)
(Working,30000)
(Working,13000)
(Working,6000)
(Working,50000)
(Working,10000)
(Working,50000)
(BenchTeam,70000)
(BenchTeam,80000)
(BenchProj,)
(BenchProj,)

grunt> describe fj3
fj3: {stat: chararray,sal: int}
grunt> grp = group fj3 by stat;
grunt> res = foreach grp generate
>>    group as stat , SUM(fj3.sal) as tot;
grunt> dump res

(Working,228000)
(BenchProj,)
(BenchTeam,150000)
____________________________________________














   

Pig Lab4

cogroup:-
_________

to get seperate inner bags (data groups) for each dataset.
 so that, we can perform seperate aggregations on each data set.

  ds1
_______
 (a,10)
 (b,20)
 (a,30)
 (b,40)
_________

  ds2
_________
 (a,30)
 (c,30)
 (c,40)
 (a,20)
_________

cg = cogroup ds1 by $0, ds2 by $0;
 --> group, ds1, ds2
(a,{(a,10),(a,30)},{(a,30),(a,20)})
(b,{(b,20),(b,40)},{})
(c,{},{(c,30),(c,40)})

res = foreach cg generate
      group as mykey, 
  COUNT(ds1) as cnt1,
  COUNT(ds2) as cnt2;
(a,2,2)
(b,2,0)
(c,0,2)
_________________________________

[training@localhost ~]$ cat > sales1
p1,2000
p2,3000
p1,4000
p1,5000
p2,4000
p3,5000
[training@localhost ~]$ cat > sales2
p1,6000
p2,8000
p1,1000
p1,5000
p1,6000
p2,6000
p2,8000
[training@localhost ~]$ hadoop fs -copyFromLocal sales1  pdemo


[training@localhost ~]$ hadoop fs -copyFromLocal sales2  pdemo

grunt> s1 = load 'pdemo/sales1'                
>>     using PigStorage(',')
>>    as (pid:chararray, price:int);
grunt> s2 = load 'pdemo/sales2'         
>>     using PigStorage(',')        
>>    as (pid:chararray, price:int);
grunt> cg = cogroup s1 by pid, s2 by pid;
grunt> describe cg
cg: {group: chararray,s1: {pid: chararray,price: int},s2: {pid: chararray,price: int}}
grunt> dump cg

(p1,{(p1,2000),(p1,4000),(p1,5000)},{(p1,6000),(p1,1000),(p1,5000),(p1,6000)})
(p2,{(p2,3000),(p2,4000)},{(p2,8000),(p2,6000),(p2,8000)})
(p3,{(p3,5000)},{})

grunt> res = foreach cg generate
>>    group as pid, SUM(s1.price) as tot1,
>>    SUM(s2.price) as tot2;
grunt> dump res

(p1,11000,18000)
(p2,7000,22000)
(p3,5000,)

___________________________________

cleaning nulls

grunt> describe res;
res: {pid: chararray,tot1: long,tot2: long}.

grunt> res = foreach res generate             
>>    pid,                                
>>    (tot1 is null   ?   0:tot1) as tot1,
>>   (tot2 is null ? 0:tot2) as tot2;     
grunt> 


(p1,11000,18000)
(p2,7000,22000)
(p3,5000,0)

    
grunt> res = foreach res generate
>>      *,  tot1+tot2 as totall;
grunt> dump res

(p1,11000,18000,29000)
(p2,7000,22000,29000)
(p3,5000,0,5000)

___________________________________

assignment:-
______________

task: cleaning 
schema -->
  trid, prid, price, mrp, qnt, discount

if  price missed, replace it by mrp.
if qnt missed , replace it by 1.
if discount missed , replace it by 0.

 sankara.deva2016@gmail.com
_______________________________

union:-
______

 ds1
________
name, sal
__________
aaa,10000
bbbb,30000
cccc,40000
___________

ds2
____________
name,sal
___________
 xxx,30000
 yyy,40000
 zzz,60000
______________

 e = union ds1  , ds2



if all files schema is different.
 ds3
_________
sal, name
___________
10000,abc
20000,def
____________

ds4 = foreach ds3 generate name, sal;

ds = union ds1, ds2, ds4;
___________________________

if files have different number of fields.

[training@localhost ~]$ cat > e1
ravi,30000,m
rani,40000,f
giri,50000,m
[training@localhost ~]$ cat > e2
hari,60000,11
hara,70000,12
[training@localhost ~]$ hadoop fs -copyFromLocal e1 pdemo
[training@localhost ~]$ hadoop fs -copyFromLocal e2 pdemo

grunt> e1 = load 'pdemo/e1' using PigStorage(',')
>>   as (name:chararray, sal:int, sex:chararray);
grunt> e2 = load 'pdemo/e2' using PigStorage(',')
>>   as (name:chararray, sal:int, dno:int);
grunt> ee1 = foreach e1 generate *,0 as dno;
grunt> ee2 = foreach e2 generate name,
>>         sal,'*' as sex, dno;
grunt> ee = union ee1, ee2;
grunt> 

________________________________________

distinct:- to eliminate duplicates

    ds
  _id,name_________
  (101,a)
  (102,b)
  (103,c)
  (101,a)
  (102,b)
  (101,a)
  (102,b)
  (102,x)
_____________

  ds2 = distinct ds;
  (101,a)
  (102,b)
  (103,c)
  (102,x)
___________________________

emp ---> id,name,sal,sex,dno

 what are different dno(departments)s

 e = foreach emp geneate dno;
 e2 = distinct e

 how many unique depts?

  grp = group e2 all;
 res = foreach grp generate
 COUNT(e2) as cnt; 

_____________________________

order:
______

 e = order emp by name;
 e2 = order emp by sal desc;
 e3 = order emp by sal desc,
              dno , sex desc;

--- no limit for max number of sort fields.

________________________________