大数据学习 Day7

eclipse上编写mapreduce程序，进行词频统计

开启分布式文件系统（这里为伪分布），将/opt/input/test.txt文件上传到分布式文件系统中的/input下
hdfs dfs –put /opt/input/test.txt /input/

1. 保证安装好jdk_1.8.0_32bit
2. 配置hadoop的环境变量：
   

path中加入

3. 启动eclipse

   新建MapReduceDemo Java项目，将hadoop相关的jar包配置在项目中

   右键项目 Properties ->java build path ** -> **libraries 中add external jars

   在src下新建package，命名为com.sj.mapreduce

   新建类 WordCountMapper.java

package com.sj.mapreduce;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordCountMapper extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Mapper.Context context)
            throws IOException, InterruptedException {
        // TODO Auto-generated method stub
//        super.map(key, value, context);
        String line=value.toString();
        String[] words=line.split(" ");
        for(int i=0;i            //            context.write(new Text(words[i]),new IntWritable(1) );        }    }}
            //
            context.write(new Text(words[i]),new IntWritable(1) );
        }
    }
}

新建类WordCountReducer.java

package com.sj.mapreduce;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordCountReducer extends Reducer {
    @Override
    protected void reduce(Text key, java.lang.Iterable values, org.apache.hadoop.mapreduce.Reducer.Context context) throws IOException ,InterruptedException {
        int count =0;
        for(IntWritable value :values){
            count+=value.get();
    };
        context.write(key, new IntWritable(count));
    }



}

新建类WordCountDriver.java

package com.sj.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 WordCountDriver {
    //用Exception 抛异常，用Exception去接受所有异常，多态的体现
    public static void main(String[] args) throws Exception {
        System.setProperty("hadoop.home.dir", "D:\\hadoop-common-2.2.0-bin-master\\hadoop-common-2.2.0-bin-master");

        //集群中所配置的hadoop的配置文件
        Configuration configuration = new Configuration();
        //创建job
        Job job=Job.getInstance(configuration);
        //指定本程序的jar包所在路径
        job.setJarByClass(WordCountDriver.class);
        //指定本job要使用的Mapper业务类
        job.setMapperClass(WordCountMapper.class);
        //指定mapper输出的k-v类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        //制定本job要使用的Reduce业务类
        job.setReducerClass(WordCountReducer.class);
        //设置Reduce输出的最终k-v数据类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.245.132:9000/input/test.txt"));
        FileOutputFormat.setOutputPath(job,new Path("hdfs://192.168.245.132:9000/output"));
        Boolean res=job.waitForCompletion(true);
        System.exit(res?0:-1);


    }

}

运行WordCountDriver.java
对分布式系统中/input/test.txt中的测试文件进行词频统计，得到结果