统计一个文件中相同单词的个数
使用工具：IDEA+Maven
实验基础：必须在服务器上搭建好hadoop的hdfs环境和yarn环境，具体搭建方式：https://www.jianshu.com/p/59cb748bddec
运行环境：VM虚拟机上的一个linux系统运行hadoop，在window10下的idea中进行开发
开发步骤：
1，在pom.xml文件中导入hadoop的依赖包
<properties>
<hadoop.version>2.6.0-cdh5.7.0</hadoop.version>
</properties>
<repositories>
<repository>
<id>cloudera</id>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
<releases>
<enabled>true</enabled>
</releases>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>${hadoop.version}</version>
</dependency>
</dependencies>
2，新建一个WordCountApp这个类，用来统计一个文件中相同单词的个数，并将结果保存到一个文件中
package com.zhx.hadoop.mapreduce;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
/**
* @program: hdfs
* @description: this is a MAPREDUCE‘s test project
* @author: zhx
* @create: 2021-03-09 18:09
**/
public class WordCountApp {
/**
* map读取输入文件
* keyInput：LongWritable，文件的偏移量，开始为0，后面的key是每一行的字符量的累加
* valueInput：Text，每一行的值
* keyOutput：Text，输出是每个单词的字符串
* valueOutput：LongWritable，是每个单词的个数
*/
public static class MyMapper extends Mapper<LongWritable,Text,Text,LongWritable>{
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
LongWritable one = new LongWritable(1);
//获取一行数据
String line = value.toString();
//每一行按照空格进行分割开来
String[] words = line.split(" ");
for (String word : words) {
context.write(new Text(word), one);
}
}
}
/**
* reduce:归并操作
* map的输出就是reduce的输入，所以reduce的
* keyInput：Text
* valueInput：LongWritable
* keyOutput：Text
* valueOutput：LongWritable
*/
public static class MyReduce extends Reducer<Text,LongWritable,Text,LongWritable>{
@Override
protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
long sum=0;
for (LongWritable value : values) {
sum+=value.get();
}
context.write(key,new LongWritable(sum));
}
}
/**
* Driver,封装了MapReduce作业的所有信息
*/
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//创建configuration
Configuration configuration = new Configuration();
//创建job
Job job = Job.getInstance(configuration, "wordcount");
//设置job处理类
job.setJarByClass(WordCountApp.class);
//设置作业处理输入路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置map相关系数
job.setMapperClass(MyMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
//设置reduce的相关参数
job.setReducerClass(MyReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
//设置作业处理输出路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
//提交作业
System.exit(job.waitForCompletion(true)?0:1);
}
}
3，需要把这个文件打包上传到服务器中进行运行，具体操作
1，进入这个项目的根目录，打开shell窗口，执行
mvn clean package -DskipTests，这个命令，然后可以看到在target目录下有一个"项目名-1.0.jar"这个文件，说明第一步成了
2，继续执行scp .\target\hdfs-1.0.jar root@zhx:/home/zhx/lib/这个命令，lib目录是你的服务器上要存放这个文件的位置，把第一步生成的.jar文件上传到服务器上
3，执行命令hadoop jar /home/zhx/lib/hdfs-1.0.jar com.zhx.hadoop.mapreduce.WordCountApp hdfs://zhx:8020/mapreduce_test1.txt hdfs://zhx:8020/output/wc/，如果你的hdfs上没有mapreduce_test1.txt这个文件，可以先把这个文件上传到hdfs中，上传命令hadoop fs -put mapreduce_test1.txt在服务器上的的路径 /，/是hdfs上根路径，直接上传到hdfs的根路径就可以了，为了以后方便直接把上面的命令保存到wordcountapp.sh这个文件中，然后赋予这个文件执行的权限chmod u+x wordcountapp.sh，到时候可以直接./wordcountapp.sh，就可以执行了
4，mapreduce_test1.txt的文件内容hello mapreduce hdfs mapreduce yarn hello is a hdfs yarn mapreduce5，查看是否成功，执行
hadoop fs -ls /output/wc这个命令，看是否存在part-r-00000这个文件，然后查看这个文件hadoop fs -cat /output/wc/part-r-00000，内容如下：a 1 hdfs 2 hello 2 is 1 mapreduce 3 yarn 2
4，上述3步已经基本上开发完成了，但是还是存在一些问题的
如果此刻继续执行wordcountapp.sh这个文件会出现错误，这个错误是说hdfs上已经存在了输出文件夹了，在MapReduce中输出文件夹不能事先存在
错误日志
因此有两种解决办法：1，是手动在shell中将输出文件目录删除，执行hadoop fs -rm -R /output2，是在代码中完成删除功能//创建configuration Configuration configuration = new Configuration(); //准备清理已存在的输出目录 Path outpath = new Path(args[1]); FileSystem fileSystem = FileSystem.get(configuration); if (fileSystem.exists(outpath)) { fileSystem.delete(outpath, true); System.out.println("已经删除了存在的目录"); }把这段代码加入到wordcountapp中的main函数中，然后重新打包上传，执行wordcountapp.sh这个文件，看是否出错
5，继续优化，引入combiner，这个可以本地reduce，从而可以减少MapTask输出的数据量以及网络的传输量，他可以先把每个Mapper上的词先和合并一波，然后把合并后的传给reduce
combiner.png
使用场景：求和，次数，加法这些可以，但是平均数就不行
把这个job.setCombinerClass(MyReduce.class);加入到main中//设置map相关系数 job.setMapperClass(MyMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); //通过job设置combine处理类，其实逻辑上与reduce一致 job.setCombinerClass(MyReduce.class);然后执行打包上传，执行.sh文件，查看执行过程中的日志
combiner执行结果
只要这两个不为0说明combiner起作用了
6，partitioner分区器
shuffle是通过分区partitioner 分配给Reduce的，一个Reducer对应一个记录文件，Partitioner是shuffle的一部分，partitioner执行时机：在mapper执行完成，Reducer还没有执行的时候，mapper的输出就是partitioner的输入 即<k2,v2>partitioner 分区主要是用来提高效率的 ，例如从全国基站的数据中查找北京基站的数据，如果计算时不分区全国的数据都放在一起，查询的时候就相当于全表扫描 效率非常低，如果在第一次进行Mapreducer计算的时候按照省市进行分区，每个城市的基站数据都存储在对应的每个文件，那么下次再进行查询的时候直接从北京分区里直接查找 效率很高。分区的依据是具体业务需求，可以按照省市分区，时间进行分区等。如果不手动进行分区，Hadoop有一个默认的分区规则Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。HashPartitioner是mapreduce的默认partitioner。计算方法是which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到当前的目的reducer。
新建一个PartitonerApp这个类
package com.zhx.hadoop.mapreduce;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
/**
* @program: hdfs
* @description: this is a MAPREDUCE‘s test project
* @author: zhx
* @create: 2021-03-09 18:09
**/
public class PartitonerApp {
/**
* map读取输入文件
* keyInput：LongWritable，文件的偏移量，开始为0，后面的key是每一行的字符量的累加
* valueInput：Text，每一行的值
* keyOutput：Text，输出是每个单词的字符串
* valueOutput：LongWritable，是每个单词对应的value值
* 这是测试 数据，一个文件中，第一列式手机品牌，第二列是销售数，统计每种手机总的销售数，这里输出key是品牌字符串，value是销售数
* huawei 100
xiaomi 100
huawei 200
xiaomi 300
nojiya 100
iphone 200
huawei 200
iphone 200
*/
public static class MyMapper extends Mapper<LongWritable,Text,Text,LongWritable>{
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
LongWritable one = new LongWritable(1);
//获取一行数据
String line = value.toString();
//每一行按照空格进行分割开来
String[] words = line.split(" ");
context.write(new Text(words[0]), new LongWritable(Long.parseLong(words[1])));
}
}
/**
* reduce:归并操作
* map的输出就是reduce的输入，所以reduce的
* keyInput：Text
* valueInput：LongWritable
* keyOutput：Text
* valueOutput：LongWritable
*/
public static class MyReduce extends Reducer<Text,LongWritable,Text,LongWritable>{
@Override
protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
long sum=0;
for (LongWritable value : values) {
sum+=value.get();
}
context.write(key,new LongWritable(sum));
}
}
public static class MyPartitioner extends Partitioner<Text,LongWritable>{
@Override
public int getPartition(Text key, LongWritable value, int i) {
if(key.toString().equals("huawei")){
return 0;
}
if(key.toString().equals("xiaomi")){
return 1;
}
if(key.toString().equals("iphone")){
return 2;
}
return 3;
}
}
/**
* Driver,封装了MapReduce作业的所有信息
*/
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//创建configuration
Configuration configuration = new Configuration();
//准备清理已存在的输出目录
Path outpath = new Path(args[1]);
FileSystem fileSystem = FileSystem.get(configuration);
if (fileSystem.exists(outpath)) {
fileSystem.delete(outpath, true);
System.out.println("已经删除了存在的目录");
}
//创建job
Job job = Job.getInstance(configuration, "wordcount");
//设置job处理类
job.setJarByClass(PartitonerApp.class);
//设置作业处理输入路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
//设置map相关系数
job.setMapperClass(MyMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
//通过job设置combine处理类，其实逻辑上与reduce一致
job.setCombinerClass(MyReduce.class);
//设置reduce的相关参数
job.setReducerClass(MyReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
//设置job的partitioner的处理类
job.setPartitionerClass(MyPartitioner.class);
//设置4个reducer，每个分区一个
job.setNumReduceTasks(4);
//设置作业处理输出路径
FileOutputFormat.setOutputPath(job,new Path(args[1]));
//提交作业
System.exit(job.waitForCompletion(true)?0:1);
}
}
然后打包上传，执行partitonerapp.sh文件，这个文件内容hadoop jar /home/zhx/lib/hdfs-1.0.jar com.zhx.hadoop.mapreduce.PartitonerApp hdfs://zhx:8020/mapreduce_partitioner.txt hdfs://zhx:8020/output/wc/，此时输出文件夹下会有4个文件，分别是
配置jobhistory
记录已运行完的MapReduce信息到指定的hdfs目录下，默认是不开启的，可以http://zhx:8088/打开yarn的网址，然后运行一个pi的job，可以查看Tracking UI这列下的history，是否可以打开，打不开说明需要配置
mapreduce-site.xml
<property>
<name>mapreduce.jobhistory.address</name>
<value>zhx:10020</value>
<description>mapreduce jobhistory server ip(host:port)</description>
</property>
<property>
<name>mapreduce.jobhistory.webapp.address</name>
<value>zhx:19888</value>
<description>mapreduce jobhistory server web ui host:port</description>
</property>
<property>
<name>mapreduce.jobhistory.done-dir</name>
<value>/history</value>
</property>
<property>
<name>mapreduce.jobhistory.intermediate-done-dir</name>
<value>/history/done_intermediate</value>
</property>
yarn-site.xml
<property>
<name>yarn.log-aggregation-enable</name>
<value>true</value>
</property>
然后停掉yarn，在重启yarn，在启动sbin目录下的
./mr-jobhistory-daemon.sh start historyserver
然后jps，看是否有 JobHistoryServer这个进程
最后可以在yarn的浏览器中查看执行的日志了
image.png
