在eclipse使用map reduce编写word count程序生成jar包并在虚拟机运行的步骤

---恢复内容开始---

1.首先准备一个需要统计的单词文件 word.txt，我们的单词是以空格分开的，统计时按照空格分隔即可

hello hadoop
hello yarn
hello zookeeper
hdfs hadoop
select from hadoop
select from yarn
mapreduce
mapreduce

2.上传word.txt到hdfs根目录

$ bin/hdfs dfs -put test/word.txt /

3.准备工作完成后在eclipse编写代码，分别编写map、reduce、driver等java文件

wordcountmap.java

map执行我们的word.txt 文件是按行执行，每一行执行一个map

wordcountmap.java

map执行我们的word.txt 文件是按行执行，每一行执行一个map

package com.ijeffrey.mapreduce.wordcount.client;

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;
/**
* map 输出的键值对必须和reducer输入的键值对类型一致
* @author pxy
*
*/
public class wordcountmap extends mapper<longwritable, text, text, intwritable> {

private text keyout = new text();
private intwritable valueout = new intwritable(1);

@override
protected void map(longwritable key, text value, mapper<longwritable, text, text, intwritable>.context context)
throws ioexception, interruptedexception {

string line = value.tostring();
// 我的文件记录的单词是以空格记录单词，所以这里用空格来截取
string[] words = line.split(" ");

// 遍历数组，并以k v 对的形式输出
for (string word : words) {
keyout.set(word);
context.write(keyout, valueout);
}
}

}

wordcountreducer.java

package com.ijeffrey.mapreduce.wordcount.client;

import java.io.ioexception;

import org.apache.hadoop.io.intwritable;
import org.apache.hadoop.io.text;
import org.apache.hadoop.mapreduce.reducer;

/**
* reducer 输入的键值对必须和map输出的键值对类型一致
* map <hello,1> <world,1> <hello,1> <apple,1> ....
* reduce 接收 <apple,[1]> <hello,[1,1]> <world,[1]>
* @author pxy
*
*/
public class wordcountreducer extends reducer<text, intwritable, text, intwritable> {
private intwritable valueout = new intwritable();

@override
protected void reduce(text key, iterable<intwritable> values,
reducer<text, intwritable, text, intwritable>.context context) throws ioexception, interruptedexception {
int count = 0; // 统计总数

// 遍历数组，累加求和
for(intwritable value : values){

// intwritable类型不能和int类型相加，所以需要先使用get方法转换成int类型
count += value.get();
}

// 将统计的结果转成intwritable
valueout.set(count);

// 最后reduce要输出最终的 k v 对
context.write(key, valueout);

}
}

wordcountdriver.java

package com.ijeffrey.mapreduce.wordcount.client;

import java.io.ioexception;

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;

/**
* 运行主函数
* @author pxy
*
*/
public class wordcountdriver {
public static void main(string[] args) throws ioexception, classnotfoundexception, interruptedexception {
configuration conf = new configuration();

// 获得一个job对象，用来完成一个mapreduce作业
job job = job.getinstance(conf);

// 让程序找到主入口
job.setjarbyclass(wordcountdriver.class);

// 指定输入数据的目录，指定数据计算完成后输出的目录
// sbin/yarn jar share/hadoop/xxxxxxx.jar wordcount /wordcount/input/ /wordcount/output/
fileinputformat.addinputpath(job, new path(args[0]));
fileoutputformat.setoutputpath(job, new path(args[1]));

// 告诉我调用那个map方法和reduce方法
job.setmapperclass(wordcountmap.class);
job.setreducerclass(wordcountreducer.class);

// 指定map输出键值对的类型
job.setmapoutputkeyclass(text.class);
job.setmapoutputvalueclass(intwritable.class);

// 指定reduce输出键值对的类型
job.setoutputkeyclass(text.class);
job.setoutputvalueclass(intwritable.class);

// 提交job任务
boolean result = job.waitforcompletion(true);
system.exit(result ? 0 : 1);

}
}

}

4.将编写完成的代码打成jar包,并在集群上运行

将jar上传到到服务器，启动服务后运行我们自己编写的mapreduce，统计根目录下的word.txt并将运行结果写入output

$ bin/yarn jar test/wordcount.jar com.ijeffrey.mapreduce.wordcount.client.wordcountdriver /word.txt /output

注意：运行jar的时候要添加driver的完全路径

运行完成后查看output结果:

$ bin/hdfs dfs -text /output12/part-r-00000