запустить другую работу в Hadoop

Question

Я не понимаю, как заставить работу использовать тот же выходной каталог каталог, чтобы написать другой файл в нем. Я пробовал комментировать и комментируя эту строку, но она все еще не работает. Я получаю следующее Исключение, когда я это комментирую. Во всяком случае в коде я пытаюсь запустить два отдельные задания с одним и тем же редуктором, но с другим картографом.

РЕДАКТИРОВАТЬ: И нет, вывод одного задания не является вводом другого, причина Я хочу, чтобы они в одной папке, потому что они являются входами для еще одной карты сократить работу, которую я хочу сделать.

FileOutputFormat.setOutputPath(job, new Path(args[1]));

11/04/14 13:33:11 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
Exception in thread "main" org.apache.hadoop.mapred.InvalidJobConfException: Output directory not set.
    at org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.checkOutputSpecs(FileOutputFormat.java:120)
    at org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:770)
    at org.apache.hadoop.mapreduce.Job.submit(Job.java:432)
    at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:447)
    at org.myorg.WordCount.main(WordCount.java:123)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at org.apache.hadoop.util.RunJar.main(RunJar.java:156)

---

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        // String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        // if (otherArgs.length != 2) {
        //   System.err.println("Usage: wordcount <in> <out>");
        //   System.exit(2);
        // }
        Job job = new Job(conf, "Job1");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(Mapper1.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);

        FileSystem hdfs = FileSystem.get(conf);
        Path fromPath = new Path("/user/hadoop/output/part-r-00000");
        Path toPath = new Path("/user/hadoop/output/output1");

        // renaming to output1
        boolean isRenamed = hdfs.rename(fromPath, toPath);
        if (isRenamed)
        {
            System.out.println("Renamed to /user/hadoop/output/output1!");
        }
        else
        {
            System.out.println("Not Renamed!");
        }

        job = new Job(conf, "Job2");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(Mapper2.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // FileInputFormat.addInputPath(job, new Path(args[0]));
        // FileOutputFormat.setOutputPath(job, new Path(args[1]));


        System.exit( job.waitForCompletion(true) ? 0 : 1);
    }

добавление следующего кода в мой код вызывает другие ошибки:

    job.setInputFormatClass(FileInputFormat.class);
    job.setOutputFormatClass(FileOutputFormat.class);

    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

System.exit (job.waitForCompletion (true)? 0: 1);

Exception in thread "main" java.lang.RuntimeException: java.lang.InstantiationException
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:115)
    at org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:768)
    at org.apache.hadoop.mapreduce.Job.submit(Job.java:432)
    at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:447)
    at org.myorg.WordCount.main(WordCount.java:135)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at org.apache.hadoop.util.RunJar.main(RunJar.java:156)
Caused by: java.lang.InstantiationException
    at sun.reflect.InstantiationExceptionConstructorAccessorImpl.newInstance(InstantiationExceptionConstructorAccessorImpl.java:30)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:513)
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:113)
    ... 9 more

Answer 1

Вы должны предоставить новый объект конфигурации для второго задания. Кстати, почему вы не используете эти методы для выходного формата?

job.setInputFormatClass (FileInputFormat.class); job.setOutputFormatClass (FileOutputFormat.class);

Вот блог о повторяющихся работах, это то же самое, что вы делаете.
http://codingwiththomas.blogspot.com/2011/04/controlling-hadoop-job-recursion.html

EDIT: Кстати, что вы намереваетесь записать в папку, которая является выводом предыдущего задания, то есть вводом нового задания? Это просто приведет к другому исключению, например: «Выходной путь уже существует».

Answer 2

Все файлы не обязательно должны находиться в одном каталоге. Ваша третья работа может иметь несколько путей ввода (каталоги или файлы).

см

FileInputFormat.addInputPaths(JobConf conf, String commaSeparatedPaths)

и друзья ...