(Hadoop) MapReduce — цепочка заданий — JobControl не останавливается

Мне нужно связать два задания MapReduce. Я использовал JobControl, чтобы установить job2 как зависимый от job1. Работает, выходные файлы создаются!! Но это не останавливается! В оболочке он остается в таком состоянии:

12/09/11 19:06:24 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
12/09/11 19:06:25 INFO input.FileInputFormat: Total input paths to process : 1
12/09/11 19:06:25 INFO util.NativeCodeLoader: Loaded the native-hadoop library
12/09/11 19:06:25 WARN snappy.LoadSnappy: Snappy native library not loaded
12/09/11 19:07:00 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
12/09/11 19:07:00 INFO input.FileInputFormat: Total input paths to process : 1

Как я могу это остановить? Это мой главный.

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Configuration conf2 = new Configuration();

    Job job1 = new Job(conf, "canzoni");
    job1.setJarByClass(CanzoniOrdinate.class);
    job1.setMapperClass(CanzoniMapper.class);
    job1.setReducerClass(CanzoniReducer.class);
    job1.setOutputKeyClass(Text.class);
    job1.setOutputValueClass(IntWritable.class);

    ControlledJob cJob1 = new ControlledJob(conf);
    cJob1.setJob(job1);
    FileInputFormat.addInputPath(job1, new Path(args[0]));
    FileOutputFormat.setOutputPath(job1, new Path("/user/hduser/tmp"));


    Job job2 = new Job(conf2, "songsort");
    job2.setJarByClass(CanzoniOrdinate.class);
    job2.setMapperClass(CanzoniSorterMapper.class);
    job2.setSortComparatorClass(ReverseOrder.class);
    job2.setInputFormatClass(KeyValueTextInputFormat.class);
    job2.setReducerClass(CanzoniSorterReducer.class);
    job2.setMapOutputKeyClass(IntWritable.class);
    job2.setMapOutputValueClass(Text.class);
    job2.setOutputKeyClass(Text.class);
    job2.setOutputValueClass(IntWritable.class);

    ControlledJob cJob2 = new ControlledJob(conf2);
    cJob2.setJob(job2);
    FileInputFormat.addInputPath(job2, new Path("/user/hduser/tmp/part*"));
    FileOutputFormat.setOutputPath(job2, new Path(args[1]));

    JobControl jobctrl = new JobControl("jobctrl");
    jobctrl.addJob(cJob1);
    jobctrl.addJob(cJob2);
    cJob2.addDependingJob(cJob1);
    jobctrl.run();


    ////////////////
    // NEW CODE ///   
    //////////////


    // delete jobctrl.run();
    Thread t = new Thread(jobctrl);
    t.start();
    String oldStatusJ1 = null;
    String oldStatusJ2 = null;
    while (!jobctrl.allFinished()) {
      String status =cJob1.toString();
      String status2 =cJob2.toString();
      if (!status.equals(oldStatusJ1)) {
        System.out.println(status);
        oldStatusJ1 = status;
      }
      if (!status2.equals(oldStatusJ2)) {
        System.out.println(status2);
        oldStatusJ2 = status2;
      }     
     }
    System.exit(0);

} }


hadoop mapreduce job-control chain
person Pietro Luciani    schedule 11.09.2012    source источник
comment
Я решил это, используя Thread для запуска JobControl. Я проверил, что задания были выполнены с использованием цикла while: while(!jobctrl.allFinished()) и System.exit() вне цикла. Теперь я хотел бы, чтобы задания возвращали информационные сообщения, все, что я получил, это узнать, какое задание выполняется, с помощью ControlledJob.toString(). Я не знаю, как получить информационные сообщения, такие как: номер задачи сопоставления, номер задачи уменьшения, записи на входе или выходе и т. д. Есть идея получить эти сообщения?   -  person Pietro Luciani    schedule 12.09.2012
comment
Достаточно ли job.getCounters().toString()?   -  person zsxwing    schedule 27.05.2013
comment
Это ошибка в классе JobControl?   -  person Rags    schedule 28.08.2013
comment
Это на самом деле очень полезно, просто чтобы увидеть рабочий пример того, как объединить задания MR вместе. Нигде не нашел вразумительного объяснения по этому поводу!   -  person Austin A    schedule 04.03.2015
comment
@PietroLuciani Кроме того, для менее опытных было бы очень полезно увидеть код, который вы использовали для остановки своей работы.   -  person Austin A    schedule 08.03.2015
comment
Привет, я написал код два года назад, сегодня вечером или завтра я попытаюсь исследовать код.   -  person Pietro Luciani    schedule 18.03.2015
comment
Привет, @Austin A, я редактирую свое сообщение с кодом, запрошенным вами.   -  person Pietro Luciani    schedule 19.03.2015
comment
Спасибо @PietroLuciani, я не очень много работал с потоками, так что это действительно полезно для меня!   -  person Austin A    schedule 19.03.2015

Ответы (4)


arrow_upward
6
arrow_downward

По сути, я сделал то, на что Пьетро намекал выше.

public class JobRunner implements Runnable {
  private JobControl control;

  public JobRunner(JobControl _control) {
    this.control = _control;
  }

  public void run() {
    this.control.run();
  }
}

и в моем классе map/reduce у меня есть:

public void handleRun(JobControl control) throws InterruptedException {
    JobRunner runner = new JobRunner(control);
    Thread t = new Thread(runner);
    t.start();

    while (!control.allFinished()) {
        System.out.println("Still running...");
        Thread.sleep(5000);
    }
}

в котором я просто передаю объект jobControl.

person sinemetu1    schedule 28.06.2013

arrow_upward
3
arrow_downward

Сам объект JobControl является Runnable, поэтому вы можете просто использовать его следующим образом:

new Thread(myJobControlInstance).start()
person tombrown52    schedule 03.07.2014

arrow_upward
0
arrow_downward

Просто изменение фрагмента кода, которым поделился sinemetu1.

Вы можете отказаться от вызова JobRunner, поскольку JobControl сам по себе реализует Runnable

        Thread thread = new Thread(jobControl);
        thread.start();

        while (!jobControl.allFinished()) {
            System.out.println("Still running...");
            Thread.sleep(5000);
        }

Я также наткнулся на эту ссылку, где пользователь подтверждает, что JobControl можно запускать ТОЛЬКО с новым потоком. https://www.mail-archive.com/[email protected]/msg00556.html

person shiva    schedule 17.03.2015

arrow_upward
0
arrow_downward

попробуй это:

    Thread jcThread = new Thread(jobControl);
    jcThread.start();
    System.out.println("循环判断jobControl运行状态 >>>>>>>>>>>>>>>>");
    while (true) {
        if (jobControl.allFinished()) {
        System.out.println("====>> jobControl.allFinished=" + jobControl.getSuccessfulJobList());
        jobControl.stop();
        // 如果不加 break 或者 return，程序会一直循环
        break;
    }

    if (jobControl.getFailedJobList().size() > 0) {
        succ = 0;
        System.out.println("====>> jobControl.getFailedJobList=" + jobControl.getFailedJobList());
        jobControl.stop();

        // 如果不加 break 或者 return，程序会一直循环
        break;
    }
}
person dolphinZhang    schedule 13.07.2017