调试 Hadoop 源代码

环境是 64bit Ubuntu 14.04 系统, jdk 1.7 以及 Eclipse Mars (4.5) 　　

　　这里介绍两种调试 Hadoop 源代码的方法: 利用 Eclipse 远程调试工具和打印调试日志. 这两种方法均可以调试伪分布式工作模式和完全分布式工作模式下的 Hadoop. 最后介绍我自己的方法, 可以打印你想查看的信息( 针对单个文件内部 ).

　　(1) 利用 Eclipse 进行远程调试

　　参考 http://andilyliao.iteye.com/blog/2151688 https://www.cnblogs.com/viviman/archive/2013/01/15/2861725.html http://www.sohu.com/a/216999944_820120

下面以调试 ResourceManager 为例, 介绍利用 Eclipse 远程调试的基本方法, 这可分两步进行.

　　步骤 1 调试模式下启动 Hadoop.

　　在 Hadoop 安装目录下运行如下的 Shell 脚本:

$ export YARN_NODEMANAGER_OPTS="-Xdebug -Xrunjdwp:transport=dt_socket,address=8788,server=y,suspend=y"
$ sbin/start-all.sh       # 在 /usr/local/hadoop  目录下

　　运行了脚本后会看到 Shell 命令行终端显示如下信息:

$ Listening for transport dt_socket at address: 8788

　　此时表明 ResourceManager 处于监听状态, 直到收到 debug 确认信息.

　　步骤 2 设置断点

　　在新建的 Java 工程 "hadoop-main" 中, 找到 ResourceManager 相关代码, 并在感兴趣的地方设置一些断点.

　　步骤 3 在 Eclipse 中调试 Hadoop 程序.

　　在 Eclipse 的菜单中, 依次选择 "Run" --> "Debug Configurations" --> "Remote Java Applications", 并按照要求填写远程调试器名称(自己定义一个即可), ResourceManager 所在 host 以及监听端口号等信息, 并选择 Hadoop 源代码工程, 便可进入调试模式.

　　调试过程中, ResourceManager 输出的信息被存储到日志文件夹下的 yarn-XXX-resourcemanager-localhost.log 文件 ( XXX 为当前用户名 ) 中, 可通过以下命令查看调试过程中打印的日志:

$ tail -f logs/yarn-XXX-resourcemanager-localhost.log    # 在hadoop源代码目录下

　　(2) 打印 Hadoop 调试日志 参见 Hadoop源码编辑--日志修改篇

　　Hadoop 使用了 Apache log4j 作为基本日志库, 该日志库将日志分为5个级别, 分别是 DEBUG, INFO, WARN, ERROR 和 FATAL. 这5个级别是有顺序的, 即 DEBUG < INFO < WARN < ERROR < FATAL, 分别用来指定日志信息的重要程度. 日志输出规则为: 只输出级别不低于设定级别的日志信息, 比如若级别设定为 INFO, 则 INFO, WARN, ERROR 和 FATAL 级别的日志信息都会输出, 但级别比 INFO 低的 DEBUG 则不会输出.

　　在 Hadoop 源代码中, 大部分 Java 文件中存在调试日志 ( DEBUG 级别日志 ), 但默认情况下, 日志级别是 INFO, 为了查看更详细的运行状态, 可采用以下几种方法打开 DEBUG 日志.

　　方法 1 使用 Hadoop Shell 命令.

　　可使用 Hadoop 脚本中的 daemonlog 命令查看和修改某个类的日志级别, 比如, 可通过以下命令查看 NodeManager 类的日志级别: ( 如果你的主机是 node1, ip是192.168.1.101, 而且已经绑定了, host 写这两个中的一个即可 )

$ bin/hadoop daemonlog -getlevel ${nodemanager-host}:8042 org.apache.hadoop.yarn.server.nodemanager.NodeManager

　　可通过以下命令将 NodeManager 类的日志级别:

$ bin/hadoop daemonlog -setlevel ${nodemanager-host}:8042 org.apache.hadoop.yarn.server.nodemanager.NodeManager DEBUG

