HDFS源码分析数据块之CorruptReplicasMap

CorruptReplicasMap用于存储文件系统中所有损坏数据块的信息。仅当它的所有副本损坏时一个数据块才被认定为损坏。当汇报数据块的副本时，我们隐藏所有损坏副本。一旦一个数据块被发现完好副本达到预期，它将从CorruptReplicasMap中被移除。

我们先看下CorruptReplicasMap都有哪些成员变量，如下所示：

1. // 存储损坏数据块Block与它对应每个数据节点与损坏原因集合映射关系的集合
2. private final SortedMap<Block, Map<DatanodeDescriptor, Reason>> corruptReplicasMap =
3. new TreeMap<Block, Map<DatanodeDescriptor, Reason>>();

是的，你没看错，就这一个corruptReplicasMap集合，它是一个用来存储损坏数据块Block实例与它对应每个数据节点和损坏原因集合的映射关系的集合。

我们看下CorruptReplicasMap都提供了哪些有用的方法。

一、addToCorruptReplicasMap()

标记属于指定数据节点的数据块为损坏

1. /**
2. * Mark the block belonging to datanode as corrupt.
3. * 标记属于指定数据节点的数据块为损坏
4. *
5. * @param blk Block to be added to CorruptReplicasMap
6. * @param dn DatanodeDescriptor which holds the corrupt replica
7. * @param reason a textual reason (for logging purposes)
8. * @param reasonCode the enum representation of the reason
9. */
10. void addToCorruptReplicasMap(Block blk, DatanodeDescriptor dn,
11. String reason, Reason reasonCode) {
12. / 先从corruptReplicasMap集合中查找是否存在对应数据块blk
13. Map <DatanodeDescriptor, Reason> nodes = corruptReplicasMap.get(blk);
14. // 如果不存在，构造一个HashMap<DatanodeDescriptor, Reason>集合nodes，将blk与nodes存入corruptReplicasMap
15. if (nodes == null) {
16. nodes = new HashMap<DatanodeDescriptor, Reason>();
17. corruptReplicasMap.put(blk, nodes);
18. }
19. String reasonText;
20. if (reason != null) {
21. reasonText = ” because ” + reason;
22. } else {
23. reasonText = “”;
24. }
25. // 判断nodes中是否存在对应数据节点dn，分别记录日志信息
26. if (!nodes.keySet().contains(dn)) {
27. NameNode.blockStateChangeLog.info(“BLOCK NameSystem.addToCorruptReplicasMap: “+
28. blk.getBlockName() +
29. ” added as corrupt on ” + dn +
30. ” by ” + Server.getRemoteIp() +
31. reasonText);
32. } else {
33. NameNode.blockStateChangeLog.info(“BLOCK NameSystem.addToCorruptReplicasMap: “+
34. “duplicate requested for ” +
35. blk.getBlockName() + ” to add as corrupt ” +
36. “on ” + dn +
37. ” by ” + Server.getRemoteIp() +
38. reasonText);
39. }
40. // Add the node or update the reason.
41. // 将数据节点dn、损坏原因编码reasonCode加入或更新入nodes
42. nodes.put(dn, reasonCode);
43. }

处理逻辑很简单，大体如下：

1、先从corruptReplicasMap集合中查找是否存在对应数据块blk；

2、如果不存在，构造一个HashMap<DatanodeDescriptor, Reason>集合nodes，将blk与nodes存入corruptReplicasMap；

3、判断nodes中是否存在对应数据节点dn，分别记录日志信息；

4、将数据节点dn、损坏原因编码reasonCode加入或更新入nodes。

二、removeFromCorruptReplicasMap()

将指定数据块、数据节点，根据指定原因从集合corruptReplicasMap移除

1. // 将指定数据块、数据节点，根据指定原因从集合corruptReplicasMap移除
2. boolean removeFromCorruptReplicasMap(Block blk, DatanodeDescriptor datanode,
3. Reason reason) {
4. / 先从corruptReplicasMap集合中查找是否存在对应数据块blk，获得datanodes
5. Map <DatanodeDescriptor, Reason> datanodes = corruptReplicasMap.get(blk);
6. // 如果不存在，直接返回false，表明移除失败
7. if (datanodes==null)
8. return false;
9. // if reasons can be compared but don’t match, return false.
10. // 取出数据节点datanode对应的存储损坏原因storedReason
11. Reason storedReason = datanodes.get(datanode);
12. // 判断存储损坏原因storedReason与参数损坏原因reason是否一致，不一致直接返回false，表明移除失败，
13. // 判断的依据为参数损坏原因reason不是ANY且存储损坏原因storedReason不为空的情况下，两者不一致
14. if (reason != Reason.ANY && storedReason != null &&
15. reason != storedReason) {
16. return false;
17. }
18. // 将datanode对应数据从datanodes中移除
19. if (datanodes.remove(datanode) != null) { // remove the replicas
20. // 移除datanode后，如果datanodes为空
21. if (datanodes.isEmpty()) {
22. // remove the block if there is no more corrupted replicas
23. // 将数据块blk从集合corruptReplicasMap中移除
24. corruptReplicasMap.remove(blk);
25. }
26. // 返回true，表明移除成功
27. return true;
28. }
29. // 其他情况下直接返回false，表明移除失败
30. return false;
31. }

