Language neutralization

pig lab9 : Language neutralization
Laguage neutralization.

ex: hindi to english.

[training@localhost ~]$ gedit comments
[training@localhost ~]$ gedit dictionary
[training@localhost ~]$ hadoop fs -copyFromLocal comments pdemo
[training@localhost ~]$ hadoop fs -copyFromLocal dictionary pdemo
[training@localhost ~]$

grunt> comms = load 'pdemo/comments'
>>     as (line:chararray);
grunt> dict = load 'pdemo/dictionary'
>>     using PigStorage(',')
>>    as (hindi:chararray, eng:chararray);
grunt> words = foreach comms
>>    generate FLATTEN(TOKENIZE(line)) as word;
grunt>

grunt> describe words
words: {word: chararray}
grunt> describe dict
dict: {hindi: chararray,eng: chararray}
grunt> j = join words by word left outer,
>>    dict by hindi;
grunt> jj = foreach j generate words::word as word, dict::hindi as hindi, dict::eng as eng;
grunt> dump jj
(is,,)
(is,,)
(is,,)
(xyz,,)
(acha,acha,good)
(bacha,bacha,small)
(lucha,lucha,worst)
(hadoop,,)
(oracle,,)
____________________

grunt> rset = foreach jj generate
>>    (hindi is not null ? eng:word) as word;
grunt> dump rset

(is)
(is)
(is)
(xyz)
(good)
(small)
(worst)
(hadoop)
(oracle)

__________________________

Then we can apply sentiment on this restult set.

__________________________-

Quartely sales

Pig Lab10 : Quarterly sales Report and comparing sales with Previous Quarter. Cross .
grunt> raw = load 'piglab/sales'
>>     using PigStorage(',')
>>     as (dt:chararray, amt:int);
grunt> msales = foreach raw generate 
>>      SUBSTRING(dt,0,2)  as m, amt ;
grunt> describe msales;
msales: {m: chararray,amt: int}
grunt> msales = foreach msales generate
>>         (int)m, amt;
grunt> describe msales;               
msales: {m: int,amt: int}
grunt> dump msales;

grunt> qsales = foreach msales generate
>>       (m<4 ? 1:
>>         (m<7 ? 2:
>>           (m<10 ? 3:4))) as q, amt ;
grunt> dump qsales

grunt> qsales = foreach msales generate
>>       (m<4 ? 1:
>>         (m<7 ? 2:
>>           (m<10 ? 3:4))) as q, amt ;
grunt> dump qsales

grunt> rep2 = foreach rep generate *;
grunt> cr = cross rep, rep2;
grunt> describe cr
cr: {rep::q: int,rep::tot: long,rep2::q: int,rep2::tot: long}
grunt> cr = foreach cr generate
>>     rep::q as q1, rep2::q as q2,
>>     rep::tot as tot1, rep2::tot as tot2;

grunt> describe cr;
cr: {q1: int,q2: int,tot1: long,tot2: long}
grunt> dump cr;

grunt> fltd = filter cr by ((q1-q2)==1);
grunt> dump fltd;

grunt> res = foreach fltd generate *,
>>        ((tot1-tot2)*100)/tot2 as pgrowth;
grunt> dump res

(3,2,70000,355000,-80)
(2,1,355000,140000,153)
(4,3,290000,70000,314)

_________________________________

Executing scripts

Pig Lab 11: Executing Scripts and UDFs.

Excecuting scripts(pig) :-
____________________________

   three commands (operators) are used to execute scripts.

  i) pig
  ii) exec. 
  iii) run.

i) Pig:-
 
   to execute script from Command Line(operating sys).

  $ pig  script1.pig
--> script will be executed,
   but relation aliases are not available with grunt shell. so that we  can not reuse them.

ii) exec:

--> to execute script from grunt shell.
still aliases will not be available with grunt.
   so "No reuse".

grunt> exec  script1.pig

__________________________________

iii) run:

---> to execute script from grunt shell,
   Aliases will be available with grunt. So we can reuse them.

grunt> run script1.pig
______________________________

run:
  adv --> aliases will be available.
  disadv --> overriding previous aliases with same name.

exec:
   adv --> aliases will not be available.
    so no -0verriding.
  disadv --> no reusability.

pig :
  adv --->
      -- used for production operators.
     --- can be called other evenvironments , like shell script.

disadv --> aliases will not be reflected into grunt.

_________________________________

Pig Udfs:
_____________j

   User defined functions.

   adv:
   i) Custom functionality.
   ii) Reusabilty .

