Generating GAPLESS SEQUENCE NUMBER in map reduce.

Lets say there is a large data file and using map reduce we would like to add a row number to the data. Following would be the visual representation of the approach

The following code is just representational and is not tested. 

SequenceZKMapReduce.java

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.*;
import java.util.HashMap;


public class SequenceZKMapReduce {

    //Enum for map reduce counters    public static enum SEQ_COUNTER {
        OUTPUT_RECORD    };

    //Mapper class that would output the mapper id as the key and the entire text as value    //Once the mapper finishes the work, using cleanup, its going to report the number of records to zookeper
    public static class SequenceMapper extends Mapper<Object, Text, IntWritable, Text> {

        private final IntWritable ConstKey = new IntWritable();

        //Intial setup        //Set the mapper id as the key for the data to be emmited from this mapper.        protected void setup(Context context){
            ConstKey.set(context.getTaskAttemptID().getTaskID().getId());
        }

        //Write the constant key which is the mapper id, and the value as the entire string        //Increase the counter for every record processed        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            context.write(ConstKey, value);
            context.getCounter(SEQ_COUNTER.OUTPUT_RECORD).increment(1);
        }

        //This method would be called at the close of the map task.        //We are going to use this method to write to zookeeper        protected void cleanup(Context context) {
            try {
                ZooKeeperMgr zk = new ZooKeeperMgr();
                //Connect to zookeeper, I am hardcoding the zookeeper host to localhost                zk.connect("localhost");
                //Get the existing data on the node and deserialize to a hashmap                byte[] dt = zk.readZnode();
                int ver = zk.getVersion();
                HashMap<Integer, Long> h = null;

                //If data is not empty deserialize to get the hashmap else create a new one.                if(dt!=null){
                    ByteArrayInputStream bis = new ByteArrayInputStream(dt);
                    ObjectInput in = new ObjectInputStream(bis);
                    h = (HashMap<Integer, Long>) in.readObject();
                }else{
                    h= new HashMap<Integer, Long>();
                }

                //Add the map key and the counter to the hashmap                h.put(ConstKey.get(), context.getCounter(SEQ_COUNTER.OUTPUT_RECORD).getValue());
                //Write the data back to the znode                ByteArrayOutputStream bos = new ByteArrayOutputStream();
                ObjectOutput out = new ObjectOutputStream(bos);
                out.writeObject(h);
                zk.writeZnode(bos.toByteArray(), ver); // This would throw exception if version changed between read and write
            } catch (Exception e) {
                //If there is version mismatch write exception, call the method again.                //To make it safe, some counter should be established. Otherwise it may get into a loop.                this.cleanup(context);
            }
        }

    }

    //Reducer Class that spits out a sequence number and the text as it came    public static class ReorgReducer extends Reducer<IntWritable, Text, LongWritable, Text> {

        HashMap<Integer, Long> h;
        private long sequence = 0L;
        private LongWritable sequenceHolder = new LongWritable(1);

        //In the setup phase, read the znode and extract all ranges (hashmap)        protected void setup(Context context){
            try {
                ZooKeeperMgr zk = new ZooKeeperMgr();
                //Connect to zookeeper, I am hardcoding the zookeeper host to localhost                zk.connect("localhost");
                //Get the existing data on the node and deserialize to a hashmap                byte[] dt = zk.readZnode();
                ByteArrayInputStream bis = new ByteArrayInputStream(dt);
                ObjectInput in = new ObjectInputStream(bis);
                h = (HashMap<Integer, Long>) in.readObject();
            }catch(Exception e){

            }
        }

        public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            //Get the start -1 position for the given key            sequence = getStartPosition(h, key.get());

            for (Text val : values) {
                sequence=sequence+1;
                sequenceHolder.set(sequence);
                context.write(sequenceHolder , val);
            }
        }

        //Given a key, this method looks for all the keys less than the given and sums all values.        //the idea is if a reducer if processing data from a mapper with id 102, it will first sum up the range for all ids        //less than 102 and start the sequencing after that.        private long getStartPosition(HashMap<Integer, Long> ranges, int currentKey){
            Long startPos=0L;
            for(int key : ranges.keySet()){
                if(key < currentKey) startPos = startPos + ranges.get(key);
            }
            return startPos;
        }


    }

    //Job Runner    public static int run(String inpPath, String outPath) throws Exception{

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Add Sequence Number with help of zoo keeper");

        job.setJarByClass(SequenceMapReduce.class);
        job.setMapperClass(SequenceMapper.class);
        job.setReducerClass(ReorgReducer.class);

        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(inpPath));
        FileOutputFormat.setOutputPath(job, new Path(outPath));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    //Main method    public static void main(String args[]) throws Exception{
        System.exit(run(args[0], args[1]));
    }
}

ZooKeeperMgr.java

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.util.concurrent.CountDownLatch;

import java.io.IOException;

public class ZooKeeperMgr {

    String nodePath = "/sequenceapp";
    ZooKeeper zookeeper;
    CountDownLatch connectStatus = new CountDownLatch(1);


    //Connect to the zookeeper and get the handle    //This method should be called from the driver, mapper and reducer setups    public void connect(String host) throws IOException, InterruptedException {
        zookeeper = new ZooKeeper(host, 5000,
                new Watcher() {
                    public void process(WatchedEvent event) {
                        if (event.getState() == Event.KeeperState.SyncConnected) {
                            connectStatus.countDown();
                        }
                    }
                });
        connectStatus.await();
    }

    //Create znode    //This method should only be called from driver    public void createZnode() throws KeeperException, InterruptedException {
        try {
            zookeeper.create(nodePath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch(InterruptedException inex) {
        } catch(KeeperException kex) {
        }
    }

    //Read from znode.    //This method to be used by mapper and reducer    public byte[] readZnode() throws KeeperException, InterruptedException {
        byte[] dt = null;
        try {
            dt = zookeeper.getData(nodePath, true, zookeeper.exists(nodePath, true));
        } catch(Exception ex) {
        }
        return dt;
    }

    //Get the znode data version    //This method to be used by mappers before writing the data    //Important to ensure that data is not overwrittten    public int getVersion() throws KeeperException, InterruptedException {
        Stat stat = zookeeper.exists(nodePath, true);
        return stat.getVersion();
    }

    //Write to znode    //This methoid to be used only by mapper    public void writeZnode(byte[] data, int version) throws KeeperException, InterruptedException {
        zookeeper.setData(nodePath, data, version);
    }

    //Delete znode    //Method to be only called by the driver to delete the node before the app closes    public void deleteNode()throws InterruptedException, KeeperException{
        zookeeper.delete(this.nodePath, -1);
    }


    //Close the connection    public void close() throws InterruptedException, KeeperException {
        zookeeper.close();
    }

    //This method is redundant. Keeping for just in case the zookeeper instance is required.    public ZooKeeper getZooKeeper() {
        if (zookeeper == null || !zookeeper.getState().equals(ZooKeeper.States.CONNECTED)) {
            throw new IllegalStateException("ZooKeeper is not connected.");
        }
        return zookeeper;
    }

}