DataLake Architecture

Steps to install Apache Oozie 5.0

Prerequisite

Java JDK 1.8+

Maven 3.0.1+

Hadoop 2.6.0+

Pig 0.10.1+ 

Step1: Download Oozie 5.x  

Download oozie-5.0.0.tar.gz form below link 

http://www-eu.apache.org/dist/oozie/

Alternatively you can use wget command in OSX and or curl in linux respectively.

wget http://www-eu.apache.org/dist/oozie/5.0.0/oozie-5.0.0.tar.gz

curl -O http://www-eu.apache.org/dist/oozie/5.0.0/oozie-5.0.0.tar.gz

Step 2: Untar & expand the source distribution

tar xvzf oozie-3.3.2.tar.gz

Step 3: Build Oozie

Go inside Oozie directory and run below command. 

cd oozie-5.0.0

bin/mkdistro.sh  -Dhadoop.version=2.7.5 -Dpig.version=0.14.0       -Djetty.version=9.4.11.v20180605 –DskipTests

Note that my installed hadoop version is 2.7.5. Provide your respective hadoop & pig version

 

Step 4: Update Hadoop core-site.xml  and restart Hadoop

<property>     <name>hadoop.proxyuser.[OOZIE_SERVER_USER].hosts</name>     <value>[OOZIE_SERVER_HOSTNAME]</value>   </property>   

<property>     <name>hadoop.proxyuser.[OOZIE_SERVER_USER].groups</name>     <value>[USER_GROUPS_THAT_ALLOW_IMPERSONATION]</value>   

</property> 

My user name is  ‘hadoop’ so in my case it is

  <property>

<name>hadoop.proxyuser.hadoop.hosts</name>

<value>*</value>

</property>

<property>

<name>hadoop.proxyuser.hadoop.groups</name>

<value>*</value>

</property>

Step 5: Setup Oozie Server

Oozie 5.0 uses jetty server, previous versions were using tomcat. Instead of bootstrap  now it invokes EmbeddedOozieServe.

Before preparing war, extjs and Hadoop jars should be available in libext directory  

cd distro/target/oozie-5.0.0-distro/oozie-5.0.0

mkdir libext

Download copy ext js from below link to libext folder

https://drive.google.com/file/d/0ByJLBTmJojjzcDhxQUsyNEFSQm8/view

Copy hadoop  and hcatalog libraries into libext

You can use below command to copy.

Before executing just update bash profile with HADOOP_HOME and  $OOZIE_HOME values

