Hadoop - конструктор аргументов для картографа

Question

Есть ли способ передать аргументы конструктора Mapper в Hadoop?Возможно, через какую-то библиотеку, которая оборачивает создание задания?

Вот мой сценарий:

public class HadoopTest {

    // Extractor turns a line into a "feature"
    public static interface Extractor {
        public String extract(String s);
    }

    // A concrete Extractor, configurable with a constructor parameter
    public static class PrefixExtractor implements Extractor {
        private int endIndex;

        public PrefixExtractor(int endIndex) { this.endIndex = endIndex; }

        public String extract(String s) { return s.substring(0, this.endIndex); }
    }

    public static class Map extends Mapper<Object, Text, Text, Text> {
        private Extractor extractor;

        // Constructor configures the extractor
        public Map(Extractor extractor) { this.extractor = extractor; }

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String feature = extractor.extract(value.toString());
            context.write(new Text(feature), new Text(value.toString()));
        }
    }

    public static class Reduce extends Reducer<Text, Text, Text, Text> {
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            for (Text val : values) context.write(key, val);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = new Job(conf, "test");
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);
    }
}

Как должно быть понятно, поскольку Mapper предоставляется только для Configuration в качестве ссылки на класс (Map.class), Hadoop не может передать аргумент конструктора и сконфигурировать конкретный Extractor.

Существуют интегрированные среды Hadoop, такие как Scoobi, Crunch, Scrunch (и, вероятно, многие другие, которых я не знаюо) которые, кажется, имеют такую ​​возможность, но я не знаю, как они это делают. РЕДАКТИРОВАТЬ: После еще большей работы со Скуби, я обнаружил, что я был частично неправ в этом.Если вы используете внешне определенный объект в «mapper», Scoobi требует, чтобы он был сериализуемым, и будет жаловаться во время выполнения, если это не так.Поэтому, возможно, правильный путь - просто сделать мой сериализуемый Extractor и десериализовать его в методе установки Mapper ...

Кроме того, я на самом деле работаю в Scala, так что решения на основе Scala определенно приветствуются (если не поощряется!)

Answer 1

Я бы посоветовал сообщить вашему мапперу, какой экстрактор использовать через объект Configuration, который вы создаете.Картограф получает конфигурацию в своем методе setup (context.getConfiguration()).Кажется, что вы не можете помещать объекты в конфигурацию, поскольку она обычно создается из файлов XML или командной строки, но вы можете установить значение enum и сделать так, чтобы mapper сам создавал свой экстрактор.Не очень красиво настраивать маппер после его создания, но я так интерпретировал API.

Answer 2

Установите имя класса реализации при отправке задания как

Configuration conf = new Configuration();
conf.set("PrefixExtractorClass", "com.my.class.ThreePrefixExtractor");

или используйте параметр -D из командной строки, чтобы установить параметр PrefixExtractorClass.

Нижеэто реализация в mapper

Extractor extractor = null;
protected void setup(Context context) throws IOException,
            InterruptedException
{
    try {
        Configuration conf = context.getConfiguration();
        String className = conf.get("PrefixExtractorClass");
        extractor = Class.forName(className);
    } Catch (ClassNotFoundException e) {
        //handle the exception
    }
}

Теперь используйте объект extractor, как требуется для функции карты.

• Jar, содержащий класс com.my.class.ThreePrefixExtractor, долженраспределяться по всем узлам.Вот статья от Cloudera о различных способах сделать это.

• В приведенном выше примере com.my.class.ThreePrefixExtractor следует расширить класс Extractor.

При использовании этого подхода реализация mapper может быть сделана общей.Это подход (использующий Class.forName), принятый большинством сред для создания подключаемых компонентов, которые реализуют определенный интерфейс.

Answer 3

