hadoop - 如何在MapReduce 框架中编写JOB1 和JOB2 之间的顺序代码？

Question

我有一个包含 2 个作业的 MapReduce 应用程序。
我需要在 Job1 和 Job2 之间运行一小段代码。换句话说，一个小代码在 Job1 的最终输出上运行，Job2 中的所有映射器都能够使用这个小代码的输出。

这段小代码不需要并行运行。它是一个顺序代码，应该在单台机器上运行并在 HDFS 中写入输出。

我想知道如何在我的应用程序代码中编写 Job1 和 Job2 之间的顺序代码，它将在一台机器上运行并读取 的输出Job1 来自 HDFS，并将其自己的输出也写入 HDFS。

Answer 1

在您的驱动程序(主)类中，在执行 Job1 之后(通常是 JobClient.runJob(conf); 命令)和设置的第一个命令之前在 Job2 上，使用 hadoop 的 FileSystem 获取 Job1 的输出并设置 Job2 的输入，如下所示:

//...run job1

BufferedWriter bw = null;
BufferedReader br = null;  
try {
  String job1outPath = "/job1/output/";
  Path job2in = new Path("/job2/in/oneFile.txt");
  bw = new BufferedWriter(new OutputStreamWriter(fs.create(job2in,true)));            
  FileSystem fs = FileSystem.get(new Configuration());
  FileStatus[] status = fs.listStatus(new Path(job1outPath));
  for (int i=0;i<status.length;i++){
     br=new BufferedReader(new InputStreamReader(fs.open(status[i].getPath())));
     String line;
     while ((line = br.readLine())!= null){
         bw.write("whatever"); //process the output of job1 as you wish
         bw.newLine();
     }
  }
} catch (IOException e) {
  System.err.println(e);
} finally {
  try{
    bw.close();
    br.close(); //not sure if you need this inside the for loop
  } catch (IOException io) {
    System.out.println(io);
  }
}

//optionally delete "/job1/output/*" here
//...run job2 using "/job2/in" as your input path

这将在您的主节点中按顺序运行。如果愿意，您还可以使用 System.out.println() 命令在您的终端中查看有用的信息。

希望它可以编译，我没有通过编辑器传递它，但我相信你可以让它工作。