In this article, We will see how to process Sensex Log (Share Market) which is in PDF format using Big Data Technology, We will see step by step process execution of the project.
Problem Statement: Analyse the data in Hadoop Eco-system to:
- Take the complete PDF Input data on HDFS
- Develop a Map-Reduce Use Case to get the below-filtered results from the HDFS Input data(Excel data)
If TYPE OF TRADING is –>’SIP’
– OPEN_BALANCE > 25000 & FLTUATION_RATE > 10 –> store “HighDemandMarket”
-CLOSING_BALANCE<22000 & FLTUATION_RATE IN BETWEEN 20 – 30 –> store “OnGoingMarketStretegy”
If TYPE OF TRADING is –>’SHORTTERM
– OPEN_BALANCE < 5000 –> store “WealthyProducts”
– SensexLoc –> “NewYork OR Mumbai” –> “ReliableProducts
else
store in “OtherProducts”
NOTE: In the mentioned file names only 5 outputs have to be generated
- Develop a PIG Script to filter the Map Reduce Output in the below fashion
– Provide the Unique data
– Sort the Unique data based on SensexID.
- EXPORT the same PIG Output from HDFS to MySQL using SQOOP
- Store the same PIG Output in a HIVE External Table
Attribute Information or Dataset Details:
Data: Input Format – .PDF (Our Input Data is in PDF Format)
Like this below created 3000 records on my own
Technology Used
- Apache Hadoop (HDFS)
- Mapreduce
- Apache Pig
- Apache Hive
- MySQL
- Shell Script
- Apache Sqoop
- Linux
- Java
Flow Chart
Mapreduce Process
Mapreduce Code
PdfInputDriver.java
Java
x
45
1
package com.bhavesh.poc.sensex;2
import java.io.IOException;3
import org.apache.hadoop.conf.Configuration;4
import org.apache.hadoop.fs.Path;5
import org.apache.hadoop.io.IntWritable;6
import org.apache.hadoop.io.LongWritable;7
import org.apache.hadoop.io.Text;8
import org.apache.hadoop.mapreduce.Job;9
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;10
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;11
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;12
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;13
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;14
import org.apache.hadoop.util.GenericOptionsParser;15
16
public class PdfInputDriver {17
18
public static void main(String[] args) throws IOException,19
InterruptedException, ClassNotFoundException {20
Configuration conf = new Configuration();21
//GenericOptionsParser is a utility to parse command line arguments generic to the Hadoop framework22
GenericOptionsParser parser = new GenericOptionsParser(conf, args);23
24
//Returns an array of Strings containing only application-specific arguments25
args = parser.getRemainingArgs();26
27
Job job = new Job(conf, "PdfSensexDetails");28
job.setJarByClass(PdfInputDriver.class);29
job.setOutputKeyClass(Text.class);30
31
job.setOutputValueClass(Text.class);32
// Custom InputFormat class33
job.setInputFormatClass(PdfInputFormat.class);34
35
//job.setOutputFormatClass(TextOutputFormat.class);36
LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);37
MultipleOutputs.addNamedOutput(job, "text", TextOutputFormat.class,IntWritable.class, Text.class);38
39
FileInputFormat.setInputPaths(job, new Path(args[0]));40
FileOutputFormat.setOutputPath(job, new Path(args[1]));41
job.setMapperClass(SensexTradeMapper.class);42
job.setReducerClass(SensexTradeReducer.class);43
System.out.println(job.waitForCompletion(true));44
}45
}PdfInputFormat.java
Java
1
17
1
package com.bhavesh.poc.sensex;2
import java.io.IOException;3
import org.apache.hadoop.io.LongWritable;4
import org.apache.hadoop.io.Text;5
import org.apache.hadoop.mapreduce.InputSplit;6
import org.apache.hadoop.mapreduce.RecordReader;7
import org.apache.hadoop.mapreduce.TaskAttemptContext;8
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;9
public class PdfInputFormat extends FileInputFormat {10
11
public RecordReader createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException,12
InterruptedException {13
14
return new PdfRecordReader();15
}16
}17
PdfRecordReader.java
Java
1
83
1
package com.bhavesh.poc.sensex;2
import java.io.IOException;3
import org.apache.hadoop.conf.Configuration;4
import org.apache.hadoop.fs.FSDataInputStream;5
import org.apache.hadoop.fs.FileSystem;6
import org.apache.hadoop.fs.Path;7
import org.apache.hadoop.io.LongWritable;8
import org.apache.hadoop.io.Text;9
import org.apache.hadoop.mapreduce.InputSplit;10
import org.apache.hadoop.mapreduce.RecordReader;11
import org.apache.hadoop.mapreduce.TaskAttemptContext;12
import org.apache.hadoop.mapreduce.lib.input.FileSplit;13
import org.apache.pdfbox.pdmodel.PDDocument;14
import org.apache.pdfbox.util.PDFTextStripper;15
public class PdfRecordReader extends RecordReader{16
private String[] lines = null;17
private LongWritable key = null;18
private Text value = null;19
20
public void initialize(InputSplit genericSplit, TaskAttemptContext context)21
throws IOException, InterruptedException {22
FileSplit split = (FileSplit) genericSplit;23
Configuration job = context.getConfiguration();24
final Path file = split.getPath();25
26
/*27
* The below code contains the logic for opening the file and seek to28
* the start of the split. Here we are applying the Pdf Parsing logic29
*/30
31
FileSystem fs = file.getFileSystem(job);32
FSDataInputStream fileIn = fs.open(split.getPath());33
PDDocument pdf = null;34
String parsedText = null;35
PDFTextStripper stripper; // Here we are coverting PDF to textinputformate36
pdf = PDDocument.load(fileIn);37
stripper = new PDFTextStripper();38
parsedText = stripper.getText(pdf); 39
this.lines = parsedText.split("\n");40
}41
42
public boolean nextKeyValue() throws IOException, InterruptedException {43
if (key == null) {44
key = new LongWritable();45
key.set(1);46
value = new Text();47
value.set(lines[0]);48
} else {49
int temp = (int) key.get();50
if (temp < (lines.length - 1)) {51
int count = (int) key.get();52
value = new Text();53
value.set(lines[count]);54
count = count + 1;55
key = new LongWritable(count);56
} else {57
return false;58
}59
}60
if (key == null || value == null) {61
return false;62
} else {63
return true;64
}65
}66
67
68
public LongWritable getCurrentKey() throws IOException,69
InterruptedException {70
return key;71
}72
73
public Text getCurrentValue() throws IOException, InterruptedException {74
return value;75
}76
77
public float getProgress() throws IOException, InterruptedException {78
return 0;79
}80
81
public void close() throws IOException {82
}83
}SensexTradeMapper.java
Java
1
57
1
package com.bhavesh.poc.sensex;2
import java.io.IOException;3
import java.util.StringTokenizer;4
import org.apache.hadoop.io.LongWritable;5
import org.apache.hadoop.io.Text;6
import org.apache.hadoop.mapreduce.Mapper;7
public class SensexTradeMapper extends Mapper<LongWritable, Text, Text, Text> {8
private Text word = new Text();9
protected void map(LongWritable key, Text value, Context context)10
throws IOException, InterruptedException {11
String TradeType;12
String Location;13
int Open_bal;14
int Close_bal;15
int Fluct_rate;16
String str[] = value.toString().split(",");17
TradeType =str[2].toString();18
Location = str[3].toString();19
String Opbal = str[4];20
Open_bal = Integer.parseInt(Opbal);21
String Clbal = str[5];22
Close_bal = Integer.parseInt(Clbal);23
String fcrate = str[6].trim();24
Fluct_rate = Integer.parseInt(fcrate);25
if (TradeType.equals(new String("SIP")))26
{27
if (Open_bal > 20000 && Fluct_rate <= 15)28
{29
word.set(new String("HighDemandMarket")); //word == key30
}31
else if(Close_bal > 30000 && Fluct_rate > 15 && Fluct_rate < 25)32
{33
word.set(new String("OnGoingMarketStretegy"));34
}35
}36
else if(TradeType.equals(new String("SHORTTERM")))37
{38
if (Open_bal > 5000 && Open_bal < 30000)39
{40
word.set(new String("WealthyProducts"));41
}42
else if((Location.equals(new String("NEWYORK"))) ||43
(Location.equals(new String("MUMBAI"))))44
{45
word.set(new String("ReliableProducts"));46
}47
}48
else49
{50
word.set(new String("OtherProducts"));51
}52
String rec = str[0]+"\t"+str[1]+"\t"+str[2]+"\t"+str[3]+"\t"+str[4]+"\t"+str[5]+"\t"+str[6];53
54
//context.progress();55
context.write(word, new Text(rec));56
}57
}SensexTradeReducer.java
Java
1
38
1
package com.poc.ssx;2
import java.io.IOException;3
import org.apache.hadoop.io.IntWritable;4
import org.apache.hadoop.io.LongWritable;5
import org.apache.hadoop.io.Text;6
import org.apache.hadoop.mapreduce.Reducer;7
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;8
public class SensexTradeReducer extends9
Reducer<Text, Text, IntWritable, Text> {10
private MultipleOutputs<IntWritable, Text> multipleOutputs;11
protected void reduce(Text key, Iterable<Text> values,12
Context context) throws IOException, InterruptedException {13
String str1 = null;14
int ssxid = 0;15
for (Text value : values) {16
String str[] = value.toString().split("\t");17
String sxid = str[0].trim();18
ssxid = Integer.parseInt(sxid);19
str1 = str[1]+"\t"+str[2]+"\t"+str[3]+"\t"+str[4]+"\t"+str[5]+"\t"+str[6];20
//context.write(new IntWritable(ssxid), new Text(ssxname+","+TradeType+","+Location+","+Open_bal+","+Close_bal+","+Fluct_rate));21
multipleOutputs.write(new IntWritable(ssxid),22
new Text(str1),23
generateFileName(key));24
}25
}26
String generateFileName(Text key){27
28
return key.toString(); 29
}30
31
public void setup(Context context){32
multipleOutputs = new MultipleOutputs<IntWritable, Text>(context);33
}34
35
public void cleanup(final Context context) throws IOException, InterruptedException{36
multipleOutputs.close();37
} 38
}