package hadoopWordCount;  

import java.io.IOException;  

import java.util.*;  

import org.apache.hadoop.fs.Path;  

import org.apache.hadoop.conf.*;  

import org.apache.hadoop.io.*;  

import org.apache.hadoop.mapreduce.*;  

import org.apache.hadoop.mapreduce.lib.input.*;  

import org.apache.hadoop.mapreduce.lib.output.*;  

import org.apache.hadoop.util.*;  

public class hadoopWordCount extends Configured implements Tool  

{  

    public static class MapWordCount extends Mapper  

    {  

        public void map (LongWritable key, Text value, Context context) throws IOException, InterruptedException  

        {  

            String line = value.toString();  

            StringTokenizer tokenizer = new StringTokenizer (line);  

            while (tokenizer.hasMoreTokens())  

            {  

                context.write (new Text(tokenizer.nextToken()), new IntWritable(1));  

            }  

        }  

    }  

    public static class ReduceWordCount extends Reducer  

    {  

        public void reduce (Text key, Iterable values, Context context) throws IOException, InterruptedException  

        {  

            int sum = 0;  

            for (IntWritable val: values)  

                sum += val.get();  

            context.write (key, new IntWritable(sum));  

        }  

    }  

    public int run (String [] args) throws Exception  

    {  

        Job job = new Job (getConf());  

        job.setJarByClass (hadoopWordCount.class);  

        job.setJobName ("hadoop word count");  

        job.setOutputKeyClass (Text.class);  

        job.setOutputValueClass (IntWritable.class);  

        job.setMapperClass (MapWordCount.class);  

        job.setReducerClass (ReduceWordCount.class);  

        job.setInputFormatClass (TextInputFormat.class);  

        job.setOutputFormatClass (TextOutputFormat.class);  

        FileInputFormat.setInputPaths (job, new Path(args[1]));  

        FileOutputFormat.setOutputPath (job, new Path(args[2]));  

        boolean success = job.waitForCompletion (true);  

        return success ? 0:1;  

    }  

    public static void main (String [] args) throws Exception  

    {  

        System.out.println ("hello world for hadoop word count!");  

        ToolRunner.run (new hadoopWordCount(),args);  

    }  

}  

2.javac编译

我把该WordCount.java放在/home/taohang/hadoopWordCount/src/hadoopWordCount.java

hadoopWordCount

           |

           bin和src两个文件夹

javac -classpath /home/taohang/Hadoop/hadoop-core-1.2.1.jar   -d  /home/taohang/hadoopWordCount/bin           /home/taohang/hadoopWordCount/src/hadoopWordCount.java

javac 中的-classpath参数指明这些类库的位置,这些类库都打包在hadoop-core-1.2.1.jar

/home/taohang/hadoopWordCount/bin 是编译后的文件放置的目录

/home/taohang/hadoopWordCount/src/hadoopWordCount.java是源文件所在的目录

3.打包成.jar

接下来，用jar命令打jar包。首先写MANIFEST.MF文件，主要作用在于指定jar包中main函数所在的类；如果有多个main函数，则指定运行jar包的时候，具体进入哪个main函数。把MANIFEST.MF放到src文件夹中，MANIFEST.MF文件内容如下：

Manifest-Version: 1.0  

Created-By: 1.6.0 (Sun Microsystems Inc.)  

Main-Class: hadoopWordCount.hadoopWordCount  

接下来，cd到bin目录下——这个很重要，在别的目录下运行jar命令，虽然也能顺利打出jar包，但是当用java命令或者hadoop jar命令来运行的时候，总是找不到入口类（main函数）——然后运行jar命令打包，如下：

jar -cvfm helloWordCount.jar /home/taohang/hadoopWordCount/src/MANIFEST.MF 

4.准备输入的数据文件

首先建立一个input路径，用命令

bin/hadoop fs -mkdir input

检查是否创建成功，用命令

bin/hadoop fs -ls

在hadoopWordCount下新建一个文件夹data，放入try.txt，try.txt的内容是hello hello hello world。接下来，将这个文件放到hdfs的./input目录中，用命令

bin/hadoop fs -put   /home/taohang/hadoopWordCount/data/try.txt  ./input

第一个是本地路径，第二个是hadoop路径。

5.在hadoop上运行hadoopWordCount程序

~/hadoop$ hadoop jar /home/taohang/hadoopWordCount/bin/helloWordCount.jar hadoopWordCount input/try.txt outputwordcount

第一个参数是jar包的本地路径；第二个参数其实由helloWordCount类的main函数决定（args[1]），是输入数据的路径，是hdfs路径；第三个参数是输出结果的路径。需要注意的是，输出结果的路径必须是不存在的。否则hadoop认为这已经有了结果，会拒绝执行任务。

查看输出结果

hadoop dfs -ls outputwordcount

在terminal中会出现

查看程序运行结果：
hadoop dfs -cat outputwordcount/part-r-00000

结果是

hello    3
    world    1