hadoop - HadoopでJobControlを使用する方法

Question

2つのファイルを1つにマージしたい。読むために2つのマッパーを作成し、参加するために1つのレデューサーを作成しました。

        JobConf classifiedConf = new JobConf(new Configuration());
            classifiedConf.setJarByClass(myjob.class);
    classifiedConf.setJobName("classifiedjob");
    FileInputFormat.setInputPaths(classifiedConf,classifiedInputPath );
    classifiedConf.setMapperClass(ClassifiedMapper.class);
    classifiedConf.setMapOutputKeyClass(TextPair.class);
    classifiedConf.setMapOutputValueClass(Text.class);
    Job classifiedJob = new Job(classifiedConf);
    //first mapper config

    JobConf featureConf = new JobConf(new Configuration());
    featureConf.setJobName("featureJob");
            featureConf.setJarByClass(myjob.class);
    FileInputFormat.setInputPaths(featureConf, featuresInputPath);
    featureConf.setMapperClass(FeatureMapper.class);
    featureConf.setMapOutputKeyClass(TextPair.class);
    featureConf.setMapOutputValueClass(Text.class);
    Job featureJob = new Job(featureConf);
            //second mapper config

    JobConf joinConf = new JobConf(new Configuration());
    joinConf.setJobName("joinJob");
            joinConf.setJarByClass(myjob.class);
    joinConf.setReducerClass(JoinReducer.class);
    joinConf.setOutputKeyClass(Text.class);
    joinConf.setOutputValueClass(Text.class);
    Job joinJob = new Job(joinConf);
             //reducer config
             //JobControl config
            joinJob.addDependingJob(featureJob);
    joinJob.addDependingJob(classifiedJob);
    secondJob.addDependingJob(joinJob);
    JobControl jobControl = new JobControl("jobControl");
    jobControl.addJob(classifiedJob);
    jobControl.addJob(featureJob);
    jobControl.addJob(secondJob);

    Thread thread = new Thread(jobControl);
    thread.start();
    while(jobControl.allFinished()){
        jobControl.stop();
    }

しかし、私はこのメッセージを受け取ります：警告mapred.JobClient：

Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
WARN mapred.JobClient: No job jar file set.  User classes may not be found. See JobConf(Class) or JobConf#setJar(String).

誰か助けてください..................

Answer 1

どのバージョンの Hadoop を使用していますか?

あなたが得る警告はプログラムを停止しますか?

setJarByClass() を使用する必要はありません。私のスニペットを見ることができます。 setJarByClass() メソッドを使用せずに実行できます。

JobConf job = new JobConf(PageRankJob.class);
job.setJobName("PageRankJob");

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

job.setMapperClass(PageRankMapper.class);
job.setReducerClass(PageRankReducer.class);

job.setInputFormat(TextInputFormat.class);
job.setOutputFormat(TextOutputFormat.class);

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

JobClient.runJob(job);

Answer 2

この方法で Job を実装する必要があります。

public class MyApp extends Configured implements Tool {

    public int run(String[] args) throws Exception {
      // Configuration processed by ToolRunner
      Configuration conf = getConf();

      // Create a JobConf using the processed conf
      JobConf job = new JobConf(conf, MyApp.class);

      // Process custom command-line options
      Path in = new Path(args[1]);
      Path out = new Path(args[2]);

      // Specify various job-specific parameters     
      job.setJobName("my-app");
      job.setInputPath(in);
      job.setOutputPath(out);
      job.setMapperClass(MyMapper.class);
      job.setReducerClass(MyReducer.class);

      // Submit the job, then poll for progress until the job is complete
      JobClient.runJob(job);
      return 0;
    }

    public static void main(String[] args) throws Exception {
      // Let ToolRunner handle generic command-line options 
      int res = ToolRunner.run(new Configuration(), new MyApp(), args);

      System.exit(res);
    }
}

これは、Hadoop のドキュメントhereから直接引用したものです。

したがって、基本的にあなたの仕事はから継承しConfiguredて実装する必要がありますTool。これにより、実装が強制されrun()ます。次に、を使用してメインクラスからジョブを開始するToolrunner.run(<your job>, <args>)と、警告が消えます。

Answer 3

ドライバーにこのコードが必要ですjob.setJarByClass(MapperClassName.class);