Hadoop源码学习系列—— HAR归档Job设计要点

TOUVE

　　hadoop版本：CDH3u2
　　  hadoop 中生成har文件是通过mapred job实现的，这个主要的class是“HadoopArchives.java”，它是tools包里面的一个类。这个类有800多行code，包含map reduce 自定义的inputformat和其它辅助方法，细读起来还需要花点时间。

 一个har job命令行举例：

# 归档"/fc/src/2011/1[0-2]"这三个文件夹到"/fc/har/2011/"
hadoop archive -archiveName combine.har -p /fc/src/2011/ 10 11 12 /fc/har/2011/

几个重要的知识点：

1. 生成文件列表文件

 举例命令：

sudo -u hdfs hadoop archive -archiveName src.har -p /user/heipark/fc/ /user/heipark
　　下面是input path，即"/user/heipark/fc/" 目录结构（红色为文件夹）：
　　/user/heipark/fc/
　　- 0000
　　- 1111
　　- 2222
　　- 3333
　　- 4444
　　    + sub1
　　- 5555
　　        + dir1
　　- 7777
　　- 8888
　　    + sub2
　　- 6666
　　运行命令过程中生成如下文件："/user/hdfs/.staging/har_93ftj7/_har_src_files"
　　har_93ftj7：har为固定前缀，后面为随机字符串
　　_har_src_files：固定文件名，它是输入文件的meta文件，算法采用广度优先，内容如下（为了便于阅读，版式上略做调整）：

 _har_src_files文件如下：

0 / dir 0000 1111 2222 3333 4444 sub1 sub2   // ① 
1066026506 /0000 file  // ② 
1066026506 /1111 file  // ② 
38268381 /2222 file      // ② 
62016424 /3333 file      // ② 
54855349 /4444 file      // ② 

0 /sub1 dir 5555 dir1
909247 /sub1/5555 file

0 /sub1/dir1 dir 7777 8888
34193078 /sub1/dir1/7777 file
2095606385 /sub1/dir1/8888 file

0 /sub2 dir 6666
95644473 /sub2/6666 file
　　① 这行的信息表示top level path的文件夹和文件。"0"：dir的大小都为0；"/"：表示当前为输入文件的top level path；"dir"：表示当前是一个dir；"1111 2222 3333 4444 sub1 sub2"：top level path 下面的文件和文件夹
　　② 这4行表示"top level path"下面的5个文件，第一列为文件大小，第二列为文件名，第三列标识当前是一个文件

2. 如何确定map和reduce task数量

  2.1 map task数量： 

公式：

totalSize / partSize
　　    totalSize：所有输入文件总大小，这个是在递归文件列表的时候累加得来的。
　　    partSize： 默认为2GB，它是一个常量，只能通过修改代码变更。

  2.2 reduce 数量
　　reduce数量为1。reduce主要作用是汇总map输出的文件meta 信息，并生成index文件。

3. 关于"HArchiveInputFormat"
　　它的作用是读取"_har_src_files"内容（input文件列表），并根据split size（totalSize/numMapTask）生成input split。
　　  这里生成split过程比较巧妙，FileSplit是根据"_har_src_files"的偏移量分割的，简单说就是一行一行读取，并累加读取文件大小，当达到"split size"就输出一个FileSplit，最终将不足split size的文件放到一个split中，代码如下：

new FileSplit(src, startPos, size, (String[]) null)
　　src："_har_src_files"文件
　　startPos：一个split开始行的的bytes，这个偏移是_har_src_files文件的bytes偏移
　　size：一个split，在_har_src_files文件bytes偏移大小，可以这样理解：startPos + size = endPos

4. 关于Input Path
　　因为map task不直接读取input file，而是通过"_har_src_files"读取自己处理的FileSplit中文件，所以map task真正的输入路径是"_har_src_files"所在目录。

FileInputFormat.addInputPath(conf, jobDirectory);
　　jobDirectory：conf中的“har.job.dir”，就是"_har_src_files"所在文件夹。

5. 关于Map Task
　　Map Task的输入是"_har_src_files"文件中属于当前map处理的行。
　　map输出的key是当前文件或者文件夹名称的hash值，value是文件或者文件夹的meta信息。
　　如果当前行是"dir"，处理逻辑如下：

# "_har_src_files"中一个dir行
# 0 / dir 0000 1111 2222 3333 4444 sub1 sub2

①输出到reduce的信息
/ dir none 0 0 0000 1111 2222 3333 4444 sub1 sub2



　　"/"： 目前处理的路径
  "dir none 0 0"：表示文件类型是dir
  "0000 1111 2222 3333 4444 sub1 sub2"：表示"/"路径下包含的文件和路径
　　如果当前行为"file"，处理逻辑如下：

 

# "_har_src_files"文件行
# 1066026506 /0000 file

① copy 文件"0000"到har文件内"part-X"中

② 输出到reduce信息
/0000 file part-1 0 1066026506
　　"/0000"：当前处理文件
　　"file"：当前处理的是一个文件
　　"part-1"：文件"0000"输出文件名
　　"0"：文件"0000"在"part-1"中起始偏移bytes
　　"1066026506"：问"0000"文件大小

6. 关于reduce
　　reduce的工作就是生成"_masterindex"、"_index"两个索引文件。
　　先说"_index"文件，它直接汇总并输出map端output的value值，如下：

 "_index"文件：

/ dir none 0 0 0000 1111 2222 3333 4444 sub1 sub2
/0000 file part-1 0 1066026506
/1111 file part-1 1066026506 1066026506
/2222 file part-1 2132053012 38268381
/3333 file part-1 2170321393 62016424
/4444 file part-0 0 54855349
/sub1 dir none 0 0 5555 dir1
/sub2 dir none 0 0 6666
/sub1/5555 file part-0 54855349 909247
/sub1/dir1 dir none 0 0 7777 8888
/sub2/6666 file part-2 0 95644473
/sub1/dir1/7777 file part-0 55764596 34193078
/sub1/dir1/8888 file part-0 89957674 2095606385
　　再说“_masterindex” ，它是"_index"文件的索引文件。每一行索引一批文件（每批1000个，不足1000个算也算一批）它的第一行为“1”，这个值是version信息，这也许HarFileSystem开发人员打算支持对har文件的更新留下的功能。
　　reduce的输入value是经过排序的，又因为key为value中path部分的hash，所以reduce的key值是追加变大的。
　　本例"_masterindex"：

#文件version 
1 

 #startIndex endIndex startPos indexStream.getPos() 
0 771396412 0 493
　　startIndex：第一个文件startIndex(即文件path的hash）。默认值为0，处理玩一批文件后，会把endIndex赋值给startIndex。
　　     endIndex ：最后一个文件的endIndex （即文件path的hash值）
　　startPos：本批文件在"_index"文件中bytes偏移值
　　     indexStream.getPos()：本批文件在"_index"中结束bytes值
　　-- heipark