Image processing

Blurring:
A sharp point get smeared out into a disc shape is called blurring. In image terms, 
one pixel from the source image will spread over and mixed with the surrounding pixels. The operation need for this is called CONVOLUTION.
Working of convolution: imagine sliding a rectangular array of numbers over our image. This array is called the convolution kernel. For every pixel in the image, we take the corresponding numbers from the kernel and the pixels they are over, multiply them together and add all the results together to make the new pixel.

Basic idea
Using Sobel edge detection algorithm distributive image processing can be done. It has the property that the new pixel value is not only depending on the pixel’s original value, but also the value of the surrounding pixels. The image will be considered as a two dimensional array of pixels this image can be partitioned into subsets and operates in parallel by distributing subsets in various data tasks.

There are three steps:

• Track each pixel inside the image excluding the edges.
• Track the edges using the edge detection algorithm
• Divide various splits in the image and do the above steps as various map tasks.

Convolution at a single pixel:

Consider a image with pixels at the fourth quadrant of 2D space. 

Let its length as hin x and y directions.  In x direction the pixels will be from 0 to m-1 and in y direction also the pixels will be from 0 to n-1.  

Place a kernel above the image. Its kernel coefficient will also be same as that of image pixels.

Compute the kernel half width

     M2= floor[m/2]
     N2=floor[n/2]

     Let sum=0, vary the values from –n to +n and  -m to +m.
       for(int k=-m2;k<=+m2;k++)
       for(int j=-n2;j<=+n2;j++)
     compute the sum value that is the pixel value that is obtained by computing the neighbouring pixel values and x and y coordinate values.
       sum= sum + h(j + m2, k + n2) f(x - j, y - k);
                g(x,y)=sum;

Convolution of the image using edge detection algorithm:

Consider the image with pixels at the fourth quadrant .Let its length as hin x and y directions.  In x direction the pixels will be from 0 to m-1 and in y direction also the pixels will be from 0 to n-1.  

Place a kernel above the image. Associate the kernel coefficient with the image pixels.

Compute the kernel half width

    M2= floor[m/2]
    N2=floor [n/2]
    Create an output image of size M*N.
    Assign zero to all pixel values.  g(x,y)=0
for y = n2 to N-n2-1 loop
for x = m2 to M-m2-1 loop
Compute g(x, y) using previous algorithm .The values including the edges will be computed.
Creating splits: Use getSplits
Dividing the splits into records: use getRecordReader
InputSplit
A chunk of the input that is processed by a single map.
RecordReader
An input split is divided into records, and the map processes each record (key-value pair) in turn. A RecordReader is used to generate record key-value pairs.

File input format
InputFormat describes the input-specification for a Map-Reduce job. Base class for all implementations of InputFormat that use files as their data source. Provides, methods to define which files are included as the input. The key value pair can be implemented as
lnputFormat<K,V> FilelnputFormat<K,V>.
Reuse the methods provided for defining which files are to be included as input to a job.
Define a filter to accept only certain image formats.
Define a custom implementation for generating splits for the input files.

Hadoop Streaming

Hadoop streaming is a utility that is used along with distributed system. This allows a user to create and run map reduce jobs with any other executable language, and script (shell scripting)  as the mapper and the reducer. Before starting work with streaming. The path and the directories have to be set as in Java. The path have to be set for input, output, mapper, combiner, and reducer.

input    <path>                   DFS input file(s) for the Map step
-output   <path>                   DFS output directory for the Reduce step
-mapper   <cmd|JavaClassName>      The streaming command to run
-combiner <JavaClassName>          Combiner has to be a Java class
-reducer  <cmd|JavaClassName>      The streaming command to run

  The mapper and the reducer will be read from the appropriate directories using stdin and stdout commands. Then the whole utility will create a job, submit the job to appropriate cluster and monitor the job until it completes. When an executable is specified for mappers, each mapper task will launch the executable as a separate process when the mapper is initialized. As the mapper task runs, it converts its inputs into lines and feed the lines to the stdin of the process. In the meantime, the mapper collects the line oriented outputs from the stdout of the process and converts each line into a key/value pair, which is collected as the output of the mapper.
   When an executable is specified for reducers, each reducer task will launch the executable as a separate process then the reducer is initialized. As the reducer task runs, it converts its input key/values pairs into lines and feeds the lines to the stdin of the process. In the meantime, the reducer collects the line oriented outputs from the stdout of the process, converts each line into a key/value pair, which is collected as the output of the reducer. By default, the prefix of a line up to the first tab character is the key and the the rest of the line (excluding the tab character) is the value.
  In practical application , streaming in hadoop can be developed with limited knowledge of java. But in developing two shell script have to write. They are mapper.sh and reducer.sh . Both files have to work in sync. The real place the logical correspondence breaks down is that in a one machine scripting environment shellMapper.sh and shellReducer.sh will each run as a single process and data will flow directly from one process to the other. With Hadoop the shellMapper.sh file will be sent to every machine on the cluster that has data chunks and each such machine will run it's own chunk through the shellMapper.sh process on each machine. The output from those scripts doesn't run a reduce on each of those machines. Instead the output is sorted so that different lines from various mapping jobs are streamed across the network to different machines (Hadoop defaults to four machines) where the reduce(s) can be performed.
