用 Hadoop 进行分布式并行编程(2)

验

Hadoop 支持 Linux 及 Windows 操作系统, 但其官方网站声明 Hadoop 的分布式操作在 Windows 上未做严格测试，建议只把 Windows 作为 Hadoop 的开发平台。在 Windows 环境上的安装步骤如下( Linux 平台类似，且更简单一些):

(1)在 Windows 下，需要先安装 Cgywin, 安装 Cgywin 时注意一定要选择安装 openssh (在 Net category )。安装完成之后，把 Cgywin 的安装目录如 c:\cygwin\bin 加到系统环境变量 PATH 中，这是因为运行 Hadoop 要执行一些 linux 环境下的脚本和命令。

(2)安装 Java 1.5.x，并将 JAVA_HOME 环境变量设置为 Java 的安装根目录如 C:\Program Files\Java\jdk1.5.0_01。

(3)到 Hadoop 官方网站 下载Hadoop Core, 最新的稳定版本是 0.16.0. 将下载后的安装包解压到一个目录，本文假定解压到 c:\hadoop-0.16.0。

4)修改 conf/hadoop-env.sh 文件，在其中设置 JAVA_HOME 环境变量： export JAVA_HOME="C:\Program Files\Java\jdk1.5.0_01” (因为路径中 Program Files 中间有空格，一定要用双引号将路径引起来)

至此，一切就绪，可以运行 Hadoop 了。以下的运行过程，需要启动 cygwin, 进入模拟 Linux 环境。在下载的 Hadoop Core 包中，带有几个示例程序并且已经打包成了 hadoop-0.16.0-examples.jar。其中有一个 WordCount 程序，功能是统计一批文本文件中各个单词出现的次数，我们先来看看怎么运行这个程序。Hadoop 共有三种运行模式: 单机(非分布式)模式，伪分布式运行模式，分布式运行模式，其中前两种运行模式体现不了 Hadoop 分布式计算的优势，并没有什么实际意义，但对程序的测试及调试很有帮助，我们先从这两种模式入手，了解基于 Hadoop 的分布式并行程序是如何编写和运行的。

单机(非分布式)模式

这种模式在一台单机上运行，没有分布式文件系统，而是直接读写本地操作系统的文件系统。

码清单1

               

$ cd /cygdrive/c/hadoop-0.16.0

$ mkdir test-in 

$ cd test-in

#在 test-in 目录下创建两个文本文件, WordCount 程序将统计其中各个单词出现次数

$ echo "hello world bye world" >file1.txt  

$ echo "hello hadoop goodbye hadoop" >file2.txt

$ cd ..

$ bin/hadoop jar hadoop-0.16.0-examples.jar wordcount test-in test-out

#执行完毕，下面查看执行结果：

$ cd test-out

$ cat part-00000

bye     1

goodbye  1

hadoop  2

hello    2

world   2

 

注意事项：运行 bin/hadoop jar hadoop-0.16.0-examples.jar wordcount test-in test-out 时，务必注意第一个参数是 jar, 不是 -jar, 当你用 -jar 时，不会告诉你是参数错了，报告出来的错误信息是：Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/hadoop/util/ProgramDriver, 笔者当时以为是 classpath 的设置问题，浪费了不少时间。通过分析 bin/hadoop 脚本可知，-jar 并不是 bin/hadoop 脚本定义的参数，此脚本会把 -jar 作为 Java 的参数，Java 的-jar 参数表示执行一个 Jar 文件(这个 Jar 文件必须是一个可执行的 Jar,即在 MANIFEST 中定义了主类), 此时外部定义的 classpath 是不起作用的，因而会抛出 java.lang.NoClassDefFoundError 异常。而 jar 是 bin/hadoop 脚本定义的参数，会调用 Hadoop 自己的一个工具类 RunJar，这个工具类也能够执行一个 Jar 文件，并且外部定义的 classpath 有效。

伪分布式运行模式

这 种模式也是在一台单机上运行，但用不同的 Java 进程模仿分布式运行中的各类结点 ( NameNode, DataNode, JobTracker, TaskTracker, Secondary NameNode )，请注意分布式运行中的这几个结点的区别：

从分布式存储的角度来说，集群中的结点由一个 NameNode 和若干个 DataNode 组成, 另有一个 Secondary NameNode 作为 NameNode 的备份。 从分布式应用的角度来说，集群中的结点由一个 JobTracker 和若干个 TaskTracker 组成，JobTracker 负责任务的调度，TaskTracker 负责并行执行任务。TaskTracker 必须运行在 DataNode 上，这样便于数据的本地计算。JobTracker 和 NameNode 则无须在同一台机器上。

(1) 按代码清单2修改 conf/hadoop-site.xml。注意 conf/hadoop-default.xml 中是 Hadoop 缺省的参数，你可以通过读此文件了解 Hadoop 中有哪些参数可供配置，但不要修改此文件。可通过修改 conf/hadoop-site.xml 改变缺省参数值，此文件中设置的参数值会覆盖 conf/hadoop-default.xml 的同名参数。

