使用Cygwin模拟Linux环境安装配置运行基于单机的Hadoop

使用Cygwin模拟Linux环境安装配置运行基于单机的Hadoop

其实，使用Cygwin模拟Linux环境来运行Hadoop是非常轻松的，只需要简单地配置一下就可以运行基于单机的Hadoop。

这里，比较关键的就是Cygwin的安装，在选择安装的时候一定要安装openssh，否则不会成功的，下面简单说一下Cygwin的安装及其配置：

Cygwin的下载安装 首先点击http:///setup.exe下载setup.exe，例如保存到桌面，点击就可以进行下载安装了。 在选择安装类型的时候，最好是选择第一个，直接从网络上下载并紧接着安装，如图所示： 然后选择安装路径、安装文件存放路径、连接方式(这里选择Use IE5 Settings)、下载站点镜像，自动创建下载文件列表，接下来一步比较重要了：选择安装类型，可以单击最顶层的All后面的循环样式图标切换安装类型，是的最顶层All行的最后一个单词为Install，如图所示： 其实，如果你选择了Install安装类型，就已经选择了openssh包。 为让你看到openssh包，你可以在Net [图标] Install 下面看到与网络有关的包，如图所示： 向下滑动滚动条，可以看到openssh，如图所示： 在Cirrent下如果显示版本号，说明该包已经被此次安装选择上了，否则的话会显示一个Skip，意思是跳过该包，并不会安装该包的。 最后就等着下载安装了，这个过程可能会花费一点时间的。 Cygwin的配置 安装完成之后，例如我的Cygwin安装在G:\Cygwin\目录下面，进行配置如下： 设置环境变量： 在系统变量中新建变量【变量名：CYGWIN，变量值：ntsec tty】；编辑添加变量【变量名：Path，变量值：G:\Cygwin\bin;其它的保留】。 OK，基本配置好了，可以配置Hadoop了。

Hadoop目前的有几个版本：hadoop-0.16.4、hadoop-0.18.0，到Apache下载一个并解压缩即可。

将解压缩的Hadoop放到G盘下，例如我的是：G:\hadoop-0.16.4。

配置Hadoop只需要修改G:\hadoop-0.16.4\conf目录下的hadoop-env.sh文件即可，打开它你可以看到：

# The java implementation to use. Required. # export JAVA_HOME=/usr/lib/j2sdk1.5-sun

将第二行的注释符号去掉，同时指定在你的机器上JAVA_HOME的值，如下为我修改的内容：

# The java implementation to use. Required. export JAVA_HOME="D:\Program Files\Java\jdk1.6.0_07"

这里要注意，如果你的JDK安装目录中存在空格，需要使用双引号引起来，否则就会报错。

启动Cygwin，当前它是在home/yourname目录下的，如图所示：

切换到根目录下，从而进入G:\hadoop-0.16.4目录，并创建一个数据输入目录input-dir，如图所示：

下面，打开G:\hadoop-0.16.4\input-dir，在该目录下新建几个记事本文件，例如我创建了3个：input-a.txt、input-b.txt、input-c.txt。三个文件的内容如下所示：

input-a.txt：as after append actor as as add as apache as after add as input-b.txt：bench be bench believe background bench be block input-c.txt：cafe cat communications connection cat cat cat cust cafe

接着就可以执行Hadoop自带的一个统计英文单词出现频率的例子了，直接输入命令bin/hadoop jar hadoop-0.16.4-examples.jar wordcount input-dir output-dir，其中hadoop-0.16.4-examples.jar是G:\hadoop-0.16.4包中的例子，input-dir是数据输入目录，在里面已经存在我们创建的三个文件了，output-dir是经过Hadoop处理后的输出结果的目录，这里，你需要对Google的MapReduce算法有一个简单的了解，主要就是在处理数据的时候是怎样的一个流程，引用IBM上的一片技术文章片段了解一下：

