public static class ExampleMapper extends Mapper<Object, Text, Text, AgeCountPair>
    {
        @Override
        protected void map(Object key, Text value, Context context)
            throws IOException, InterruptedException
        {
            String lineString = value.toString();

            String[] splitString = lineString.split(" ");

            if(splitString.length != 3)
                return;

            long age;
            long count;

            try {
                age = Long.parseLong(splitString[1]);
                count = Long.parseLong(splitString[2]);
            } catch(NumberFormatException nfe) {
                return;
            }
            context.write(new Text(splitString[0]), new AgeCountPair(age, count));
        }
    }

    public static class ExampleReducer extends Reducer<Text, AgeCountPair, Object, Text>
    {
        @Override
        protected void reduce(Text key, Iterable<AgeCountPair> values, Context context)
            throws IOException, InterruptedException
        {
            Map<Long, Long> map = new TreeMap<Long, Long>();

            long sum = 0;

            for(AgeCountPair value : values)
            {
                long age = value.getAge();
                long count = value.getCount();

                if(false == map.containsKey(age)) {
                    map.put(age, count);
                }else{
                    map.put(age, map.get(age) + count);
                }

                sum += count;
            }

            for(Map.Entry<Long, Long> entry : map.entrySet())
            {
                long age = entry.getKey();
                long cnt = entry.getValue();

                float percentage = cnt / (float)sum;

                context.write(null, new Text(String.format("%s %d %d %f", key.toString(), age, cnt, percentage)));
            }
        }
    }

    public static class AgeCountPair implements Writable {
        private long age;
        private long count;

        public AgeCountPair(long age, long count)
        {
            this.age = age;
            this.count = count;
        }

        @Override
        public void write(DataOutput out) throws IOException {
            out.writeLong(age);
            out.writeLong(count);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            age = in.readLong();
            count = in.readLong();
        }

        public long getAge() {
            return age;
        }

        public void setAge(long age) {
            this.age = age;
        }

        public long getCount() {
            return count;
        }

        public void setCount(long count) {
            this.count = count;
        }
    }