hadoop新手。我想计算文本中每行单词的共现，比如一个单词与其他单词出现在同一行的次数。为此，我创建了一个特殊的单词对类，因此mapreduce将给我一对单词，然后是计数。问题是，结果是一团糟，我不知道我错在哪里。
我的单词对课是这样的：

public class Par implements Writable,WritableComparable<Par> {

    public String palabra;
    public String vecino;

    public Par(String palabra, String vecino) {
        this.palabra = palabra;
        this.vecino = vecino;
    }

    public Par() {
        this.palabra = new String();
        this.vecino = new String();
    }

    @Override
    public int compareTo(Par otra) {
        int retorno = this.palabra.compareTo(otra.palabra);
        if(retorno != 0){
            return retorno;
        }
        return this.vecino.compareTo(otra.vecino);
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(palabra);
         out.writeUTF(vecino);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        palabra = in.readUTF();
        vecino = in.readUTF();
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((palabra == null) ? 0 : palabra.hashCode());
        result = prime * result + ((vecino == null) ? 0 : vecino.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        Par other = (Par) obj;
        if (palabra == null) {
            if (other.palabra != null)
                return false;
        } else if (!palabra.equals(other.palabra))
            return false;
        if (vecino == null) {
            if (other.vecino != null)
                return false;
        } else if (!vecino.equals(other.vecino))
            return false;
        return true;
    }

    @Override
    public String toString() {
        return "Par [" + palabra + " , " + vecino + "]";
    }

}

我的Map器是：

public class Co_OcurrenciaMapper extends Mapper<LongWritable, Text, Par, IntWritable> {
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        IntWritable one = new IntWritable(1);
        String[] palabras = value.toString().split("\\s+");
        if (palabras.length > 1) {
            for (int i = 0; i < palabras.length - 1; i++) {
                for (int j = i + 1; j < palabras.length; j++) {
                    context.write(new Par(palabras[i], palabras[j]), one);
                }
            }
        }
    }
}

我用mapreduce得到的结果是：

[cloudera@quickstart Desktop]$ hadoop fs -cat salidaO11/part-r-00000 |head -15
Par [ , &c.]    35
Par [ , &c.']   2
Par [ , &c.,]   4
Par [ , &c]]    23
Par [ , ']  6
Par [ , ''Od's] 1
Par [ , ''Tis]  2
Par [ , ''tis]  1
Par [ , ''twas] 1
Par [ , '--O]   1
Par [ , 'A] 17
Par [ , 'ARTEMIDORUS.'] 1
Par [ , 'Above] 1
Par [ , 'Achilles]  2
Par [ , 'Ad]    3
cat: Unable to write to output stream.

我错在哪里？一个朋友建议把这两个词合并成一个字符串，但我认为它不那么优雅。