引 用 用 MapReduce 来处理大数据集的过程, 这个 MapReduce 的计算过程简而言之，就是将大数据集分解为成百上千的小数据集，每个(或若干个)数据集分别由集群中的一个结点(一般就是一台普通的计算机)进行处理并生成中间结果，然后这些中间结果又由大量的结点进行合并, 形成最终结果。 计算模型的核心是 Map 和 Reduce 两个函数，这两个函数由用户负责实现，功能是按一定的映射规则将输入的 <key, value> 对转换成另一个或一批 <key, value> 对输出。 表一 Map 和 Reduce 函数 函数 输入 输出 说明 Map <k1, v1> List(<k2,v2>) 1. 将小数据集进一步解析成一批 <key,value> 对，输入 Map 函数中进行处理。 2. 每一个输入的 <k1,v1> 会输出一批 <k2,v2>。 <k2,v2> 是计算的中间结果。 Reduce <k2,List(v2)> <k3,v3> 输入的中间结果 <k2,List(v2)> 中的 List(v2) 表示是一批属于同一个 k2 的 value 基于 MapReduce 计算模型编写分布式并行程序非常简单，程序员的主要编码工作就是实现 Map 和 Reduce 函数，其它的并行编程中的种种复杂问题，如分布式存储，工作调度，负载平衡，容错处理，网络通信等，均由 MapReduce 框架(比如 Hadoop )负责处理，程序员完全不用操心。 本地计算 数据存储在哪一台计算机上，就由这台计算机进行这部分数据的计算，这样可以减少数据在网络上的传输，降低对网络带宽的需求。在 Hadoop 这样的基于集群的分布式并行系统中，计算结点可以很方便地扩充，而因它所能够提供的计算能力近乎是无限的，但是由是数据需要在不同的计算机之间流动，故网络带宽变成了瓶颈，是非常宝贵的，“本地计算”是最有效的一种节约网络带宽的手段，业界把这形容为“移动计算比移动数据更经济”。 任务粒度 把原始大数据集切割成小数据集时，通常让小数据集小于或等于 HDFS 中一个 Block 的大小(缺省是 64M)，这样能够保证一个小数据集位于一台计算机上，便于本地计算。有 M 个小数据集待处理，就启动 M 个 Map 任务，注意这 M 个 Map 任务分布于 N 台计算机上并行运行，Reduce 任务的数量 R 则可由用户指定。 Partition 把 Map 任务输出的中间结果按 key 的范围划分成 R 份( R 是预先定义的 Reduce 任务的个数)，划分时通常使用 hash 函数如: hash(key) mod R，这样可以保证某一段范围内的 key，一定是由一个 Reduce 任务来处理，可以简化 Reduce 的过程。 Combine 在 partition 之前，还可以对中间结果先做 combine，即将中间结果中有相同 key的 <key, value> 对合并成一对。combine 的过程与 Reduce 的过程类似，很多情况下就可以直接使用 Reduce 函数，但 combine 是作为 Map 任务的一部分，在执行完 Map 函数后紧接着执行的。Combine 能够减少中间结果中 <key, value> 对的数目，从而减少网络流量。 Reduce 任务从 Map 任务结点取中间结果 Map 任务的中间结果在做完 Combine 和 Partition 之后，以文件形式存于本地磁盘。中间结果文件的位置会通知主控 JobTracker, JobTracker 再通知 Reduce 任务到哪一个 DataNode 上去取中间结果。注意所有的 Map 任务产生中间结果均按其 Key 用同一个 Hash 函数划分成了 R 份，R 个 Reduce 任务各自负责一段 Key 区间。每个 Reduce 需要向许多个 Map 任务结点取得落在其负责的 Key 区间内的中间结果，然后执行 Reduce 函数，形成一个最终的结果文件。 任务管道 有 R 个 Reduce 任务，就会有 R 个最终结果，很多情况下这 R 个最终结果并不需要合并成一个最终结果。因为这 R 个最终结果又可以做为另一个计算任务的输入，开始另一个并行计算任务。

执行过程如下所示：