　　其中, nodemanager-host 为 NodeManager 服务所在的 host, 8042 是 NodeManager 的 HTTP 端口号.

　　方法 2 通过 Web 界面.

　　用户可以通过 Web 界面查看和修改某个类的日志级别, 比如, 可通过以下 URL 修改 NodeManager 类的日志级别:

http://${nodemanager-host}:8042/loglevel

　　方法 3 修改 log4j.properties 文件. （亲测有效） 参见 Apache log4j 官网

　　以上两种方式只能暂时修改日志级别, 当 Hadoop 重启后会被重置, 如果要永久性改变日志级别, 可在目标节点配置目录下的 log4j.properties 文件中添加以下配置选项:

$ log4j.logger.org.apache.hadoop.yarn.server.nodemanager.NodeManager=DEBUG

　　3.1 此外, 有时为了专门调试某个 Java 文件, 需要把该文件的相关日志输出到一个单独文件中, 可在 log4j.properties 中添加以下内容:

# 定义输出方式为自定义的 TTOUT
log4j.logger.org.apache.hadoop.yarn.server.nodemanager.NodeManager=DEBUG,TTOUT
# 设置 TTOUT 的输出方式为输出到文件
log4j.appender.TTOUT=org.apache.log4j.FileAppender
# 设置文件路径
log4j.appender.TTOUT.File=${hadoop.log.dir}/NodeManager.log
# 设置文件布局
log4j.appender.TTOUT.layout=org.apache.log4j.PatternLayout
# 设置文件格式
log4j.appender.TTOUT.layout.ConversionPattern=%d{ISO8601} %p %c{2}: %m%n

　　这些配置选项会把 NodeManager.java 中的 DEBUG 日志写到日志目录下的 NodeManager.log 文件中。这些对应的是NodeManager的自定义的LOG，如下所示。即 org.apache.hadoop.yarn.server.nodemanager.NodeManager 对应 NodeManager.class 。输出的日志也是它自定义的LOG的输出。

// NodeManager.java
private static final Log LOG = LogFactory.getLog(NodeManager.class);

　　在阅读源代码的过程中, 为了跟踪某个变量值的变化, 读者可能需要自己添加一些 DEBUG 日志. 在 Hadoop 源代码中, 大部分类会定义一个日志打印对象, 通过该对象可打印各个级别的日志. 比如, 在 NodeManager 中用以下代码定义对象 LOG:

public static final Log LOG = LogFactory.getLog(NodeManager.class);

　　用户可使用 LOG 对象打印调试日志. 比如, 可在 NodeManager 的 main 函数首行添加以下代码:

LOG.debug("Start to lauch NodeManager....");

　　然后重新编译 Hadoop 源代码, 并将 org.apache.hadoop.yarn.server.nodemanager.NodeManager 的调试级别修改为 DEBUG, 重新启动 Hadoop 后便可以看到该调试信息.

　　3.2 而大部分时候我们想用log4j为自己所用，输出一些自己比较关心的信息。

　　在阅读源代码的过程中, 为了跟踪某个变量值的变化, 读者可能需要自己添加一些 DEBUG 日志. 在 Hadoop 源代码中, 大部分类会定义一个日志打印对象, 通过该对象可打印各个级别的日志. 比如, 在 NodeManager 中用以下代码定义对象 LOG:

// NodeManager.java
private static final Log LOG = LogFactory.getLog(NodeManager.class); // 自带的
private static final Log LOG = LogFactory.getLog("MyNodeManager");  //自己定义的

　　用户可使用 LOG 对象打印调试日志. 比如, 可在 NodeManager 的 main 函数首行添加以下代码:

LOG.debug("Start to lauch NodeManager....");

　　然后重新编译 Hadoop 源代码,将编译好的jar包替换部署好的Hadoop集群的相应jar包之后，具体参见我的博客Hadoop 修改源码以及将修改后的源码应用到部署好的Hadoop中。 再配置 {HADOOP_HOME}/etc/hadoop/log4j.properties ,如下所示：

