CODEWITHSUNDEEP
hadoopmapreducecloudera
aa8y
aa8y

Reputation: 3942

Blank / Empty Values of Key and Value retrieved in the Mapper class

I have written a MapReduce code for running it on a CDH4 cluster. My requirement was to read the complete file as the value and the file name as the key. For that I wrote custom InputFormat and RecordReader classes.

Custom input format class: FullFileInputFormat.java

import java.io.*;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;

import FullFileRecordReader;

public class FullFileInputFormat extends FileInputFormat<Text, Text> {

    @Override
    public RecordReader<Text, Text> getRecordReader(InputSplit split, JobConf jobConf, Reporter reporter) throws IOException {
        reporter.setStatus(split.toString());
        return new FullFileRecordReader((FileSplit) split, jobConf);
    }
}

And the custom RecordReader class: FullFileRecordReader.java

import java.io.BufferedReader;
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.mapred.*;

public class FullFileRecordReader implements RecordReader<Text, Text> {

    private BufferedReader in;
    private boolean processed = false;
    private int processedBytes = 0;

    private FileSplit fileSplit;
    private JobConf conf;

    public FullFileRecordReader(FileSplit fileSplit, JobConf conf) {
        this.fileSplit = fileSplit;
        this.conf = conf;
    }

    @Override
    public void close() throws IOException {
        if (in != null) {
            in.close();
        }
    }

    @Override
    public Text createKey() {
        return new Text("");
    }

    @Override
    public Text createValue() {
        return new Text("");
    }

    @Override
    public long getPos() throws IOException {
        return processedBytes;
    }

    @Override
    public boolean next(Text key, Text value) throws IOException {
        Path filePath = fileSplit.getPath();

        if (!processed) {
            key = new Text(filePath.getName());

            value = new Text("");
            FileSystem fs = filePath.getFileSystem(conf);
            FSDataInputStream fileIn = fs.open(filePath);
            byte[] b = new byte[1024];
            int numBytes = 0;

            while ((numBytes = fileIn.read(b)) > 0) {
                value.append(b, 0, numBytes);
                processedBytes += numBytes;
            }
            processed = true;
            return true;
        }
        return false;
    }

    @Override
    public float getProgress() throws IOException {
        return 0;
    }
}

Though whenever I try to print the key-value in the RecordReader class, I get their values, but when I print the same in the mapper class, I see blank values for them. I am unable to understand why the Mapper class is unable to get any data for keys and values.

Currently I have only a Map job and no reduce job. The code is:

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.*;

import FullFileInputFormat;

public class Source {

    public static class Map extends MapReduceBase implements Mapper<Text, Text, Text, Text> {

        public void map(Text key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws java.io.IOException {
            System.out.println("Processing " + key.toString());
            System.out.println("Value: " + value.toString());
        }
    }

    public static void main(String[] args) throws Exception {
        JobConf job = new JobConf(Source.class);
        job.setJobName("Source");

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setJarByClass(Source.class);
        job.setInputFormat(FullFileInputFormat.class);
        job.setMapperClass(Map.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        JobClient.runJob(job);
    }
}

Upvotes: 2

Views: 3096

Answers (1)

Chris White
Chris White

Reputation: 30089

You're creating new instances in your next method - hadoop re-uses objects so you are expected to populate the ones passed. It should be as simple as amending as follows:

@Override
public boolean next(Text key, Text value) throws IOException {
    Path filePath = fileSplit.getPath();

    if (!processed) {
        // key = new Text(filePath.getName());
        key.set(filePath.getName());

        // value = new Text("");
        value.clear();
    }

I would also recommend pre-sizing the value text to avoid 'growing' pains of the value's underlying byte array. Text has a private method called setCapacity, so you unforntunately can't call it - but if you used a BytesWritable to buffer the file input, you can call setCapacity in side your next method, passing the fileSplit length (note this may still be wrong if your file is compressed - as the file size is the compressed size).

Upvotes: 1

Related Questions