SHIYANJUN@cbbd2ce9428e48b /cygdrive/g/hadoop-0.16.4 $ bin/hadoop jar hadoop-0.16.4-examples.jar wordcount input-dir output-dir cygpath: cannot create short name of g:\hadoop-0.16.4\logs 08/09/12 20:42:50 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName =JobTracker, sessionId= 08/09/12 20:42:51 INFO mapred.FileInputFormat: Total input paths to process : 3 08/09/12 20:42:51 INFO mapred.JobClient: Running job: job_local_1 08/09/12 20:42:51 INFO mapred.MapTask: numReduceTasks: 1 08/09/12 20:42:52 INFO mapred.JobClient: map 0% reduce 0% 08/09/12 20:42:52 INFO mapred.LocalJobRunner: file:/g:/hadoop-0.16.4/input-dir/i nput-a.txt:0+57 08/09/12 20:42:52 INFO mapred.TaskRunner: Task 'job_local_1_map_0000' done. 08/09/12 20:42:52 INFO mapred.TaskRunner: Saved output of task 'job_local_1_map_ 0000' to file:/g:/hadoop-0.16.4/output-dir 08/09/12 20:42:53 INFO mapred.MapTask: numReduceTasks: 1 08/09/12 20:42:53 INFO mapred.LocalJobRunner: file:/g:/hadoop-0.16.4/input-dir/i nput-b.txt:0+48 08/09/12 20:42:53 INFO mapred.TaskRunner: Task 'job_local_1_map_0001' done. 08/09/12 20:42:53 INFO mapred.TaskRunner: Saved output of task 'job_local_1_map_ 0001' to file:/g:/hadoop-0.16.4/output-dir 08/09/12 20:42:53 INFO mapred.MapTask: numReduceTasks: 1 08/09/12 20:42:53 INFO mapred.LocalJobRunner: file:/g:/hadoop-0.16.4/input-dir/i nput-c.txt:0+56 08/09/12 20:42:53 INFO mapred.TaskRunner: Task 'job_local_1_map_0002' done. 08/09/12 20:42:53 INFO mapred.TaskRunner: Saved output of task 'job_local_1_map_ 0002' to file:/g:/hadoop-0.16.4/output-dir 08/09/12 20:42:53 INFO mapred.JobClient: map 100% reduce 0% 08/09/12 20:42:54 INFO mapred.LocalJobRunner: reduce > reduce 08/09/12 20:42:54 INFO mapred.TaskRunner: Task 'reduce_z7f1uq' done. 08/09/12 20:42:54 INFO mapred.TaskRunner: Saved output of task 'reduce_z7f1uq' t o file:/g:/hadoop-0.16.4/output-dir 08/09/12 20:42:54 INFO mapred.JobClient: Job complete: job_local_1 08/09/12 20:42:54 INFO mapred.JobClient: Counters: 9 08/09/12 20:42:54 INFO mapred.JobClient:   Map-Reduce Framework 08/09/12 20:42:54 INFO mapred.JobClient:     Map input records=3 08/09/12 20:42:54 INFO mapred.JobClient:     Map output records=30 08/09/12 20:42:54 INFO mapred.JobClient:     Map input bytes=161 08/09/12 20:42:54 INFO mapred.JobClient:     Map output bytes=284 08/09/12 20:42:54 INFO mapred.JobClient:     Combine input records=30 08/09/12 20:42:54 INFO mapred.JobClient:     Combine output records=16 08/09/12 20:42:54 INFO mapred.JobClient:     Reduce input groups=16 08/09/12 20:42:54 INFO mapred.JobClient:     Reduce input records=16 $

上面是根据给定的输入的数据进行执行的一个过程，其实结果已经输出到output-dir目录中，你可以进行查看，在目录G:\hadoop-0.16.4\output-dir下面生成了两个文件：.part-00000.crc和part-00000。

查看处理结果，如下所示：

或者，直接到G:\hadoop-0.16.4\output-dir目录下面打开part-00000文件查看即可，内容如下所示：

actor 1 add 2 after 2 apache 1 append 1 as 6 background 1 be 2 believe 1 bench 3 block 1 cafe 2 cat 4 communications 1 connection 1 cust 1

同上面的是一样的。

这是一个非常简单的例子，而且Hadoop在其中实现了Google的MapReduce算法，用以处理数据。

我们可以简单看一下关于WordCount类的实现，在Hadoop的发行包中也附带了例子的源代码，WordCount.java类实现如下所示：