Я все еще ищу хорошие ответы, но одно (неидеальное) решение, которое я нашел, состоит в том, чтобы использовать наследование вместо композиции, превращая Map в абстрактный класс Extractor.Затем он может быть полностью разделен на подклассы для включения всех аргументов конструктора (показано ниже).

    public static abstract class Extractor extends Mapper<Object, Text, Text, Text> {
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String feature = extract(value.toString());
            context.write(new Text(feature), new Text(value.toString()));
        }

        public abstract String extract(String s);
    }

    public static abstract class PrefixExtractor extends Extractor {
        public String extract(String s) { return s.substring(0, getEndIndex()); }

        public abstract int getEndIndex();
    }

    public static class ThreePrefixExtractor extends PrefixExtractor {
        public int getEndIndex() { return 3; }
    }

Однако это не так хорошо, и я действительно чувствую, что должен быть способэто правильный путь.

(я переместил это из первоначального вопроса, чтобы сделать вещи немного менее загроможденными.)

Answer 4

Для другого подобного решения взгляните на:

https://github.com/NICTA/scoobi/blob/master/src/main/scala/com/nicta/scoobi/impl/rtt/ClassBuilder.scala

за то, как мы это делаем. Он использует отражение для создания некоторого java-исходного кода, который при запуске создаст идентичный объектный граф. Затем мы скомпилировали этот источник (используя javassist) и включили в jar-файл, отправленный в кластер.

Это довольно надежно, если вы хотите поднять его, оно обрабатывает такие вещи, как циклические графы объектов и все особые случаи (есть немало).

Answer 5

Лучшее решение, которое я нашел до сих пор, - это передать сериализованную версию объекта, которую я хочу, в Mapper и использовать отражение для создания объекта во время выполнения.

Итак, главноеМетод скажет что-то вроде:

conf.set("ExtractorConstructor", "dicta03.hw4.PrefixExtractor(3)");

Затем в Mapper мы используем вспомогательную функцию construct (определено ниже) и можем сказать:

public void setup(Context context) {
    try {
        String constructor = context.getConfiguration().get("ExtractorConstructor");
        this.extractor = (Extractor) construct(constructor);
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

---

Определениеconstruct, который использует отражение для рекурсивного конструирования объекта во время выполнения из строки:

public static Object construct(String s) throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InstantiationException, InvocationTargetException {
    if (s.matches("^[A-Za-z0-9.#]+\\(.*\\)$")) {
        Class cls = null;
        List<Object> argList = new ArrayList<Object>();
        int parenCount = 0;
        boolean quoted = false;
        boolean escaped = false;
        int argStart = -1;
        for (int i = 0; i < s.length(); i++) {
            if (escaped) {
                escaped = false;
            } else if (s.charAt(i) == '\\') {
                escaped = true;
            } else if (s.charAt(i) == '"') {
                quoted = true;
            } else if (!quoted) {
                if (s.charAt(i) == '(') {
                    if (cls == null)
                        cls = Class.forName(s.substring(0, i));
                    parenCount++;
                    argStart = i + 1;
                } else if (s.charAt(i) == ')') {
                    if (parenCount == 1)
                        argList.add(construct(s.substring(argStart, i)));
                    parenCount--;
                } else if (s.charAt(i) == ',') {
                    if (parenCount == 1) {
                        argList.add(construct(s.substring(argStart, i)));
                        argStart = i + 1;
                    }
                }
            }
        }

        Object[] args = new Object[argList.size()];
        Class[] argTypes = new Class[argList.size()];
        for (int i = 0; i < argList.size(); i++) {
            argTypes[i] = argList.get(i).getClass();
            args[i] = argList.get(i);
        }
        Constructor constructor = cls.getConstructor(argTypes);
        return constructor.newInstance(args);
    } else if (s.matches("^\".*\"$")) {
        return s.substring(1, s.length() - 1);
    } else if (s.matches("^\\d+$")) {
        return Integer.parseInt(s);
    } else {
        throw new RuntimeException("Cannot construct " + s);
    }
}

(Возможно, это не самый надежный синтаксический анализатор, но его можно легко расширить, чтобы охватить больше типов объектов.)