三、getCorruptReplicaBlockIds()

获取指定大小和起始数据块ID的损坏数据块ID数组

1. /**
2. * Return a range of corrupt replica block ids. Up to numExpectedBlocks
3. * blocks starting at the next block after startingBlockId are returned
4. * (fewer if numExpectedBlocks blocks are unavailable). If startingBlockId
5. * is null, up to numExpectedBlocks blocks are returned from the beginning.
6. * If startingBlockId cannot be found, null is returned.
7. * 获取指定大小和起始数据块ID的损坏数据块ID数组
8. *
9. * @param numExpectedBlocks Number of block ids to return.
10. *  0 <= numExpectedBlocks <= 100
11. * @param startingBlockId Block id from which to start. If null, start at
12. *  beginning.
13. * @return Up to numExpectedBlocks blocks from startingBlockId if it exists
14. *
15. */
16. long[] getCorruptReplicaBlockIds(int numExpectedBlocks,
17. Long startingBlockId) {
18. / 校验numExpectedBlocks，需要获取的数据块ID数组最多有100个元素
19. if (numExpectedBlocks < 0 || numExpectedBlocks > 100) {
20. return null;
21. }
22. // 获得corruptReplicasMap集合的数据块迭代器blockIt
23. Iterator<Block> blockIt = corruptReplicasMap.keySet().iterator();
24. // if the starting block id was specified, iterate over keys until
25. // we find the matching block. If we find a matching block, break
26. // to leave the iterator on the next block after the specified block.
27. // 如果设定了起始数据块艾迪startingBlockId
28. if (startingBlockId != null) {
29. boolean isBlockFound = false;
30. // 遍历corruptReplicasMap，查看是否存在startingBlockId，如果存在，跳出循环，此时已记录住迭代器的位置了
31. while (blockIt.hasNext()) {
32. Block b = blockIt.next();
33. if (b.getBlockId() == startingBlockId) {
34. isBlockFound = true;
35. break;
36. }
37. }
38. // 如果不存在，直接返回null
39. if (!isBlockFound) {
40. return null;
41. }
42. }
43. // 构造一个存储数据块ID的列表corruptReplicaBlockIds
44. ArrayList<Long> corruptReplicaBlockIds = new ArrayList<Long>();
45. // append up to numExpectedBlocks blockIds to our list
46. // 遍历corruptReplicasMap，将最多numExpectedBlocks个数据块ID添加到列表corruptReplicaBlockIds，
47. // 此时的迭代器可能不是从头开始取数据的，在startingBlockId需要并存在的情况下，它是从下一个元素开始获取的
48. for(int i=0; i<numExpectedBlocks && blockIt.hasNext(); i++) {
49. corruptReplicaBlockIds.add(blockIt.next().getBlockId());
50. }
51. // 将数据块ID列表corruptReplicaBlockIds转换成数组ret
52. long[] ret = new long[corruptReplicaBlockIds.size()];
53. for(int i=0; i<ret.length; i++) {
54. ret[i] = corruptReplicaBlockIds.get(i);
55. }
56. // 返回数据块ID数组ret
57. return ret;
58. }

四、getNodes()

根据损坏数据块获取对应数据节点集合

1. /**
2. * Get Nodes which have corrupt replicas of Block
3. * 根据损坏数据块获取对应数据节点集合
4. *
5. * @param blk Block for which nodes are requested
6. * @return collection of nodes. Null if does not exists
7. */
8. Collection<DatanodeDescriptor> getNodes(Block blk) {
9. Map <DatanodeDescriptor, Reason> nodes = corruptReplicasMap.get(blk);
10. if (nodes == null)
11. return null;
12. return nodes.keySet();
13. }

五、isReplicaCorrupt()

检测指定数据块和数据节点是否为损坏的

1. /**
2. * Check if replica belonging to Datanode is corrupt
3. * 检测指定数据块和数据节点是否为损坏的
4. *
5. * @param blk Block to check
6. * @param node DatanodeDescriptor which holds the replica
7. * @return true if replica is corrupt, false if does not exists in this map
8. */
9. boolean isReplicaCorrupt(Block blk, DatanodeDescriptor node) {
10. Collection<DatanodeDescriptor> nodes = getNodes(blk);
11. return ((nodes != null) && (nodes.contains(node)));
12. }

六、numCorruptReplicas()

获取给定数据块对应数据节点数量

1. // 获取给定数据块对应数据节点数量
2. int numCorruptReplicas(Block blk) {
3. Collection<DatanodeDescriptor> nodes = getNodes(blk);
4. return (nodes == null) ? 0 : nodes.size();
5. }

七、size()

获取损坏数据块数量

1. // 获取损坏数据块数量
2. int size() {
3. return corruptReplicasMap.size();
4. }