package org.apache.hadoop.examples; import java.io.IOException; import java.util.ArrayList; import java.util.Iterator; import java.util.List; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapred.JobClient; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.MapReduceBase; import org.apache.hadoop.mapred.Mapper; import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Reducer; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; /** /** * 这是一个Hadoop Map/Reduce应用的例子。 * 读取输入的文件，实现将文件的每一行分解成单个的单词并统计单词的出现频率。 * 输出结果是被分解的单词的列表及其词频。 * 使用如下命令可以运行： bin/hadoop jar build/hadoop-examples.jar wordcount *            [-m <i>maps</i>] [-r <i>reduces</i>] <i>in-dir</i> <i>out-dir</i> */ public class WordCount extends Configured implements Tool { /**    * MapClass是一个内部静态类。统计数据文件中每一行的单词。    */ public static class MapClass extends MapReduceBase     implements Mapper<LongWritable, Text, Text, IntWritable> {         private final static IntWritable one = new IntWritable(1);     private Text word = new Text();         public void map(LongWritable key, Text value,                     OutputCollector<Text, IntWritable> output,                     Reporter reporter) throws IOException {       String line = value.toString();       StringTokenizer itr = new StringTokenizer(line);       while (itr.hasMoreTokens()) {         word.set(itr.nextToken());         output.collect(word, one);       }     } } /**    * Reduce是一个内部静态类。作为统计单词数量的中间结果类，由于这个例子简单无须执行中间结果的合并。    */ public static class Reduce extends MapReduceBase     implements Reducer<Text, IntWritable, Text, IntWritable> {         public void reduce(Text key, Iterator<IntWritable> values,                        OutputCollector<Text, IntWritable> output,                        Reporter reporter) throws IOException {       int sum = 0;       while (values.hasNext()) {         sum += values.next().get();       }       output.collect(key, new IntWritable(sum));     } } static int printUsage() { // 提示输入命令的用法     System.out.println("wordcount [-m <maps>] [-r <reduces>] <input> <output>");     ToolRunner.printGenericCommandUsage(System.out);     return -1; } /**    * map/reduce程序的驱动部分，用于实现提交map/reduce任务。    */ public int run(String[] args) throws Exception {     JobConf conf = new JobConf(getConf(), WordCount.class);     conf.setJobName("wordcount");     // the keys are words (strings)     conf.setOutputKeyClass(Text.class);     // the values are counts (ints)     conf.setOutputValueClass(IntWritable.class);         conf.setMapperClass(MapClass.class);            conf.setCombinerClass(Reduce.class);     conf.setReducerClass(Reduce.class);         List<String> other_args = new ArrayList<String>();     for(int i=0; i < args.length; ++i) {       try {         if ("-m".equals(args[i])) {           conf.setNumMapTasks(Integer.parseInt(args[++i]));         } else if ("-r".equals(args[i])) {           conf.setNumReduceTasks(Integer.parseInt(args[++i]));         } else {           other_args.add(args[i]);         }       } catch (NumberFormatException except) {         System.out.println("ERROR: Integer expected instead of " + args[i]);         return printUsage();       } catch (ArrayIndexOutOfBoundsException except) {         System.out.println("ERROR: Required parameter missing from " +                            args[i-1]);         return printUsage();       }     }     // Make sure there are exactly 2 parameters left.     if (other_args.size() != 2) {       System.out.println("ERROR: Wrong number of parameters: " +                          other_args.size() + " instead of 2.");       return printUsage();     }     conf.setInputPath(new Path(other_args.get(0)));     conf.setOutputPath(new Path(other_args.get(1)));             JobClient.runJob(conf);     return 0; } public static void main(String[] args) throws Exception {     int res = ToolRunner.run(new Configuration(), new WordCount(), args);     System.exit(res); } }

通过对着这个例子进行简单的说明，大致了解一下MapReduce算法的思想及其实现。

 

解决Window环境下启动Hadoop0.21.0时出现的 java.lang.NoClassDefFoundError: org/apache/hadoop/util/PlatformName 异常

在Window下启动Hadoop-0.21.0版本时，会出现下面这样的错误提示：

 1 java.lang.NoClassDefFoundError: org/apache/hadoop/util/PlatformName
 2 Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.util.PlatformName
 3 
 4         at java.net.URLClassLoader$1.run(URLClassLoader.java:202)
 5         at java.security.AccessController.doPrivileged(Native Method)
 6         at java.net.URLClassLoader.findClass(URLClassLoader.java:190)
 7         at java.lang.ClassLoader.loadClass(ClassLoader.java:307)
 8         at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:301)
 9         at java.lang.ClassLoader.loadClass(ClassLoader.java:248)
10 Could not find the main class: org.apache.hadoop.util.PlatformName.  Program wil
11 l exit.

经过不断的查找原因和尝试，终于有了解决这个错误的办法，只需要将${HADOOP_HOME}/bin/hadoop-config.sh文件中的第190行的一下的内容

JAVA_PLATFORM=`CLASSPATH=${CLASSPATH} ${JAVA} -Xmx32m ${HADOOP_JAVA_PLATFORM_OPTS} org.apache.hadoop.util.PlatformName | sed -e "s/ /_/g"`

修改成以下的内容即可

JAVA_PLATFORM=`CLASSPATH=${CLASSPATH} ${JAVA} -Xmx32m -classpath ${HADOOP_COMMON_HOME}/hadoop-common-0.21.0.jar org.apache.hadoop.util.PlatformName | sed -e "s/ /_/g"`