MapReduce V1：Job提交流程之JobClient端分析

01

public

static

void

main(String[] args)

throws

Exception {

02

Configuration conf =

new

Configuration();

03

String[] otherArgs =

new

GenericOptionsParser(conf, args).getRemainingArgs();

04

if

(otherArgs.length !=

2

) {

05

System.err.println(

"Usage:wordcount <in> <out>"

);

06

System.exit(

2

);

07

}

08

Job job =

new

Job(conf,

"word count"

);

09

job.setJarByClass(WordCount.

class

);

10

job.setMapperClass(TokenizerMapper.

class

);

11

job.setCombinerClass(IntSumReducer.

class

);

12

job.setReducerClass(IntSumReducer.

class

);

13

job.setOutputKeyClass(Text.

class

);

14

job.setOutputValueClass(IntWritable.

class

);

15

FileInputFormat.addInputPath(job,

new

Path(otherArgs[

0

]));

16

FileOutputFormat.setOutputPath(job,

new

Path(otherArgs[

1

]));

17

System.exit(job.waitForCompletion(

true

) ?

0

:

1

);

18

}

1

JobConf jobCopy = job;

2

Path jobStagingArea = JobSubmissionFiles.getStagingDir(JobClient.

this

, jobCopy);

// 获取到StagingArea目录

3

JobID jobId = jobSubmitClient.getNewJobId();

4

Path submitJobDir =

new

Path(jobStagingArea, jobId.toString());

1

private

String getStagingAreaDirInternal(String user)

throws

IOException {

2

final

Path stagingRootDir =

new

Path(conf.get(

"mapreduce.jobtracker.staging.root.dir"

,

"/tmp/hadoop/mapred/staging"

));

3

final

FileSystem fs = stagingRootDir.getFileSystem(conf);

4

return

fs.makeQualified(

new

Path(stagingRootDir, user+

"/.staging"

)).toString();

5

}

1

Path filesDir = JobSubmissionFiles.getJobDistCacheFiles(submitJobDir);

2

Path archivesDir = JobSubmissionFiles.getJobDistCacheArchives(submitJobDir);

3

Path libjarsDir = JobSubmissionFiles.getJobDistCacheLibjars(submitJobDir);

4

...

5

Path submitJarFile = JobSubmissionFiles.getJobJar(submitJobDir);

6

job.setJar(submitJarFile.toString());

7

fs.copyFromLocalFile(originalJarFile, submitJarFile);

1

Configuration conf =job.getConfiguration();

2

InputFormat<?, ?> input = ReflectionUtils.newInstance(job.getInputFormatClass(), conf);

3

List<InputSplit> splits = input.getSplits(job);

1

T[] array = (T[]) splits.toArray(

new

InputSplit[splits.size()]);

2

// sort the splits into order based on size, so that the biggest

3

// go first

4

Arrays.sort(array,

new

SplitComparator());

// 根据InputSplit的长度做了一个逆序排序

5

JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array);

// 将split及其元数据信息写入HDFS

01

SerializationFactory factory =

new

SerializationFactory(conf);

02

int

i =

0

;

03

long

offset = out.getPos();

04

for

(T split:array) {

05

long

prevCount = out.getPos();

06

Text.writeString(out, split.getClass().getName());

07

Serializer<T> serializer = factory.getSerializer((Class<T>) split.getClass());

08

serializer.open(out);

09

serializer.serialize(split);

// 将split序列化写入到HDFS文件中

10

long

currCount = out.getPos();

11

String[] locations = split.getLocations();

12

final

int

max_loc = conf.getInt(MAX_SPLIT_LOCATIONS,

10

);

13

if

(locations.length > max_loc) {

14

LOG.warn(

"Max block location exceeded for split:"

+ split +

" splitsize:"

+ locations.length +

" maxsize:"

+ max_loc);

15

locations = Arrays.copyOf(locations, max_loc);

16

}

17

info[i++] =

new

JobSplit.SplitMetaInfo(locations, offset, split.getLength());

// 创建SplitMetaInfo实例

18

offset += currCount - prevCount;

19

}

1

private

Path file;

2

private

long

start;

3

private

long

length;

4

private

String[] hosts;

1

@Override

2

public

void

write(DataOutput out)

throws

IOException {

3

Text.writeString(out, file.toString());

4

out.writeLong(start);

5

out.writeLong(length);

6

}

01

private

static

void

writeJobSplitMetaInfo(FileSystem fs, Path filename,

02

FsPermission p,

int

splitMetaInfoVersion,

03

JobSplit.SplitMetaInfo[] allSplitMetaInfo)

throws

IOException {

04

// write the splits meta-info to a file for the job tracker

05

FSDataOutputStream out = FileSystem.create(fs, filename, p);

06

out.write(JobSplit.META_SPLIT_FILE_HEADER);

// 写入META头信息：META_SPLIT_FILE_HEADER = "META-SPL".getBytes("UTF-8");

07

WritableUtils.writeVInt(out, splitMetaInfoVersion);

// META版本信息：1

08

WritableUtils.writeVInt(out, allSplitMetaInfo.length);

// META对象的数量：每个InputSplit对应一个SplitMetaInfo

09

for

(JobSplit.SplitMetaInfo splitMetaInfo :allSplitMetaInfo) {

10

splitMetaInfo.write(out);

// 每个都进行存储

11

}

12

out.close();

13

}

1

public

void

write(DataOutput out)

throws

IOException {

2

WritableUtils.writeVInt(out, locations.length);

// location个数

3

for

(

int

i =

0

; i < locations.length; i++) {

4

Text.writeString(out, locations[i]);

// 写入每一个location位置信息

5

}

6

WritableUtils.writeVLong(out, startOffset);

// 偏移量

7

WritableUtils.writeVLong(out, inputDataLength);

// 数据长度

8

}

1

FSDataOutputStream out = createFile(fs, JobSubmissionFiles.getJobSplitFile(jobSubmitDir), conf);

2

SplitMetaInfo[] info = writeNewSplits(conf, splits, out);

1

writeJobSplitMetaInfo(fs,JobSubmissionFiles.getJobSplitMetaFile(jobSubmitDir),

2

new

FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION), splitVersion, info);

01

...

02

Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);

03

jobCopy.set(

"mapreduce.job.dir"

, submitJobDir.toString());

04

...

05

// Write job file to JobTracker's fs

06

FSDataOutputStream out = FileSystem.create(fs, submitJobFile,

new

FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));

07

...

08

try

{

09

jobCopy.writeXml(out);

10

}

finally

{

11

out.close();

12

}