Hadoopのソースを読もう

Hadoop

そんで，JobClient::submitJob() （<- ちょっとc++ぽいけど後でたどるためにこう書くことにする）は，

String jobId = jobSubmitClient.getNewJobId();

Path submitJobDir = new Path(job.getSystemDir(), jobId);

FileSystem fs = getFs();

LOG.debug("default FileSystem: " + fs.getUri());

fs.delete(submitJobDir);

Path submitJobFile = new Path(submitJobDir, "job.xml");

Path submitJarFile = new Path(submitJobDir, "job.jar");

Path submitSplitFile = new Path(submitJobDir, "job.split");



において，HDFSのファイルシステムオブジェクトをゲットして，登録されたdirectoryを基にしてファイルの置き場所を確保．その後，なぞのDistributedCacheオブジェクトからarchiveオブジェクトおよびfileオブジェクトにタイムスタンプをセット(I don't get it==igdt)．

String originalJarPath = job.getJar();

short replication = (short)job.getInt("mapred.submit.replication", 10);



if (originalJarPath != null) { // copy jar to JobTracker's fs

// use jar name if job is not named.

if ("".equals(job.getJobName())){

job.setJobName(new Path(originalJarPath).getName());

}

job.setJar(submitJarFile.toString());

fs.copyFromLocalFile(new Path(originalJarPath), submitJarFile);

fs.setReplication(submitJarFile, replication);

}



これまたわからないけど，どうやら，JobTrackerとよばれるmapreduce管理プロセスへmapやreduceのjarファイルを渡しているみたい．この時，HDFSを経由して渡そうとしているんでしょうね．ただ，なんでreplicationが10なのか不明．いっぺんに効率よくするためかなぁ．

で，そのあとユーザの指定とワーキングディレクトリの指定をしています．

// Check the input specification

job.getInputFormat().validateInput(job);



// Check the output specification

job.getOutputFormat().checkOutputSpecs(fs, job);



入力データと出力データのチェック．