While job submission, the executable may not be in the system. if they don't, you will need to use "-file" option to tell the framework to pack your executable files as a part of job submission.

-file myPythonScript.py
    -file myDictionary.txt

For mapping jobs only,  Often want to process input data using a map function only. To do this, simply set mapred.reduce.tasks to zero. The map/reduce framework will not create any reducer tasks. Rather, the outputs of the mapper tasks will be the final output of the job.
Fro adding additional plugins, just as map and reduce process, specify other plugins also.

 -inputformat JavaClassName
   -outputformat JavaClassName
   -partitioner JavaClassName
   -combiner JavaClassName

The class supplied for the input format will should return the key/value pairs from the input format class. If it is not mentioned TextInputFormat  is used as the default.  The -cacheFile and -cacheArchive options allow you to make files and archives available to the tasks. The argument is a URI to the file or archive that you have already uploaded to HDFS. These files and archives are cached across jobs. You can specify additional configuration variables by using  jobconf.
There are so  many other options that can be specified in the streaming. They are:

• Switching between local hadoop and remote clusters.
• dfs  configuration
• job tracker configuration
• passing environment variable to streaming commands.
• File uploaded to HDFS

Hadoop JobConf

Job Conf

It is a primary interface created by the user to describe the map reduce concept in Hadoop main frame for execution. The frame works try to execute the function to the best, due to certain reasons as following:

Some parameters will be marked as final by the admin, it cannot be altered.

Some parameters are declared straight but, some other parameters are slightly related to the rest of the frame work and job configuration and is more complex.

In map reduce concept, job conf is used in mapper class, reducer class, partitioner, input format and output format. It is used in map reduce because optionally job conf is used in advanced effects such as Comparators and files in distributed cache.

The configuration using JobConf is as follows:

Step1: create a new JobConf,

JobConf job=new Jobconf(new Configuration(), classname.class);

Step2: Specify carious job specific parameters,
    It’s already mentioned jobconf is used to specify functions for mapper class, reducer class, input format, output format and partitioner. So they are configured using the new job conf object.

job.setJobName("myjob"); /* job name is set*/
    
     FileInputFormat.setInputPaths(job, new Path("in"));
     FileOutputFormat.setOutputPath(job, new Path("out"));
    
     job.setMapperClass(MyJob.MyMapper.class);
     job.setCombinerClass(MyJob.MyReducer.class);
     job.setReducerClass(MyJob.MyReducer.class);
    
     job.setInputFormat(SequenceFileInputFormat.class);
     job.setOutputFormat(SequenceFileOutputFormat.class);


it is included in the packages,
java.lang.Object
  extended by org.apache.hadoop.conf.Configuration
      extended by org.apache.hadoop.mapred.JobConf



Example with the wordcount program:

public static void main(String[] args) throws Exception {
    JobConf conf = new JobConf(WordCount.class); /* new job conf object is created.*/
    conf.setJobName("wordcount"); /* job name is set*/
       
    conf.setOutputKeyClass(Text.class);
    conf.setOutputValueClass(IntWritable.class);
 /*The setOutputKeyClass() and setOutputValueClass() methods control the output types for the map and the reduce functions,*/
           
    conf.setMapperClass(Map.class);
    conf.setCombinerClass(Reduce.class);
    conf.setReducerClass(Reduce.class);
       
    conf.setInputFormat(TextInputFormat.class);
    conf.setOutputFormat(TextOutputFormat.class);
       
    FileInputFormat.setInputPaths(conf, new Path(args[0]));
    FileOutputFormat.setOutputPath(conf, new Path(args[1]));
       
    JobClient.runJob(conf);/* JobClient is the primary interface for the user-job to interact with the cluster. JobClient provides facilities to submit jobs, track their progress, access component-tasks' reports/logs, get the Map-Reduce cluster status information etc.*/
 }