Udf life cycle:

step 1) Develop UDF class
step 2) Export into jar file.
step 3) register jar file into pig.
step 4) create temporary function for  the  UDF class.
step 5) call the function.
__________________________

[training@localhost ~]$ cat > samp1
101,ravi
102,mani
103,Deva
104,Devi
105,AmAr
[training@localhost ~]$ hadoop fs -copyFromLocal samp1  piglab
[training@localhost ~]$

grunt> s = load 'piglab/samp1'
>>    using PigStorage(',')
>>   as (id:int, name:chararray);

eclipse navigations:
i) create java project.

file --> new --> java project.
  ex: PigDemo

ii) create package>

  pigDemo ---> new ---> package.

    ex:    pig.test

  iii) configure pig jar.

src -- build path --> configure build path --> libraries ---> add external jars.
 
  /usr/lib/pig/pig-core.jar

  iv) create jar class

pig.test ---> new --> class

   FirstUpper

  v)  export into jar.

pigdemo ---> export --> java --java jar --
    /home/training/Desktop/pigudfs.jar

_____________________
package pig.test;
import java.io.IOException;

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

public class FirstUpper  extends EvalFunc<String>
{
  public String   exec(Tuple v) throws IOException
  {    //   raVI  --> Ravi

     String name = (String)v.get(0);
   String fc = name.substring(0,1).toUpperCase();
   String rc = name.substring(1).toLowerCase();
   String n = fc+rc;
   return n;
  }

}

grunt> register  Desktop/pigudfs.jar;

grunt> define cconvert pig.test.FirstUpper();

grunt> r = foreach s generate
>>       id, cconvert(name) as name;

grunt> dump r;

______________________________

[training@localhost ~]$ cat > f1
100     200     120
300     450     780
120     56      90
1000    3456    789
[training@localhost ~]$ hadoop fs -copyFromLocal f1 piglab
[training@localhost ~]$

task:
  write udf , to find max value for a row.

package pig.test;

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; 
  if (a>big) big=a;
  if (b>big) big=b;
  if (c>big) big=c;
  return  new Integer(big);
}

}

export into jar.

    /home/training/Desktop/pigudfs.jar

grunt> s1 = load 'piglab/f1'
>>     as (a:int, b:int, c:int);
grunt> register Desktop/pigudfs.jar;
grunt> define rowmax pig.test.RowMax();
grunt> r1 = foreach s1 generate  *,
>>           rowmax(*) as rmax;
grunt> dump r1
(100,200,120,200)
(300,450,780,780)
(120,56,90,120)
(1000,3456,789,3456)

[training@localhost ~]$ cat f2
-10,-30,-56,-23,-21,-5
1,2,3,45,67,9
[training@localhost ~]$ hadoop fs -copyFromLocal f2 piglab
[training@localhost ~]$

package pig.test;

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

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

public class DynRowMax  extends EvalFunc<Integer>
{
public Integer  exec(Tuple v) throws IOException
{
  List<Object>  olist  =  v.getAll();
   int max = 0;//  10  30  20
   int cnt=0;
   for( Object o : olist){
       cnt++;
       int val= (Integer)o;
       if (cnt==1)  max = val;
                max = Math.max(val, max);
   }
  return new Integer(max);
}

}

export into jar   /home/training/Desktop/pigudfs.jar

grunt> register Desktop/pigudfs.jar;
grunt> define dynmax pig.test.DynRowMax();
grunt> ss = load 'piglab/f2'   
>>    using PigStorage(',')
>>   as (a:int, b:int, c:int, d:int,
>>     e:int, f:int);
grunt> define  rmax pig.test.RowMax();
grunt> rr = foreach ss generate *,
>>      dynmax(*) as max;
grunt> dump rr

Flame lab

Flume Lab
********************************
sreeram hadoop notes
sreeram flume notes for practice

conf/agent1.conf
_______________________

# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = src1
a1.sinks = s1
a1.channels = c1

# Describe/configure source1
a1.sources.src1.type = exec
a1.sources.src1.shell = /bin/bash -c
a1.sources.src1.command=cat f1

# Describe sin
a1.sinks.s1.type = hdfs
a1.sinks.s1.hdfs.path=hdfs://quickstart.cloudera/user/cloudera/myFlume

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1200
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.src1.channels = c1
a1.sinks.s1.channel = c1

############################################
submit above flow using following command

[cloudera@quickstart ~]$ hadoop fs -mkdir myFlume

