java - 在 Hadoop 中设置可写？

Question

我正在尝试在 Hadoop 中创建一个 SetWritable。这是我的实现。我刚开始使用 MapReduce，我不知道我应该怎么做。我写了下面的代码，但它不起作用。

Custom Writable(需要设置):

public class TextPair implements Writable {

    private Text first;
    public HashSet<String> valueSet = new HashSet<String>();
    public TextPair() {

    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(valueSet.size());
        Iterator<String> it = valueSet.iterator();
        while (it.hasNext()) {
            this.first = new Text(it.next());
            first.write(out);
        }
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        Iterator<String> it = valueSet.iterator();
        while (it.hasNext()) {
            this.first = new Text(it.next());
            first.readFields(in);
        }
    }

}

映射器代码:

public class TokenizerMapper extends Mapper<Object, Text, Text, TextPair> {

    ArrayList<String> al = new ArrayList<String>();
    TextPair tp = new TextPair();

    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

        String [] val = value.toString().substring(2,value.toString().length()).split(" ");

        for(String v: val) {
            tp.valueSet.add(v);
        }
        String [] vals = value.toString().split(" ");

        for(int i=0; i<vals.length-1; i++) {
            setKey(vals[0],vals[i+1]);
            System.out.println(getKey());
            context.write(new Text(getKey()), tp); 
        }
    }

    public void setKey(String first,String second) {

        al.clear();
        al.add(first);
        al.add(second);

        java.util.Collections.sort(al);
    }

    public String getKey() {

        String tp = al.get(0)+al.get(1);
        return tp;
    }
 }

我基本上是在尝试从 Mapper 发出一个 SetWritable 作为值。请建议我需要进行哪些更改。谢谢!

Answer 1

我会说你的读写方式有问题。您需要知道 Set 有多大，并使用它来读取正确数量的 Text 对象。

我将您的版本更改为一组文本对象，因为它们可以轻松读取和写入。

public class TextWritable implements Writable {

    private Set<Text> values;

    public TextPair() {
        values = new HashSet<Text>();
    }

    @Override
    public void write(DataOutput out) throws IOException {

        // Write out the size of the Set
        out.writeInt(valueSet.size());

        // Write out each Text object
        for(Text t : values) {
            t.write(out);
        }
    }

    @Override
    public void readFields(DataInput in) throws IOException {

        // Make sure we have a HashSet to fill up
        values = new HashSet<Text>();

        // Get the number of elements in the set
        int size = in.readInt();

        // Read the correct number of Text objects
        for(int i=0; i<size; i++) {
            Text t = new Text();
            t.readFields(in);
            values.add(t);
        }
    }
}

您应该为此添加一些辅助类，以便将元素添加到 Set。

我也看不到您在 map 方法中clear Set 的位置。如果您不清除它，它可能会在每次调用 map 方法时变得越来越大。

参见 Hadoop ArrayWritable供引用。