// log4j.properties
# 我的NodeManager Logs
log4j.logger.MyNodeManager=DEBUG,mynodemanager
#设置OUT的输出方式为输出到文件
log4j.appender.mynodemanager=org.apache.log4j.FileAppender
#设置文件路径
log4j.appender.mynodemanager.File=${hadoop.log.dir}/MyNodeManager.log
#设置文件的布局
log4j.appender.mynodemanager.layout=org.apache.log4j.PatternLayout
#设置文件的格式
log4j.appender.mynodemanager.layout.ConversionPattern=%d{ISO8601} %p %c{2}: %m%n
#设置该日志操作不与父类日志操作重叠
log4j.additivity.MyNodeManager=false

　　重新启动 Hadoop 后，便可以在我们指定的文件下看到该调试信息文件。

(3) 我自己的方法

这里需要先修改源代码. 然后重新编译 Hadoop 源代码,将编译好的jar包替换部署好的Hadoop集群的相应jar包之后，具体参见我的博客Hadoop 修改源码以及将修改后的源码应用到部署好的Hadoop中。最后重新启动 Hadoop 后, 只要执行到该类,就会显示你要查看的信息.

第一步: 先修改源代码

比如我们想要查看 DFSUtil.java 中的 locatedBlocks2Locations(List<LocatedBlock> blocks) 方法, 该方法用来创建 BlockLocation .

首先, 这是我们之前说的类和方法. DFSUtil.java 在 hadoop-2.7.3-src/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs 文件里.

// 在 org.apache.hadoop.hdfs.DFSUtil

public class DFSUtil {
// ......
/**
* Convert a List<LocatedBlock> to BlockLocation[]
* @param blocks A List<LocatedBlock> to be converted
* @return converted array of BlockLocation
*/
public static BlockLocation[] locatedBlocks2Locations(List<LocatedBlock> blocks) {
if (blocks == null) {
return new BlockLocation[0];
}
int nrBlocks = blocks.size();
BlockLocation[] blkLocations = new BlockLocation[nrBlocks];
if (nrBlocks == 0) {
return blkLocations;
}
int idx = 0;
for (LocatedBlock blk : blocks) {
assert idx < nrBlocks : "Incorrect index";
// 改为DatanodeInfoWithStorage[] 或者在调用时 ((DatanodeInfoWithStorage)location[hCnt]).getStorageType()
DatanodeInfo[] locations = blk.getLocations();
String[] hosts = new String[locations.length];
String[] xferAddrs = new String[locations.length];
String[] racks = new String[locations.length];
for (int hCnt = 0; hCnt < locations.length; hCnt++) {
hosts[hCnt] = locations[hCnt].getHostName();
xferAddrs[hCnt] = locations[hCnt].getXferAddr();
NodeBase node = new NodeBase(xferAddrs[hCnt],
locations[hCnt].getNetworkLocation());
racks[hCnt] = node.toString();
}
DatanodeInfo[] cachedLocations = blk.getCachedLocations();
String[] cachedHosts = new String[cachedLocations.length];
for (int i=0; i<cachedLocations.length; i++) {
cachedHosts[i] = cachedLocations[i].getHostName();
}
blkLocations[idx] = new BlockLocation(xferAddrs, hosts, cachedHosts,
racks,
blk.getStartOffset(),
blk.getBlockSize(),
blk.isCorrupt());
idx++;
}
return blkLocations;
}

// ......
}

我们要想知道该方法内部的一些具体信息, 就先添加我自己设计的方法, 注意, 最开始要加上包, 如下所示:

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;  // 如果有数组

