Hadoop第五章：词频统计

Hadoop第一章：环境搭建
Hadoop第二章：集群搭建（上）
Hadoop第二章：集群搭建（中）
Hadoop第二章：集群搭建（下）
Hadoop第三章：Shell命令
Hadoop第四章：Client客户端
Hadoop第四章：Client客户端2.0
Hadoop第五章：词频统计

• 系列文章目录
• 前言
• 一、创建项目
• 二、基本环境搭建
• • 1.添加依赖
  • 2.创建日志
  • 3.创建包和类
• 三、编写函数
• • 1.WordCountMapper
  • 2.WordCountReducer
  • 3.WordCountDriver
• 四、本地运行
• 五、集群运行
• • 1.添加依赖并打包
  • 2.启动集群并上传jar包
• 总结

之前由于学校学习压力，已经断更快两个月，从现在起开始恢复，最低一周一篇。
之前我们使用了hadoop自带的hadoop词频统计jar包进行了词频统计，现在我们手写一个，更加深刻的理解一下hadoop。

博主用的是最新版的idea，项目创建和之前有一些所以重新演示一下。

        
            org.apache.hadoop
            hadoop-client
            3.3.2
        
        
            junit
            junit
            4.13.2
        
        
            org.slf4j
            slf4j-log4j12
            1.7.36
        
    

创建log4j.properties文件

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

package com.atguigu.mapreduce.wordcount;

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

import java.io.IOException;

public class WordCountMapper extends Mapper {
	//为了节省空间，将k-v设置到函数外
    private Text outK=new Text();
    private IntWritable outV=new IntWritable(1);


    @Override
    protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
    	//获取一行输入数据
        String line = value.toString();
		//将数据切分
        String[] words = line.split(" ");
		//循环每个单词进行k-v输出
        for (String word : words) {
            outK.set(word);
			//将参数传递到reduce
            context.write(outK,outV);
        }
    }
}

package com.atguigu.mapreduce.wordcount;

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

import java.io.IOException;

public class WordCountReducer extends Reducer {
	//全局变量输出类型
    private IntWritable outV = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable values,Context context) throws IOException, InterruptedException {		//设立一个计数器
        int sum=0;
        //统计单词出现个数
        for (IntWritable value : values) {
            sum+=value.get();
        }
		//转换结果类型
        outV.set(sum);
		//输出结果
        context.write(key,outV);
    }
}

package com.atguigu.mapreduce.wordcount;

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;

import java.io.IOException;

public class WordCountDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
    	//1.获取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

		//2.设置jar包路径
        job.setJarByClass(WordCountDriver.class);

		//3.关联mapper和reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

		//4.设置map输出kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
		//5.设置最终输出kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
		//6.设置输入路径和输出路径
        FileInputFormat.setInputPaths(job,new Path("D:\learn\hadoop\wordcount\input"));
        FileOutputFormat.setOutputPath(job,new Path("D:\learn\hadoop\wordcount\output"));
		//7.提交job
        boolean result = job.waitForCompletion(true);

        System.exit(result?0:1);
    }
}

之前已经配置过本地的hadoop环境了，就不再重复了。

这代表运行成功

到自己设置的输出目录进行查看。

为了方便讲个文件上传到集群，我们需要打包，所以要先添加打包的依赖。
在pom.xml进行依赖添加

        
            
                maven-compiler-plugin
                3.6.1
                
                    1.8
                    1.8
                
            
            
                maven-assembly-plugin
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                
            
        
    

刚刚的代码我们需要人工设置输入输出路径，在集群中，显然是很不方便的，所以咱们再写一个包，将之前的三个代码一次性复制过来简单修改一下。


现在，我们进行打包

完成后，可在右边查看

可以看到，这里有两个jar包，大小差别很大，说一下有什么区别，第一个jar包里只包含代码需要将需要的hadoop运行环境提前配置好，第二个jar含有需要的依赖，可以直接调用。

可以简单修改一下文件名。
在集群上传测试数据。



集群运行。
hadoop jar 包命 包类 输出路径 输出路径
包类可以从idea中获取
com.atguigu.mapreduce.wordcount2.WordCountDriver

hadoop jar wc.jar com.atguigu.mapreduce.wordcount2.WordCountDriver /wcinput /wcoutput

查看结果

到此词频统计的项目基本就结束了，开始把之前欠的 博客补一补。