码清单 2

               

 

    fs.default.name

    localhost:9000

 

 

    mapred.job.tracker

    localhost:9001

 

 

    dfs.replication

    1

 

 

参数 fs.default.name 指定 NameNode 的 IP 地址和端口号。缺省值是 file:///, 表示使用本地文件系统, 用于单机非分布式模式。此处我们指定使用运行于本机 localhost 上的 NameNode。

参数 mapred.job.tracker 指定 JobTracker 的 IP 地址和端口号。缺省值是 local, 表示在本地同一 Java 进程内执行 JobTracker 和 TaskTracker, 用于单机非分布式模式。此处我们指定使用运行于本机 localhost 上的 JobTracker ( 用一个单独的 Java 进程做 JobTracker )。

参数 dfs.replication 指定 HDFS 中每个 Block 被复制的次数，起数据冗余备份的作用。 在典型的生产系统中，这个数常常设置为3。

(2)配置 SSH,如代码清单3所示:

码清单 3

               

$ ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa

$ cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

 

配置完后，执行一下 ssh localhost, 确认你的机器可以用 SSH 连接，并且连接时不需要手工输入密码。

(3)格式化一个新的分布式文件系统, 如代码清单4所示:

码清单 4

               

$ cd /cygdrive/c/hadoop-0.16.0

$ bin/hadoop namenode –format

 

(4) 启动 hadoop 进程, 如代码清单5所示。控制台上的输出信息应该显示启动了 namenode, datanode, secondary namenode, jobtracker, tasktracker。启动完成之后，通过 ps –ef 应该可以看到启动了5个新的 java 进程。

码清单 5

               

$ bin/start-all.sh 

$ ps –ef

 

(5) 运行 wordcount 应用, 如代码清单6所示:

码清单 6

               

$ bin/hadoop dfs -put ./test-in input 

#将本地文件系统上的 ./test-in 目录拷到 HDFS 的根目录上，目录名改为 input

#执行 bin/hadoop dfs –help 可以学习各种 HDFS 命令的使用。

$ bin/hadoop jar hadoop-0.16.0-examples.jar wordcount input output

#查看执行结果:

#将文件从 HDFS 拷到本地文件系统中再查看：

$ bin/hadoop dfs -get output output

$ cat output/*

#也可以直接查看

$ bin/hadoop dfs -cat output/*

$ bin/stop-all.sh #停止 hadoop 进程

 

故障诊断

(1) 执行 $ bin/start-all.sh 启动 Hadoop 进程后，会启动5个 java 进程, 同时会在 /tmp 目录下创建五个 pid 文件记录这些进程 ID 号。通过这五个文件，可以得知 namenode, datanode, secondary namenode, jobtracker, tasktracker 分别对应于哪一个 Java 进程。当你觉得 Hadoop 工作不正常时，可以首先查看这5个 java 进程是否在正常运行。

(2) 使用 web 接口。访问 可以查看 JobTracker 的运行状态。访问 可以查看 TaskTracker 的运行状态。访问 可以查看 NameNode 以及整个分布式文件系统的状态，浏览分布式文件系统中的文件以及 log 等。

(3) 查看 ${HADOOP_HOME}/logs 目录下的 log 文件，namenode, datanode, secondary namenode, jobtracker, tasktracker 各有一个对应的 log 文件，每一次运行的计算任务也有对应用 log 文件。分析这些 log 文件有助于找到故障原因。

束语

现 在，你已经了解了 MapReduce 计算模型，分布式文件系统 HDFS，分布式并行计算等的基本原理, 并且有了一个可以运行的 Hadoop 环境，运行了一个基于 Hadoop 的并行程序。在下一篇文章中，你将了解到如何针对一个具体的计算任务，基于 Hadoop 编写自己的分布式并行程序并将其部署运行等内容。

 

第 2 部分

程序实例与分析

我们先来看看 Hadoop 自带的示例程序 WordCount，这个程序用于统计一批文本文件中单词出现的频率，完整的代码可在下载的 Hadoop 安装包中得到(在 src/examples 目录中)。

1.实现Map类

见代码清单1。这个类实现 Mapper 接口中的 map 方法，输入参数中的 value 是文本文件中的一行，利用 StringTokenizer 将这个字符串拆成单词，然后将输出结果 <单词,1> 写入到 org.apache.hadoop.mapred.OutputCollector 中。OutputCollector 由 Hadoop 框架提供, 负责收集 Mapper 和 Reducer 的输出数据，实现 map 函数和 reduce 函数时，只需要简单地将其输出的 对往 OutputCollector 中一丢即可，剩余的事框架自会帮你处理好。

代码中 LongWritable, IntWritable, Text 均是 Hadoop 中实现的用于封装 Java 数据类型的类，这些类都能够被串行化从而便于在分布式环境中进行数据交换，你可以将它们分别视为 long, int, String 的替代品。Reporter 则可用于报告整个应用的运行进度，本例中未使用。