How to use multiple fields with MapReduce?

Question

I would like to understand how I could use MapReduce model to aggregate multiple fields.

For example if I got a data file such as this:

id, site, name, qty, price
00, testA, NameA,1,1
01, testB,NameA,2,3
02, testB,NameB,5,7

and want to implement this aggregation on MapReduce:

select site,name, (qty*price) as total
from PO where name ='NameA' 
group by site,name,total 
order by site;

How would I go about doing it.

I'm able to aggregate by site (key), total (value) but not sure how to include name column.

I need to understand how I could work with multiple fields in MapReduce. Is there an example I could look at? Or do I need to use Hbase?

Answer 1

You could implement WritableComparable and create your own CompositeKey with several fields, for example:

public static class CompositeKey implements WritableComparable<CompositeKey> {
    public final Text site;
    public final Text name;
    public final LongWritable total;

    public CompositeKey(Text site, Text name, LongWritable total) {
        this.site = site;
        this.name = name;
        this.total = total;
    }

    @Override
    public int compareTo(CompositeKey o) {
        int siteCmp = site.compareTo(o.site);
        if (siteCmp != 0) {
            return siteCmp;
        }
        int nameCmp = name.compareTo(o.name);
        if (nameCmp != 0) {
            return nameCmp;
        }
        return total.compareTo(o.total);
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        CompositeKey that = (CompositeKey) o;

        if (name != null ? !name.equals(that.name) : that.name != null) return false;
        if (site != null ? !site.equals(that.site) : that.site != null) return false;
        if (total != null ? !total.equals(that.total) : that.total != null) return false;

        return true;
    }

    @Override
    public int hashCode() {
        int result = site != null ? site.hashCode() : 0;
        result = 31 * result + (name != null ? name.hashCode() : 0);
        result = 31 * result + (total != null ? total.hashCode() : 0);
        return result;
    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        site.write(dataOutput);
        name.write(dataOutput);
        total.write(dataOutput);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        site.readFields(dataInput);
        name.readFields(dataInput);
        total.readFields(dataInput);
    }
}