1.创建maven工程

  配置hadoop的依赖环境；

pom.xml
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>hadoop</groupId>
  <artifactId>hadoop1</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>2.6.1</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
  </dependencies>
</project>

2.测试所使用数据：1995年美国航空数据，字段名；

3.编写MyMapper1.java

package test;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MyMapper1 extends Mapper<LongWritable,Text,Text,Text>{

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
            //得到待处理的字符串,value是行
            String line = value.toString();

            //字符串分割，typeList数组中存放着字段类型。
            String [] typeList = line.split(",");
            String month = null;
            String reason = null;
            if((!typeList[1].equals("CancellationCode")) && typeList.length>22) {
                reason = typeList[22].toString();   //得到这一行数据的原因
                month = typeList[1].toString();     //得到这一行数据的月份
                context.write(new Text(reason), new Text(month));
            }
    }
}

4.编写MyReducer1.java

package test;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class MyReducer1 extends Reducer<Text, Text, Text, LongWritable>{

    
    //得到的数据为：// <k2取消原因：v2s月份[1,1,2,2,2,..]>
    @Override
    protected void reduce(Text k2, Iterable<Text> v2s, Reducer<Text, Text, Text, LongWritable>.Context context)
            throws IOException, InterruptedException {

            long sum = 0;
            //对1月份进行统计
            for(Text month : v2s) {
                if(month.toString().equals("1")) {
                    sum += 1L;
                };
            }
        
            //输出格式为：  <原因:次数>
            context.write(k2, new LongWritable(sum));
        }
}

5.编写入口程序WordCount.java

package test;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
        Job job = Job.getInstance(conf,WordCount.class.getSimpleName());
        
        //set main class
        job.setJarByClass(WordCount.class);
        
        //set <k2,v2> type
        job.setMapperClass(MyMapper1.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        
        //set <k3,v3> type
        job.setReducerClass(MyReducer1.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        
        //输入、输出路径设置；
        FileInputFormat.setInputPaths(job, new Path("/input"));
        FileOutputFormat.setOutputPath(job, new Path("/output"));
        
        System.out.println(job.waitForCompletion(true));
    }
}

6.打成jar包，上传到服务器中并执行

https://blog.csdn.net/u010521842/article/details/75042771   作者：yanzhelee