[cloudera@quickstart ~]$ flume-ng agent --conf conf --conf-file conf/agent1.conf --name a1 -Dflume.root.logger=INFO,console

[cloudera@quickstart ~]$ hadoop fs -ls myFlume

note: By default sink will write in  sequence format

The Risk in above flow is, if channel failed or channel system down,
data will be missed.  to provide fault tolerence for channel use following flow.

************************************

sreeram hadoop notes:
sreeram flume notes for practice:

conf/agent2.conf
___________________

# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = src1
a1.sinks = s1 s2
a1.channels = c1 c2

# Describe/configure source1
a1.sources.src1.type = exec
a1.sources.src1.shell = /bin/bash -c
a1.sources.src1.command=cat f1

# Describe sin
a1.sinks.s1.type = hdfs
a1.sinks.s1.hdfs.path=hdfs://quickstart.cloudera/user/cloudera/urFlume

a1.sinks.s2.type = hdfs
a1.sinks.s2.hdfs.path=hdfs://quickstart.cloudera/user/cloudera/urFlume

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1200
a1.channels.c1.transactionCapacity = 100

a1.channels.c2.type = memory
a1.channels.c2.capacity = 1200
a1.channels.c2.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.src1.channels = c1 c2
a1.sinks.s1.channel = c1
a1.sinks.s2.channel = c2

##################################

[cloudera@quickstart ~]$ hadoop fs -mkdir urFlume
[cloudera@quickstart ~]$ flume-ng agent --conf conf --conf-file conf/agent2.conf --name a1 -Dflume.root.logger=INFO,console

[cloudera@quickstart ~]$ hadoop fs -ls urFlume

from above flow, src1 is writing into c1 and c2 channels,
  if one channel fails, still data available in another.
but if no failure happend data will be duplicated in target hadoop directory.
so before processing data, we need eliminated duplicate records.

*****************************************

sreeram hadoop notes:
sreeram flume notes for practice:
Task --> importing from Hive Table

conf/agent3.conf
______________________________
# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = src1
a1.sinks = s1
a1.channels = c1

# Describe/configure source1
a1.sources.src1.type = exec
a1.sources.src1.shell = /bin/bash -c
a1.sources.src1.command=hive -e 'select * from mraw'

# Describe sin
a1.sinks.s1.type = hdfs
a1.sinks.s1.hdfs.path=hdfs://quickstart.cloudera/user/cloudera/ourFlume

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1200
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.src1.channels = c1
a1.sinks.s1.channel = c1

#############################################

[cloudera@quickstart ~]$ hive
hive> create table mraw(line string);
OK
Time taken: 2.218 seconds
hive> load data local inpath 'f1' into table mraw;
hive> select * from mraw limit 5;
OK
aaaaaaaaaaaaaaaaaaaaaaaa
bbbbbbbbbbbbbbbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
cccccccccccccccccccccccccccccc
ddddddddddddddddddddddddddd
Time taken: 0.341 seconds, Fetched: 5 row(s)
hive>

[cloudera@quickstart ~]$ hadoop fs -mkdir ourFlume

[cloudera@quickstart ~]$ flume-ng agent --conf conf --conf-file conf/agent3.conf --name a1 -Dflume.root.logger=INFO,console

[cloudera@quickstart ~]$ hadoop fs -ls ourFlume

in all above cases , output will be in sequence file format.

******************************************

sreeram hadoop notes:
sreeram flume notes for practice:

conf/agent4.conf
________________________________

# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = src1
a1.sinks = s1
a1.channels = c1

# Describe/configure source1
a1.sources.src1.type = exec
a1.sources.src1.shell = /bin/bash -c
a1.sources.src1.command=cat f1

# Describe sin
a1.sinks.s1.type = hdfs
a1.sinks.s1.hdfs.fileType=DataStream
a1.sinks.s1.hdfs.writeFormat=Text
a1.sinks.s1.hdfs.path=hdfs://quickstart.cloudera/user/cloudera/naFlume

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1200
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.src1.channels = c1
a1.sinks.s1.channel = c1

######################################

above flow will write output in Text Format.

[cloudera@quickstart ~]$ hadoop fs -mkdir naFlume

[cloudera@quickstart ~]$ flume-ng agent --conf conf --conf-file conf/agent4.conf --name a1 -Dflume.root.logger=INFO,console

[cloudera@quickstart ~]$ hadoop fs -ls naFlume