/**
* 向指定的文件中写入内容; 如果是静态,则在前面添加关键字 static .
* 这里是追加写, 想重新写,先删除生成的文件, 或者把文件删除代码注释取消
* @author zhangchao
* @version 2018年1月9号 下午14:57
* @param filecontent, 要写入文件的内容 String 或 Object
*/
void writeToFile(Object filecontent){
String path = "/home/hadoop/";
String filename = "MyTest.txt";
String filenameTemp = path + filename;
String filein = filecontent + "\r\n";  //新写入的行，换行

// 如果文件不存在,创建文件.
File file=new File(filenameTemp);
try {
// 若文件存在,先删除已经存在的文件. 如果不想每次手动删除文件,则取消这一块注释
//if(file.exist()){
//  file.delete();
//}
// 若文件不存在, 创建文件.
if (!file.exists()) {
file.getParentFile().mkdirs();
file.createNewFile();
}
} catch (IOException e) {
e.printStackTrace();
}

// 向指定文件中写入文字
FileWriter fileWriter;
try {
// 打开一个写文件器，构造函数中的第二个参数true表示以追加形式写文件
fileWriter = new FileWriter(filenameTemp,true);
//使用缓冲区比不使用缓冲区效果更好，因为每趟磁盘操作都比内存操作要花费更多时间。
//通过BufferedWriter和FileWriter的连接，BufferedWriter可以暂存一堆数据，然后到满时再实际写入磁盘
//这样就可以减少对磁盘操作的次数。如果想要强制把缓冲区立即写入,只要调用writer.flush();这个方法就可以要求缓冲区马上把内容写下去
BufferedWriter bufferedWriter=new BufferedWriter(fileWriter);
bufferedWriter.write(filein);
bufferedWriter.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

最后, 把该方法添加到 DFSUtil 类中, ( 注意, 包不要和原有的重复; 并且如果是静态的方法, 需要添加 static 关键字 ), 如下所示:

// 在 org.apache.hadoop.hdfs.DFSUtil

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;  // 如果有数组

public class DFSUtil { // ......

/**
* 向指定的文件中写入内容; 如果是静态,则在前面添加关键字 static .
* 这里是追加写, 想重新写,先删除生成的文件, 或者把文件删除代码注释取消
* @author zhangchao
* @version 2018年1月9号 下午14:57
* @param filecontent, 要写入文件的内容 String 或 Object
*/
void writeToFile(Object filecontent){
String path = "/home/hadoop/hadooplogs/";  // 目录
String filename = "MyTest.txt";  // 写入数据的文件名
String filenameTemp = path + filename;
String filein = filecontent + "\r\n";  //新写入的行，换行

// 如果文件不存在,创建文件. 取消这一块注释
File file=new File(filenameTemp);
try {
// 若文件存在,先删除已经存在的文件
//if(file.exist()){
//     file.delete();
//}
// 若文件不存在, 创建文件
if (!file.exists()) {
file.getParentFile().mkdirs();
file.createNewFile();
}
} catch (IOException e) {
e.printStackTrace();
}

// 向指定文件中写入文字
FileWriter fileWriter;
try {
// 打开一个写文件器，构造函数中的第二个参数true表示以追加形式写文件
fileWriter = new FileWriter(filenameTemp,true);
//使用缓冲区比不使用缓冲区效果更好，因为每趟磁盘操作都比内存操作要花费更多时间。
//通过BufferedWriter和FileWriter的连接，BufferedWriter可以暂存一堆数据，然后到满时再实际写入磁盘
//这样就可以减少对磁盘操作的次数。如果想要强制把缓冲区立即写入,只要调用writer.flush();这个方法就可以要求缓冲区马上把内容写下去
BufferedWriter bufferedWriter=new BufferedWriter(fileWriter);
bufferedWriter.write(filein);
bufferedWriter.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

/**
* Convert a List<LocatedBlock> to BlockLocation[]
* @param blocks A List<LocatedBlock> to be converted
* @return converted array of BlockLocation
*/
public static BlockLocation[] locatedBlocks2Locations(List<LocatedBlock> blocks) {
if (blocks == null) {
return new BlockLocation[0];
}
int nrBlocks = blocks.size();
writeToFile("blocks.size() = " + nrBlocks);  // 这是我添加的, 我想知道有几个块. 最后运行集群的时候,只要调用 DFSUtil 类的该方法, 就会调用我自己设计的方法 writeToFile(), 从而完成创建文件并写入相关数据的操作.
BlockLocation[] blkLocations = new BlockLocation[nrBlocks];
if (nrBlocks == 0) {
return blkLocations;
}
int idx = 0;
for (LocatedBlock blk : blocks) {
assert idx < nrBlocks : "Incorrect index";
// 改为DatanodeInfoWithStorage[] 或者在调用时 ((DatanodeInfoWithStorage)location[hCnt]).getStorageType()
DatanodeInfo[] locations = blk.getLocations();
String[] hosts = new String[locations.length];
String[] xferAddrs = new String[locations.length];
String[] racks = new String[locations.length];
for (int hCnt = 0; hCnt < locations.length; hCnt++) {
hosts[hCnt] = locations[hCnt].getHostName();
xferAddrs[hCnt] = locations[hCnt].getXferAddr();
NodeBase node = new NodeBase(xferAddrs[hCnt],
locations[hCnt].getNetworkLocation());
racks[hCnt] = node.toString();
}
writeToFile(Arrays.asList(hosts));  // 要把数组写进文件, 需要借助 Arrays, Arrays.asList(...), 把数组转化为 List<T> , 这样就可以写入到文件.
DatanodeInfo[] cachedLocations = blk.getCachedLocations();
String[] cachedHosts = new String[cachedLocations.length];
for (int i=0; i<cachedLocations.length; i++) {
cachedHosts[i] = cachedLocations[i].getHostName();
}
blkLocations[idx] = new BlockLocation(xferAddrs, hosts, cachedHosts,
racks,
blk.getStartOffset(),
blk.getBlockSize(),
blk.isCorrupt());
idx++;
}
return blkLocations;
}

// ......
}

第二步: 编译 Hadoop 源代码,将编译好的jar包替换部署好的Hadoop集群的相应jar包

修改好代码之后, 我们知道, DFSUtil.java 在 hadoop-2.7.3-src/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs 文件里, 最深的一层包含 pom.xml ( 即可Maven ) 是 hadoop-2.7.3-src/hadoop-hdfs-project/hadoop-hdfs , 所以

// 先切换到 root 用户
su root
cd hadoop-2.7.3-src/hadoop-hdfs-project/hadoop-hdfs  // hadoop-2.7.3-src 放在哪,就从那进
mvn package -Pdist -DskipTests -Dtar

Maven 编译成功的话, 会显示:

BUILD SUCCESS

编译成功之后, 就会在 hadoop-2.7.3-src/hadoop-hdfs-project/hadoop-hdfs 文件下生成 target 文件夹, 里面存放 Maven 好的 jar 包, 这里会生成 hadoop-2.7.3-src/hadoop-hdfs-project/hadoop-hdfs/target/hadoop-hdfs-2.7.3.jar ,( 在该 Jar 包内, 我们通过压缩软件查看会发现 org/apache/hadoop/hdfs/DFSUtil.class, ) . 最后就是将该 jar 包替换到部署好的 Hadoop 的相应 jar 包, 即替换 hadoop-2.7.3/share/hadoop/hdfs/hadoop-hdfs-2.7.3.jar .

// $CLUSTER_SRC_HOME 是 hadoop-src-2.7.3 所在位置
// $HADOOP_HOME 是 hadoop-2.7.3 所在的位置

// 这里是在自己机器上单机部署的情况
cp $CLUSTER_SRC_HOME/hadoop-hdfs-project/hadoop-hdfs/target/hadoop-hdfs-2.7.3.jar $HADOOP_HOME/share/hadoop/hdfs/hadoop-hdfs-2.7.3.jar

// 如果是全分布式的,需要向集群的每台机器拷贝.
scp $CLUSTER_SRC_HOME/hadoop-hdfs-project/hadoop-hdfs/target/hadoop-hdfs-2.7.3.jar username@IP:$HADOOP_HOME/share/hadoop/hdfs/hadoop-hdfs-2.7.3.jar

第三步: 重新启动 Hadoop

只要集群调用 DFSUtil 类的 locatedBlocks2Locations(List<LocatedBlock> blocks) 方法, 就会调用该函数内部我自己设计的方法, 完成创建文件并写入信息. 实际上,运行个wordcount, 该方法会被调用, 因为集群需要创建 BlockLocation .

最后会创建 /home/hadoop/MyTest.txt , 文件内部有写入的信息.