cp $HADOOP_HOME/share/hadoop/common/*.jar $OOZIE_HOME/libext/

cp $HADOOP_HOME/share/hadoop/common/lib/*.jar $OOZIE_HOME/libext/

cp $HADOOP_HOME/share/hadoop/mapreduce/*.jar $OOZIE_HOME/libext/

cp $HADOOP_HOME/share/hadoop/mapreduce/lib/*.jar $OOZIE_HOME/libext/

cp $HADOOP_HOME/share/hadoop/hdfs/*.jar $OOZIE_HOME/libext/

cp $HADOOP_HOME/share/hadoop/hdfs/lib/*.jar $OOZIE_HOME/libext/

cp $HADOOP_HOME/share/hadoop/yarn/*.jar $OOZIE_HOME/libext/

cp $HADOOP_HOME/share/hadoop/yarn/lib/*.jar $OOZIE_HOME/libext

Now run oozie-setup.sh

bin/oozie-setup.sh

Step 6: Create share lib

 Delete any previous share library in HDFS if exist.

hadoop dfs -rmr /user/hadoop/share

Create share lib

bin/oozie-setup.sh  sharelib create -fs hdfs://localhost:9000 -locallib oozie-sharelib-5.0.0.tar.gz

Step 7: Create Oozie DB

 bin/ooziedb.sh create -sqlfile oozie.sql –run

Step 8: Start Oozie as a daemon process:

bin/oozied.sh start

URL for the Oozie Web Console is http://localhost:11000/oozie

Data Automation System Architecture

Cascading .. just go for it

After an year  of some rigorous MapReduce coding and customizing  a lot of hadoop classes,   InputFormat, RecordReader, Partitioner, Comparator, Writables  etc, we heard about Cascading,  found it intriguing and thought of give it a try. Initially it took a bit time specially when a have thought of every thing as a key value pair for long and now you have to  think in in terms of taps, sinks pipes and flow.

After a while, we were amazed by cascading in built support and decided to use it in production.   Our Code base is much shorter and simpler now. Now no more customization of classes. Application  building & testing time decreased significantly. 
 
So Dear MapReduce programmer, what are you waiting for .. time to switch to cascading.

http://www.cascading.org/
http://www.cascading.org/documentation/

http://hortonworks.com/blog/cascading-hadoop-big-data-whatever/


:) Happy Cascading

Combine file Input format in MapReduce

Hadoop is good for processing small number of large files rather large numbers of small files.  It splits large input files into so called input splits and each part is process parallely by different mapper. For example inside hadoop cluster of 256MB block size, a 1GB input file  files is split into 4 parts and 4 mapper processes them parallaly.

What if files are small say 10 MB, in this case each file is processed using separate mapper resulting into large number of mapper hence makes hadoop underutilized.

Solution is to combine small size files as input to single mapper  by using combine input format.  Combine input format should be custom implemented.

Combine input format use  simple logic that  "The key to map function is combination  of filename and byteoffset" .

A custom writable  FileLineWritable  is used for this  as shown below 

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

//two varible as key filename, and byteoffset 

public class FileLineWritable implements WritableComparable<FileLineWritable>
{

    public long offset;
    public String fileName;

    public void readFields(DataInput in) throws IOException
    {
        this.offset = in.readLong();
        this.fileName = Text.readString(in);
    }

    public void write(DataOutput out) throws IOException
    {
        out.writeLong(offset);
        Text.writeString(out, fileName);
    }

    //Compare file name first then offset
   
    public int compareTo(FileLineWritable that)
    {
        int cmp = this.fileName.compareTo(that.fileName);
        if (cmp != 0)
        {
            return cmp;
        }
        return (int) Math.signum((double) (this.offset - that.offset));
    }

   
    @Override
    public int hashCode()
    {               // generated hashCode()
        final int prime = 31;
        int result = 1;
        result = prime * result + ((fileName == null) ? 0 : fileName.hashCode());
        result = prime * result + (int) (offset ^ (offset >>> 32));
        return result;
    }

    @Override
    public boolean equals(Object obj)
    {  // generated equals()
        if (this == obj)
        {
            return true;
        }
        if (obj == null)
        {
            return false;
        }
        if (getClass() != obj.getClass())
        {
            return false;
        }
        FileLineWritable other = (FileLineWritable) obj;
        if (fileName == null)
        {
            if (other.fileName != null)
            {
                return false;
            }
        }
        else if (!fileName.equals(other.fileName))
        {
            return false;
        }
        if (offset != other.offset)
        {
            return false;
        }
        return true;
    }
}




Now customized combine file input format needs to be implemented which uses  FileLineWritable  (as key) and extends CombineFileInputFormat  as shown  below-




public class CFInputFormat extends CombineFileInputFormat<FileLineWritable, Text>
{

    public CFInputFormat()
    {
        super();
        setMaxSplitSize(67108864); // 64 MB, default block size on hadoop
    }

    public RecordReader<FileLineWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context)
            throws IOException
    {
        return new CombineFileRecordReader<FileLineWritable, Text>((CombineFileSplit) split, context, CFRecordReader.class);
    }

    @Override
    protected boolean isSplitable(JobContext context, Path file)
    {
        return false;
    }
}



Also Note that "isSplitable" value is false to prevent input files to splits further, by default its true.


Next is to implement custom record reader to read record line by line.

import java.io.IOException;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.util.LineReader;

public class CFRecordReader extends RecordReader<FileLineWritable, Text>
{

    private long startOffset;
    private long end;
    private long pos;
    private FileSystem fs;
    private Path path;
    private FileLineWritable key;
    private Text value;

    private FSDataInputStream fileIn;
    private LineReader reader;

    public CFRecordReader(CombineFileSplit split, TaskAttemptContext context, Integer index)
            throws IOException
    {
        this.path = split.getPath(index);
        fs = this.path.getFileSystem(context.getConfiguration());
        this.startOffset = split.getOffset(index);
        this.end = startOffset + split.getLength(index);

        fileIn = fs.open(path);
        reader = new LineReader(fileIn);
        this.pos = startOffset;
    }

    @Override
    public void initialize(InputSplit arg0, TaskAttemptContext arg1)
            throws IOException,
            InterruptedException
    {
  // Won't be called, use custom Constructor
        // `CFRecordReader(CombineFileSplit split, TaskAttemptContext context, Integer index)`
        // instead
    }

    @Override
    public void close() throws IOException
    {
    }

    @Override
    public float getProgress() throws IOException
    {
        if (startOffset == end)
        {
            return 0;
        }
        return Math.min(1.0f, (pos - startOffset) / (float) (end - startOffset));
    }

    @Override
    public FileLineWritable getCurrentKey() throws IOException,
            InterruptedException
    {
        return key;
    }

    @Override
    public Text getCurrentValue() throws IOException,
            InterruptedException
    {
        return value;
    }

    @Override
    public boolean nextKeyValue() throws IOException
    {
        if (key == null)
        {
            key = new FileLineWritable();
            key.fileName = path.getName();
        }
        key.offset = pos;
        if (value == null)
        {
            value = new Text();
        }
        int newSize = 0;
        if (pos < end)
        {
            newSize = reader.readLine(value);
            pos += newSize;
        }
        if (newSize == 0)
        {
            key = null;
            value = null;
            return false;
        }
        else
        {
            return true;
        }
    }
}




 The complete code can be downloaded from my Github repo for MaxTemprature problem.



:) learning goes on forever





 

Avro to XML Convertor

 Couldn't find any Avro to XML convertor so created one.

The  software can be downloaded from here :

Avro-XML convertor

Here's the release notes   :

Avro-XML convertor Release note

Build from source code & contribute :

Githhub _ Avro-XML Convertor

Pig script for finding max temprature

  Finding the max temperature using Pig script

The input data set can be obtain here :
   https://drive.google.com/file/d/0BwiqVGNpnBVIbDZ6Q1V1RThxYXc/edit?usp=sharing

My Hadoop path is  :  /usr/local/hadoop/
Hadoop user is        :  /home/hduser



Steps :

• Start  Hadoop   hduser@kaustuv-studio14:/home/kaustuv$ /usr/local/hadoop/bin/start-all.sh 

•  Copy & ensure input exist in HDFS   (use -copyFromLocal command )

                hduser@kaustuv-studio14:/usr/local/hadoop$ bin/hadoop dfs -ls /home/hduser/

                           ( This will list weather.txt  file )

•  Start Pig grunt shell in MapReduce mode   hduser@kaustuv-studio14:/home/kaustuv$ pig

•   Write the following Max temp pig script  

A = load '/home/hduser/weather.txt' AS (f1: chararray);
B = foreach A generate SUBSTRING(f1, 4, 8) AS (year: chararray), SUBSTRING(f1, 38,43) AS (temp: chararray) ;
C = group B by $0;
Max_temp = foreach C generate group,
MAX(B.temp);
store Max_temp INTO 'MAX_Temp_Output' ;

Internally pig script is converted  into MapReduce program we can check the progress of this MR program via  web interfaces of namenode & job tracker  also.

 Output will be stored under  MAX_Temp_Output folder inside users home directory here  '/user/hduser'.

• Output can be verified  using 'cat' command 

hduser@kaustuv-studio14:/usr/local/hadoop$ bin/hadoop dfs -cat /user/hduser/MAX_Temp_New_Output/part-r-00000

This will list 

Warning: $HADOOP_HOME is deprecated.

1941    106.2
1942    183.9
1943    176.7
1944    156.2
1945    130.6
1946    152.3
1947    191.1
1948    175.9
1949    181.1
1950    208.8
1951    168.8
1952    122.6
1953    126.5
1954    232.3
1955    130.2
1956    114.6
1957    187.7
1958    184.5
1959    229.9
1960    204.7
1961    173.8
1962    130.8
1963    187.9
1964    144.3
1965    186.1
1966    155.9
1967    173.8
1968     93.8
1969    146.4
1970    181.4
1971    136.4
1972    128.5
1973    119.9
1974    203.2
1975    132.3
1976    157.8
1977    150.6
1978    140.0
1979    158.9
1980    119.3
1981    217.2
1982    141.2
1983    122.1
1984    154.2
1985    146.0
1986    187.9
1987    219.2
1988    164.0
1989    120.2
1990    118.0
1991    142.4
1992    149.3
1993    190.6
1994    157.4
1995    145.6
1996    107.2
1997    219.0
1998    125.2
1999    143.0
2000    195.0
2001    147.4
2002    180.2
2003    111.4
2004    168.8
2005    194.4
2006    153.8
2007    155.2
2008    140.0
 


 

Steps to install Pig in Ubuntu

1. Download tar file from this link :
( Do check that pig version support installed Hadoop version)
e.g. My ubuntu  has Hadoop 1.2.1 installed and  have installed pig 0.12.1
Hadoop installed under : /usr/local/hadoop
hduser  (hadoop group user) is under : /home/hduser

2. Open terminal and logged in as hduser
                su hduser

3. Unpack the downloaded tar file using command
                tar xzf  pig-0.12.1.tar.gz
    This will install pig under /home/hduser/

4. Edit bash file 
               gedit ~/.bashrc

Append following line   
export PIG_HOME=/home/hduser/pig-0.12.1 
export PATH=$PATH:$PIG_HOME/bin
export PIG_CLASSPATH=/usr/local/hadoop/conf

5.  In terminal type
               pig 
This will open grunt (interactive shell for running pig commands)  in Mapreduce mode.
(*Before starting pig make sure you have started Hadoop
 /usr/local/hadoop$  bin/start-all.sh    )



 Learning goes on forever :)
 

MapReduce MaxTemprature program execution in standalone mode using Eclipse

 

MapReduce program execution in standalone mode ( Eclipse )  

Finding the maximum temperature of each year

A Sample weather data row shown below :

1st Column is year and sixth is temperature (Fahrenheit )

+   1942   1    5.8     2.1    ---    114.0    58.0 

+   1942   5   14.0     6.9    ---    101.1   215.1

  MapReduce program consist of  :

• Map Class  : Map Function
• Reduce Class : Reduce Function
• Driver Class  : Main Method

Map Function  :

Takes text file as input (TextInputFormat ),
for each line of input file it emits year and temperature as output key/ value pair

I/P key --> Byteoffset ( type : LongWritable)
I/p value  --> Line  ( type : Text)
o/p Key  -> year  ( type : text)
o/p value -> temperature ( type : float)

 Map Class :
public static class Map extends Mapper<LongWritable, Text, Text, FloatWritable>
    {
         public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
        {
            String line = value.toString();
            String year = line.substring(4,9);
            float temp = Float.parseFloat(line.substring(38,43));
         
                context.write(new Text(year), new FloatWritable(temp));
          }
     
    }


Reduce function  :
Takes Map output as Input and emits maximum temperature of  every year as temp/year (as key /value) pair

I/p  key -> year  ( type : text)
I/p value ->  Temprature list  ( type :  Float )
o/p Key  -> year  ( type : text)
o/p value -> temperature ( type : float)



Reduce Class:

public static class Reduce extends Reducer<Text, FloatWritable, Text, FloatWritable>
    {
        public void reduce(Text key, Iterable<FloatWritable> values, Context context) throws IOException, InterruptedException
        {
            float maxValue = Float.MIN_VALUE ;
            for (FloatWritable value : values)
            {
                maxValue = Math.max(maxValue,value.get());
            }
            context.write(key, new FloatWritable(maxValue) );
       }
    } 
 

Driver ( Main method) :

       public static void main(String[] args) throws Exception
        {
        Configuration conf = new Configuration();
          Job job = new Job(conf,"MaxTemp");
       
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FloatWritable.class);

        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.setJarByClass(MaxTemp.class);
        job.waitForCompletion(true);
    }  

Execution Steps in Eclipse :

Step 1 :
Open Eclipse : File -> New ->  Java Project

Enter project name : MaxTemp

Press Finish

Step 2 :

Under Package Explorer window

Right click on MaxTemp   New->Class

Enter  under Name  MaxTemp and check Public static void main ()

Finish

Step 3 :
Paste the below programe code under MaxTemp.java

import java.io.IOException;
import java.util.*;
import java.lang.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;


public class MaxTemp
{

    public static class Map extends Mapper<LongWritable, Text, Text, FloatWritable>
    {
         public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
        {
            String line = value.toString();
            String year = line.substring(4,9);
            float temp = Float.parseFloat(line.substring(38,43));
         
                context.write(new Text(year), new FloatWritable(temp));
          }
     
    }

    public static class Reduce extends Reducer<Text, FloatWritable, Text, FloatWritable>
    {
        public void reduce(Text key, Iterable<FloatWritable> values, Context context) throws IOException, InterruptedException
        {
            float maxValue = Float.MIN_VALUE ;
            for (FloatWritable value : values)
            {
                maxValue = Math.max(maxValue,value.get());
            }
            context.write(key, new FloatWritable(maxValue) );
       }
    } 

   
        public static void main(String[] args) throws Exception
        {
        Configuration conf = new Configuration();

        Job job = new Job(conf,"MaxTemp");
       
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FloatWritable.class);

        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.setJarByClass(MaxTemp.class);
        job.waitForCompletion(true);
    }      
}




Step 4 :
Package Explorer --> MaxTemp-->Build Path --> Configure Build Path --> Java Build Path --> libraries -> Add Library

Add following libraries and press OK
 


Step5 :
Under Package Explorer
 MaxTemp -> Run As -> Run Configuration

Under main tab edit
Name : MaxTemp
Project:  MaxTemp
Main Class : MaxTemp

Under argument tab edit
Program Arguments :
input output

Apply ->Close


Step 6 :
Create new folder as input under your MaxTemp directory in your workspace folder  ( my case its home/workspace/MaxTemp)

Step 7 :
Copy the sample data ( provided below) and store as a text file under input folder.

Step 8 :
Run the program and check the result under output folder( ../MaxTemp/output).

Sample weather data can be download from the link below :

https://drive.google.com/file/d/0BwiqVGNpnBVIbDZ6Q1V1RThxYXc/edit?usp=sharing


The project  can be obtain from my github dir :
https://github.com/kaustuvkunal/Bigdata/tree/master/MaxTemp


Happy Coding..