[cloudera@quickstart ~]$ hadoop fs -cat naFlume/FlumeData.1471351131067
bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
cccccccccccccccccccccccccccccc
ddddddddddddddddddddddddddd
eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee
ddddddddddddddddddddddddddddd
ccccccccccccccccccccccccccccccc
cccccccccccccccccccccccccccccc
ccccccccccccccccccccccccccccccc

above output is in Text Format.

******************************************

Spark SQL and hql

Spark Sql and Hql
[cloudera@quickstart ~]$ sudo find / -name

'hive-site.xml'

[cloudera@quickstart ~]$ sudo chmod -R 777

/usr/lib/spark/conf
[cloudera@quickstart ~]$ cp

/etc/hive/conf.dist/hive-site.xml

/usr/lib/spark/conf
_____________________________________
from hive-site.xml --> 

hive.metastore.warehouse.dir

from spark 2.0.0  onwards above opt is

depricated
    use following option..

------>   spark.sql.warehouse.dir
_____________________________________________

____  [ tested in cloudera 5.8
          spark version 1.6.0 ]

[cloudera@quickstart ~]$ls

/usr/lib/hue/apps/beeswax/data/sample_07.csv

[cloudera@quickstart ~]$ head -n 2

/usr/lib/hue/apps/beeswax/data/sample_07.csv
_____________________
val hq  = new 

org.apache.spark.sql.hive.HiveContext(sc)

hq.sql("create database sparkdb")

  hq.sql("CREATE TABLE sample_07 (code 

string,description string,total_emp   

int,salary int) ROW FORMAT DELIMITED FIELDS 

TERMINATED BY '\t' STORED AS TextFile")

[cloudera@quickstart ~]$ hadoop fs -mkdir   

sparks

[cloudera@quickstart ~]$ hadoop fs - 

copyFromLocal 

/usr/lib/hue/apps/beeswax/data/sample_07.csv 

sparks
[cloudera@quickstart ~]$ hadoop fs -ls   

sparks

hq.sql("LOAD DATA INPATH 

'/user/cloudera/sparks/sample_07.csv' 

OVERWRITE INTO TABLE sample_07")

val df = hq.sql("SELECT * from sample_07")

__________________________________________
scala> df.filter(df("salary") > 150000).show

()
+-------+--------------------+---------

+------+
|   code|         description|total_emp|

salary|
+-------+--------------------+---------

+------+
|11-1011|    Chief executives|   299160|

151370|
|29-1022|Oral and maxillof...|     5040|

178440|
|29-1023|       Orthodontists|     5350|

185340|
|29-1024|     Prosthodontists|      380|

169360|
|29-1061|   Anesthesiologists|    31030|

192780|
|29-1062|Family and genera...|   113250|

153640|
|29-1063| Internists, general|    46260|

167270|
|29-1064|Obstetricians and...|    21340|

183600|
|29-1067|            Surgeons|    50260|

191410|
|29-1069|Physicians and su...|   237400|

155150|
+-------+--------------------+---------

+------+
____________________________________________

Flume

[cloudera@quickstart ~]$ cat flumeexample.conf
a1.sources= src1
a1.channels = c1
a1.sinks = s1

a1.sources.src1.type = exec
a1.sources.src1.command = cat f1.txt

a1.sinks.s1.type=hdfs
a1.sinks.s1.hdfs.path= hdfs://quickstart.cloudera/user1

a1.channels.c1.type = memory
a1.channels.c1.capacity =1000
a1.channels.c1.transactionCapacity =100

a1.sources.src1.channels = c1
a1.sinks.s1.channel = c1

[cloudera@quickstart ~]$ flume-ng agent     --conf conf      --conf-file flumeexample.conf      --name a1      -Dflume.root.logger=INFO,consoleflume-ng agent     --conf conf      --conf-file flumeexample.conf      --name a1      -Dflume.root.logger=INFO,console

Flume

[cloudera@quickstart ~]$ cat flumeexample.conf
a1.sources= src1
a1.channels = c1
a1.sinks = s1

a1.sources.src1.type = exec
a1.sources.src1.command = cat f1.txt

a1.sinks.s1.type=logger

a1.channels.c1.type = memory
a1.channels.c1.capacity =1000
a1.channels.c1.transactionCapacity =100

a1.sources.src1.channels = c1
a1.sinks.s1.channel = c1

[cloudera@quickstart ~]$ flume-ng agent     --conf conf      --conf-file flumeexample.conf      --name a1      -Dflume.root.logger=INFO,console