hadoop - 如何在 MapReduce 中使用多个字段？

Question

我想了解如何使用 MapReduce 模型聚合多个字段。

例如，如果我有这样一个数据文件:

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

并希望在 MapReduce 上实现此聚合:

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

我该怎么做。

我可以按站点(键)、总计(值)进行汇总，但不确定如何包含名称列。

我需要了解如何在 MapReduce 中处理多个字段。有没有我可以看的例子？或者我需要使用 Hbase 吗？

Answer 1

您可以实现 WritableComparable 并创建您自己的包含多个字段的 CompositeKey，例如:

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);
    }
}