Hadoop 0.20.205.0 WritableComparator не учитывает настраиваемые ключи - PullRequest
Новая
       25

Hadoop 0.20.205.0 WritableComparator не учитывает настраиваемые ключи

1 голос
/ 21 декабря 2011

Я пытался запустить задание hadoop 0.20.205.0 MapReduce (однопоточное, локально), которое демонстрирует все виды странного и неожиданного поведения.Я наконец понял, почему.Это выглядит для меня как ошибка в hadoop, но, возможно, есть кое-что, чего я не понимаю.Может ли кто-нибудь дать мне совет?Мой класс setMapOutputKeyClass реализует Configurable.Метод readFields не будет правильно считываться, если сначала не вызывается setConf (я полагаю, в этом и заключается смысл интерфейса Configurable). Но, глядя на код для WritableComparator, я вижу, что когда платформа сортирует их, она создает свои объекты внутреннего ключа с помощью: 

70      key1 = newKey();
71      key2 = newKey();

И newKey () использует нулевую конфигурацию для создания ключей: 

83  public WritableComparable newKey() {
84    return ReflectionUtils.newInstance(keyClass, null);
85  }

Действительно, когда я запускаю в отладчике, я обнаруживаю, что в 

91      key1.readFields(buffer);

confвнутри key1 значение null, поэтому setConf не был вызван.

Это ошибка в hadoop или я должен использовать что-то другое, кроме Configurable, для настройки ключей?И если это ошибка, кто-нибудь знает какие-либо обходные пути?

РЕДАКТИРОВАТЬ: Вот краткий (несколько надуманный) пример задания, которое по этой причине не удается: 

// example/WrapperKey.java

package example;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.ByteWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;

/**
 * This class wraps a WritableComparable class to add one extra possible value
 * (namely null) to the range of values available for that class.
 */
public class WrapperKey<T extends WritableComparable> implements
        WritableComparable<WrapperKey<T>>, Configurable {
    private T myInstance;
    private boolean isNull;
    private Configuration conf;

    @Override
    public void setConf(Configuration conf) {
        this.conf = conf;
        Class<T> heldClass = (Class<T>) conf.getClass("example.held.class",
                null, WritableComparable.class);
        myInstance = ReflectionUtils.newInstance(heldClass, conf);
    }

    @Override
    public Configuration getConf() {
        return conf;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeBoolean(isNull);
        if (!isNull)
            myInstance.write(out);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        isNull = in.readBoolean();
        if (!isNull)
            myInstance.readFields(in);
    }

    @Override
    public int compareTo(WrapperKey<T> o) {
        if (isNull) {
            if (o.isNull)
                return 0;
            else
                return -1;
        } else if (o.isNull)
            return 1;
        else
            return myInstance.compareTo(o.myInstance);
    }

    public void clear() {
        isNull = true;
    }

    public T get() {
        return myInstance;
    }

    /**
     * Should sort the KV pairs (5,0), (3,0), and (null,0) to [(null,0), (3,0), (5,0)], but instead fails
     * with a NullPointerException because WritableComparator's internal keys
     * are not properly configured
     */
    public static void main(String[] args) throws IOException,
            InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        conf.setClass("example.held.class", ByteWritable.class,
                WritableComparable.class);
        Path p = new Path("input");
        Path startFile = new Path(p, "inputFile");
        SequenceFile.Writer writer = new SequenceFile.Writer(
                p.getFileSystem(conf), conf, startFile, WrapperKey.class,
                ByteWritable.class);
        WrapperKey<ByteWritable> key = new WrapperKey<ByteWritable>();
        key.setConf(conf);
        ByteWritable value = new ByteWritable((byte) 0);
        key.get().set((byte) 5);
        writer.append(key, value);
        key.get().set((byte) 3);
        writer.append(key, value);
        key.clear();
        writer.append(key, value);
        writer.close();

        Job j = new Job(conf, "Example job");
        j.setInputFormatClass(SequenceFileInputFormat.class);
        j.setOutputKeyClass(WrapperKey.class);
        j.setOutputValueClass(ByteWritable.class);
        j.setOutputFormatClass(SequenceFileOutputFormat.class);
        FileInputFormat.setInputPaths(j, p);
        FileOutputFormat.setOutputPath(j, new Path("output"));
        boolean completed = j.waitForCompletion(true);
        if (completed) {
            System.out
                    .println("Successfully sorted byte-pairs by key (putting all null pairs first)");
        } else {
            throw new RuntimeException("Failed to sort");
        }
    }
}

1 Ответ

0 голосов
/ 22 декабря 2011

WrapperKey реализует Configurable и реализует setConf.Простая реализация интерфейса не означает, что какой-то другой класс будет вызывать это.Платформа Hadoop может не вызывать метод setConf для ключей.

Не думаю, что это ошибка.Все типы, которые я видел, реализованы только в WritableComparable и не являются настраиваемыми.Не уверен, что в качестве обходного пути для этого, возможно, вам придется определить конкретные типы в ключе.

...