设为首页 收藏本站

运维网

 找回密码
 立即注册
查看: 2370|回复: 0

[经验分享] Hadoop（十五）MapReduce程序实例

[复制链接]
wxin

尚未签到
发表于 2017-12-17 16:35:32 | 显示全部楼层 |阅读模式
一、统计好友对数（去重）

1.1、数据准备

DSC0000.gif DSC0001.gif 　　

joe,    jon　　
joe , kia   
　　
joe, bob   
　　
joe ,ali
　　
kia,    joe   
　　
kia ,jim   
　　
kia, dee
　　
dee    ,kia   
　　
dee, ali
　　
ali    ,dee   
　　
ali, jim   
　　
ali ,bob   
　　
ali, joe   
　　
ali ,jon
　　
jon,    joe   
　　
jon ,ali
　　
bob,    joe   
　　
bob ,ali   
　　
bob, jim
　　
jim    ,kia   
　　
jim, bob   
　　
jim ,ali
　　


friends.txt　　有一个friends.txt文件，里面的一行的格式是：
　　用户名,好友名
DSC0002.png


1.2、需求分析
　　1）需求
　　统计有多少对好友
　　2）分析
　　从上面的文件格式与内容，有可能是出现用户名和好友名交换位置的两组数据，这时候这就要去重了。
　　比如说:
　　joe,  jon
　　jon,  joe
　　这样的数据，我们只能保留一组。

1.3、代码实现

　　

import org.apache.hadoop.conf.Configuration;　　
import org.apache.hadoop.conf.Configured;
　　
import org.apache.hadoop.fs.Path;
　　
import org.apache.hadoop.io.LongWritable;
　　
import org.apache.hadoop.io.NullWritable;
　　
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.TextInputFormat;
　　
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
　　
import org.apache.hadoop.util.Tool;
　　
import org.apache.hadoop.util.ToolRunner;
　　

　　
import java.io.IOException;
　　

　　

public>
static>
Text key = new Text();　　
@Override
　　

protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {　　
String[] split
= value.toString().split(",");　　

if (split.length==2){　　
String name_1
= split[0].trim();　　
String name_2
= split[1].trim();　　

if (!name_1.equals(name_2)){　　

this.key.set(　　
name_1.compareTo(name_2)
>0?　　
name_1
+","+name_2:　　
name_2
+","+name_1);　　
context.write(
this.key,NullWritable.get());　　
}
　　
}
　　
}
　　
}
　　

　　

static>
@Override　　

protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {　　
context.write(key, NullWritable.
get());　　
}
　　
}
　　

　　
@Override
　　

public int run(String[] strings) throws Exception {　　
Configuration conf
= getConf();　　
Path input
=  new Path(conf.get("iniput"));　　
Path output
=  new Path(conf.get("output"));　　

　　
Job job
= Job.getInstance(conf, this.getClass().getSimpleName() + "Lance");　　
job.setJarByClass(
this.getClass());　　

　　
job.setMapperClass(DuplicateDataMapper.
class);　　
job.setMapOutputKeyClass(Text.
class);　　
job.setMapOutputValueClass(NullWritable.
class);　　

　　
job.setReducerClass(DuplicatteDataReducer.
class);　　
job.setOutputKeyClass(Text.
class);　　
job.setOutputValueClass(NullWritable.
class);　　

　　
TextInputFormat.addInputPath(job, input);
　　
TextOutputFormat.setOutputPath(job,output);
　　

　　

return job.waitForCompletion(true)?0:1;　　
}
　　

　　

public static void main(String[] args) throws Exception {　　
System.exit(ToolRunner.run(
new DuplicateData_0010(),args));　　
}
　　
}
　　


DuplicateData_0010
二、词频统计

2.1、数据准备　　　　
　　设有4组原始文本数据：
　　Text 1: the weather is good          Text 2: today is good
　　Text 3: good weather is good       Text 4: today has good weather

2.2、需求分析
　　1）需求
　　求每篇文章每个单词出现的次数
　　2）分析
　　第一：
DSC0003.png

　　第二：
DSC0004.png

　　预期出现的结果：
DSC0005.png


2.3、代码实现
　　1）编写一个CountWordMapper类去实现Mapper

　　

/**　　
*通过继承org.apache.hadoop.mapreduce.Mapper编写自己的Mapper
　　

*/　　
public>　　
private final static IntWritable one=new IntWritable(1); //统计使用变量
　　
private Text word=new Text(); //单词变量
　　
/**
　　
* key:当前读取行的偏移量
　　
* value： 当前读取的行
　　
* context:map方法执行时上下文
　　
*/
　　
@Override
　　
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
　　
throws IOException, InterruptedException {
　　
// TODO Auto-generated method stub
　　
StringTokenizer words=new StringTokenizer(value.toString());
　　
while(words.hasMoreTokens()){
　　
word.set(words.nextToken());
　　
context.write(word, one);
　　
}
　　
}
　　
}
　　


WordCountMapper　　2）编写一个CountWordReducer类去实现Reducer


　　

/**　　
* 通过继承org.apache.hadoop.mapreduce.Reducer编写自己的Reducer
　　

*/　　
public>　　
/**
　　
* key:待统计的word
　　
* values:待统计word的所有统计标识
　　
* context:reduce方法执行时的上下文
　　
*/
　　
@Override
　　
protected void reduce(Text key, Iterable<IntWritable> values,Reducer<Text, IntWritable, Text, IntWritable>.Context
　　
context) throws IOException, InterruptedException {
　　
// TODO Auto-generated method stub
　　
int count=0;
　　
for(IntWritable one:values){
　　
count+=one.get();
　　
}
　　
context.write(key, new IntWritable(count));
　　
}
　　
}
　　


WordCountReducer　　3）编写一个WordCount作业调度的驱动程序WordCountDriver


　　

/**　　
* WordCount作业调度的驱动程序 *
　　

*/　　
public>　　
public static void main(String[] args) throws Exception {
　　
// 构建新的作业
　　
Configuration conf=new Configuration();
　　
Job job = Job.getInstance(conf, "Word Count");
　　
job.setJarByClass(WordCountDriver.class);
　　
// 设置Mapper和Reducer函数
　　
job.setMapperClass(WordCountMapper.class);
　　
job.setReducerClass(WordCountReducer.class);
　　
// 设置输出格式
　　
job.setOutputKeyClass(Text.class);
　　
job.setOutputValueClass(IntWritable.class);
　　
// 设置输入和输出目录
　　
FileInputFormat.addInputPath(job, new Path(args[0]));
　　
FileOutputFormat.setOutputPath(job, new Path(args[1]));
　　
// ᨀ交作业执行
　　
System.exit(job.waitForCompletion(true)?0:1);
　　
}
　　
}
　　


WordCountDriver
2.4、　运行测试
　　1）前期准备
　　将程序打成jar包： wordcount.jar
　　准备好Text 1-4文件
　　2）运行
　　yarn jar wordcount.jar com.briup.WordCount input output

三、成绩统计

3.1、数据准备
　　chinese.txt


　　

a|李一|88　　
a
|王二|26　　
a
|张三|99　　
a
|刘四|58　　
a
|陈五|45　　
a
|杨六|66　　
a
|赵七|78　　
a
|黄八|100　　
a
|周九|62　　
a
|吴十|12　　


chinese.txt　　english.txt

　　

b|李一|36　　
b
|王二|66　　
b
|张三|86　　
b
|刘四|56　　
b
|陈五|43　　
b
|杨六|86　　
b
|赵七|99　　
b
|黄八|80　　
b
|周九|70　　
b
|吴十|33　　


english.txt　　math.txt

　　

c|李一|83　　
c
|王二|36　　
c
|张三|92　　
c
|刘四|58　　
c
|陈五|75　　
c
|杨六|66　　
c
|赵七|63　　
c
|黄八|60　　
c
|周九|62　　
c
|吴十|72　　


math.txt　　我看查看chinese.txt查看数据格式：
DSC0006.png

　　a代表语文：李一代表名字：88代表语文成绩

3.2、需求分析
　　根据上面的数据，统计一下分数，格式如下：
DSC0007.png


3.3、代码实现
　　1）编写一个解析类解析上面的每门课的数据
　　ScoreRecordParser

　　

import org.apache.hadoop.io.Text;　　

　　

public>
private String>
private String name;　　

private String score;　　

　　

public boolean parse(String line){　　
String[] strs
=line.split("[|]");　　

if(strs.length<3){　　

return false;　　
}
　　
id
=strs[0].trim();　　
name
=strs[1].trim();　　
score
=strs[2].trim();　　

if(id.length()>0&&name.length()>0&&score.length()>0){　　

return true;　　
}
else{　　

return false;　　
}
　　
}
　　

　　

public boolean parse(Text line){　　

return parse(line.toString());　　
}
　　

　　

public String getId(){　　

return>
}　　

　　

public void setId(String>
this.id=id;　　
}
　　

　　

public String getName(){　　

return name;　　
}
　　

　　

public void setName(String name){　　

this.name=name;　　
}
　　

　　

public String getScore(){　　

return score;　　
}
　　

　　

public void setScore(String score){　　

this.score=score;　　
}
　　
}
　　


View Code　　2）实现需求
　　CalculateScore_0010

　　

import com.briup.bd1702.hadoop.mapred.utils.ScoreRecordParser;　　
import java.io.IOException;
　　
import org.apache.hadoop.conf.Configuration;
　　
import org.apache.hadoop.conf.Configured;
　　
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.TextInputFormat;
　　
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
　　
import org.apache.hadoop.util.Tool;
　　
import org.apache.hadoop.util.ToolRunner;
　　

　　

public>
private static ScoreRecordParser parser=new ScoreRecordParser();　　

　　

static>
private Text key=new Text();　　

　　
@Override
　　

protected void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException{　　

if(parser.parse(value)){　　

this.key.set(parser.getName());　　
context.write(
this.key,value);　　
}
　　
}
　　
}
　　

　　

static>
private Text value=new Text();　　

　　
@Override
　　

protected void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{　　
StringBuffer buffer
=new StringBuffer();　　

double sum=0;　　

for(Text text:values){　　

if(parser.parse(text)){　　
String>=parser.getId();　　
String score
=parser.getScore();　　

switch(id){　　

case "a":{　　
buffer.append(
"语文："+score+"\t");　　

break;　　
}
　　

case "b":{　　
buffer.append(
"英语："+score+"\t");　　

break;　　
}
　　

case "c":{　　
buffer.append(
"数学："+score+"\t");　　

break;　　
}
　　
}
　　
sum
+=Double.parseDouble(score);　　
}
　　
}
　　
buffer.append(
"总分："+sum+"\t平均分："+(sum/3));　　

this.value.set(buffer.toString());　　
context.write(key,
this.value);　　
}
　　
}
　　
@Override
　　

public int run(String[] args) throws Exception{　　
Configuration conf
=getConf();　　
Path input
=new Path(conf.get("input"));　　
Path output
=new Path(conf.get("output"));　　

　　
Job job
=Job.getInstance(conf,this.getClass().getSimpleName());　　
job.setJarByClass(
this.getClass());　　

　　
job.setMapperClass(CalculateScoreMapper.
class);　　
job.setMapOutputKeyClass(Text.
class);　　
job.setMapOutputValueClass(Text.
class);　　

　　
job.setReducerClass(CalculateScoreReducer.
class);　　
job.setOutputKeyClass(Text.
class);　　
job.setOutputValueClass(Text.
class);　　

　　
job.setInputFormatClass(TextInputFormat.
class);　　
job.setOutputFormatClass(TextOutputFormat.
class);　　

　　
TextInputFormat.addInputPath(job,input);
　　
TextOutputFormat.setOutputPath(job,output);
　　

　　

return job.waitForCompletion(true)?0:1;　　
}
　　

　　

public static void main(String[] args) throws Exception{　　
System.exit(ToolRunner.run(
new P00010_CalculateScore_0010(),args));　　
}
　　
}
　　


CalculateScore_0010
3.4、执行
　　这里执行因为有三个文件，我们用一个目录去存储，然后在-Dinput中指定这个目录就可以了 。
　　在上面的三个文件中，都是特别小的，所以三个文件要用三个数据块去存储，然后用三个map去执行者三个文件。

四、倒排索引
　　首先知道什么是倒排索引？
　　比如所我们有file_1到file_4这四篇文章，我们需要求出：某个单词，在每一篇文章出现的次数
　　比如说输出格式是这样的：
　　某个单词　　file_1:出现次数,file_2:出现次数,file_3:出现次数,file_4:出现次数

4.1、数据准备
　　file_1


　　

Welcome to MapReduce World　　


file_1　　file_2

　　

MapReduce is simple　　


file_2　　file_3

　　

MapReduce is powerful is simple 　　


file_3　　file_4

　　

hello MapReduce Bye MapReduce　　


file_4
4.2、需求分析
　　1）需求
　　实现文件输出格式如下：
　　某个单词　　file_1:出现次数,file_2:出现次数,file_3:出现次数,file_4:出现次数
　　2）分析
　　比如MapReduce这个单词，我们分析一下：
　　在map端出来的格式：
DSC0008.png 注意：f1,f2,f3,f4代表文件名

　　经过洗牌之后，进入reduce的数据格式：
DSC0009.png

　　在reduce怎么处理呢？
　　我们构建一个Map集合用来存放某个路径在这个集合中出现的次数：
DSC00010.png

　　最后就可以形成我们想要的文件：
DSC00011.png


4.3、代码实现倒排索引

　　

import java.io.IOException;　　
import java.util.HashMap;
　　
import java.util.Map;
　　
import java.util.Map.Entry;
　　
import java.util.Set;
　　
import java.util.StringTokenizer;
　　
import org.apache.hadoop.conf.Configuration;
　　
import org.apache.hadoop.conf.Configured;
　　
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.FileSplit;
　　
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
　　
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
　　
import org.apache.hadoop.util.Tool;
　　
import org.apache.hadoop.util.ToolRunner;
　　

　　

public>
static>
private Text word=new Text();　　

private Text file=new Text();　　
@Override
　　

protected void setup(Context context) throws IOException, InterruptedException{　　
String fileName
=((FileSplit)context.getInputSplit())　　
.getPath().getName();
　　
file.
set(fileName);　　
}
　　

　　
@Override
　　

protected void map(LongWritable key,　　
Text value,Context context) throws IOException, InterruptedException{
　　
StringTokenizer words
=　　
new StringTokenizer(value.toString()," ");
　　
String item=null;
　　
while(words.hasMoreTokens()){
　　
item=words.nextToken();
　　
if(item.trim().length()>=1){
　　
word.set(item.trim());
　　
context.write(word,file);
　　
}
　　
}
　　
}
　　
}
　　

　　
static>　　
private Map<String,Integer> index=new HashMap<>();
　　
private Text value=new Text();
　　
@Override
　　
protected void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{
　　
index.clear();
　　
for(Text file:values){
　　
String fileName=file.toString();
　　
if(index.get(fileName)!=null){
　　
index.put(fileName,index.get(fileName)+1);
　　
}else{
　　
index.put(fileName,1);
　　
}
　　
}
　　
StringBuffer buffer=new StringBuffer();
　　
Set<Entry<String,Integer>> entries=index.entrySet();
　　
for(Entry<String,Integer> entry:entries){
　　
buffer.append(","+entry.getKey().toString()+":"+entry.getValue().toString());
　　
}
　　
this.value.set(buffer.toString());
　　
context.write(key,value);
　　
}
　　
}
　　

　　
@Override
　　
public int run(String[] args) throws Exception{
　　
Configuration conf=getConf();
　　
Path input=new Path(conf.get("input"));
　　
Path output=new Path(conf.get("output"));
　　

　　
Job job=Job.getInstance(conf,this.getClass().getSimpleName());
　　
job.setJarByClass(this.getClass());
　　

　　
job.setMapperClass(InvertIndexMapper.class);
　　
job.setMapOutputKeyClass(Text.class);
　　
job.setMapOutputValueClass(Text.class);
　　

　　
job.setReducerClass(InvertIndexReducer.class);
　　
job.setOutputKeyClass(Text.class);
　　
job.setOutputValueClass(Text.class);
　　

　　
job.setInputFormatClass(TextInputFormat.class);
　　
job.setOutputFormatClass(TextOutputFormat.class);
　　

　　
TextInputFormat.addInputPath(job,input);
　　
TextOutputFormat.setOutputPath(job,output);
　　

　　
return job.waitForCompletion(true)?0:1;
　　
}
　　

　　
public static void main(String[] args) throws Exception{
　　
System.exit(ToolRunner.run(new P00010_InvertIndex_0010(),args));
　　
}
　　
}
　　


InvertIndex_0010　　注意：
　　这里使用了一个StringTokenizer来分割数据：
　　

StringTokenizer words=　　
new StringTokenizer(value.toString()," ");
　　
String item=null;
　　
while(words.hasMoreTokens()){
　　
item=words.nextToken();
　　
if(item.trim().length()>=1){
　　
word.set(item.trim());
　　
context.write(word,file);
　　
}
　　
}
　　


五、共现矩阵（共现次数）　

5.1、需求分析　　　
　　首先我们要知道什么是共现次数？　
　　我们分析一个用户数据来解释：
　　

joe,    jon　　
joe , kia   
　　
joe, bob   
　　
joe ,ali
　　
kia,    joe   
　　
kia ,jim   
　　
kia, dee
　　
dee    ,kia   
　　
dee, ali
　　
ali    ,dee   
　　
ali, jim   
　　
ali ,bob   
　　
ali, joe   
　　
ali ,jon
　　
jon,    joe   
　　
jon ,ali
　　
bob,    joe   
　　
bob ,ali   
　　
bob, jim
　　
jim    ,kia   
　　
jim, bob   
　　
jim ,ali
　　

　　上面这个数据中，在一行中左边是一个用户，右边是它的好友。
　　那我们可以根据上面的数据列出所有用户的好友列表：
　　

ali,bob,jim,dee,jon,joe　　
bob,jim,ali,joe
　　
dee,kia,ali
　　
jim,ali,bob,kia
　　
joe,ali,bob,kia,jon
　　
jon,joe,ali
　　
kia,dee,jim,joe
　　

　　接下来我们把每个用户的好友列表每两两组成一对，在所有用户的好友列表中去计算，这两两组成的一对共出现了几次。
　　比如说：
　　bob,jim组成了一组，在余下的好友列表中两两组成去计算共出现了几次。（除了用户本身），也就是下面的数据。
　　

　　dee,jon,joe　　
　　jim,ali,joe
　　
　　kia,ali
　　
　　ali,bob,kia
　　
　　ali,bob,kia,jon
　　
　　joe,ali
　　
　　dee,jim,joe
　　

　　接下来就是jin,dee。然后是dee,jon依次类推。。。　　　　　　　　
　　从上面的分析我们可以得出预期的结果为：　　
　　

　　ali,bob    2　　
　　ali,jim   
1　　
　　ali,joe   
2　　
　　ali,jon   
1　　
　　ali,kia   
3　　
　　bob,dee   
1　　
　　bob,jim   
1　　
　　bob,joe   
1　　
　　bob,jon   
2　　
　　bob,kia   
2　　
　　dee,jim   
2　　
　　dee,joe   
2　　
　　dee,jon   
1　　
　　jim,joe   
3　　
　　jim,jon   
1　　
　　joe,jon   
1　　
　　jon,kia   
1　　

　　我们可以分两步去写，也就是分两个MapReduce任务。第一个MapReduce计算好友列表。第二个在每两两组成一组，计算这一组所出现的次数。

5.2、代码实现
　　1）计算好友列表


　　

import com.briup.bd1702.hadoop.mapred.utils.FriendRecordParser;　　
import java.io.IOException;
　　
import org.apache.hadoop.conf.Configuration;
　　
import org.apache.hadoop.conf.Configured;
　　
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.TextInputFormat;
　　
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
　　
import org.apache.hadoop.util.Tool;
　　
import org.apache.hadoop.util.ToolRunner;
　　

　　

public>
　　

static>
private Text userName=new Text();　　

private Text friendName=new Text();　　

private FriendRecordParser parser=new FriendRecordParser();　　
@Override
　　

protected void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException{　　
parser.parse(value);
　　

if(parser.isValid()){　　
userName.
set(parser.getUserName());　　
friendName.
set(parser.getFriendName());　　
context.write(userName,friendName);
　　
System.
out.println("----"+userName+"----"+friendName+"----");　　
}
　　
}
　　
}
　　

　　

static>
private Text friendsNames=new Text();　　
@Override
　　

protected void reduce(Text key,　　
Iterable
<Text> values,Context context) throws IOException, InterruptedException{　　
StringBuffer buffer
=new StringBuffer();　　

for(Text name:values){　　
buffer.append(
","+name);　　
}
　　
System.
out.println("++++"+buffer.toString()+"++++");　　
friendsNames.
set(buffer.toString());　　
context.write(key,friendsNames);
　　
}
　　
}
　　

　　
@Override
　　

public int run(String[] args) throws Exception{　　
Configuration conf
=getConf();　　
Path input
=new Path(conf.get("input"));　　
Path output
=new Path(conf.get("output"));　　

　　
Job job
=Job.getInstance(conf,this.getClass().getSimpleName());　　
job.setJarByClass(
this.getClass());　　

　　
job.setMapperClass(FriendListMapper.
class);　　
job.setMapOutputKeyClass(Text.
class);　　
job.setMapOutputValueClass(Text.
class);　　

　　
job.setReducerClass(FriendListReducer.
class);　　
job.setOutputKeyClass(Text.
class);　　
job.setOutputValueClass(Text.
class);　　

　　
job.setInputFormatClass(TextInputFormat.
class);　　
job.setOutputFormatClass(TextOutputFormat.
class);　　

　　
TextInputFormat.addInputPath(job,input);
　　
TextOutputFormat.setOutputPath(job,output);
　　

　　

return job.waitForCompletion(true)?0:1;　　
}
　　

　　

public static void main(String[] args) throws Exception{　　
System.exit(ToolRunner.run(
new P00010_FriendList_0010(),args));　　
}
　　
}
　　


FriendList　　2）计算共现次数

　　

import java.io.IOException;　　
import org.apache.hadoop.conf.Configuration;
　　
import org.apache.hadoop.conf.Configured;
　　
import org.apache.hadoop.fs.Path;
　　
import org.apache.hadoop.io.IntWritable;
　　
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.TextInputFormat;
　　
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
　　
import org.apache.hadoop.util.Tool;
　　
import org.apache.hadoop.util.ToolRunner;
　　

　　

public>
　　

static>
private Text key=new Text();　　

private IntWritable value=new IntWritable(1);　　
@Override
　　

protected void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException{　　
String[] strs
=value.toString().split(",");　　

for(int i=1;i<strs.length-1;i++){　　

for(int j=i+1;j<strs.length;j++){　　

//这个的目的是：两个数据形成一对之后，顺序固定的问题。　　
this.key.set(strs.compareTo(strs[j])<0?
　　
strs+","+strs[j]:
　　
strs[j]+","+strs);
　　
context.write(this.key,this.value);
　　
}
　　
}
　　
}
　　
}
　　


　　
static>　　
@Override
　　
protected void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException, InterruptedException{
　　
int count=0;
　　
for(IntWritable value:values){
　　
count+=value.get();
　　
}
　　
context.write(key,new IntWritable(count));
　　
}
　　
}
　　

　　
@Override
　　
public int run(String[] args) throws Exception{
　　
Configuration conf=getConf();
　　
Path input=new Path(conf.get("input"));
　　
Path output=new Path(conf.get("output"));
　　

　　
Job job=Job.getInstance(conf,this.getClass().getSimpleName());
　　
job.setJarByClass(this.getClass());
　　

　　
job.setMapperClass(CooccurenceMapper.class);
　　
job.setMapOutputKeyClass(Text.class);
　　
job.setMapOutputValueClass(IntWritable.class);
　　

　　
job.setReducerClass(CooccurenceReducer.class);
　　
job.setOutputKeyClass(Text.class);
　　
job.setOutputValueClass(IntWritable.class);
　　

　　
job.setInputFormatClass(TextInputFormat.class);
　　
job.setOutputFormatClass(TextOutputFormat.class);
　　

　　
TextInputFormat.addInputPath(job,input);
　　
TextOutputFormat.setOutputPath(job,output);
　　

　　
return job.waitForCompletion(true)?0:1;
　　
}
　　

　　
public static void main(String[] args) throws Exception{
　　
System.exit(ToolRunner.run(new P00020_Cooccurence_0010(),args));
　　
}
　　
}
　　


Cooccurence_0010　　喜欢就点个“推荐”哦~！　　　　

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表，该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有，请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意，请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息，一经发现立即删除。若您因此触犯法律，一切后果自负，我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集，我们仅提供一个展示、介绍、观摩学习的平台，我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责，亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏，不得用于商业或者其他用途，否则，一切后果均由您自己承担，我们对此不承担任何法律责任
7、如涉及侵犯版权等问题，请您及时通知我们，我们将立即采取措施予以解决
8、联系人Email：admin@iyunv.com 网址：www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址：https://www.yunweiku.com/thread-425059-1-1.html 上篇帖子： 马士兵hadoop第五课：java开发Map/Reduce 下篇帖子： Hadoop for .NET Developers
回复

使用道具 举报

返回列表
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群，最新一手资源尽在官方微信交流群！快快加入我们吧...

扫描微信二维码查看详情

客服 E-mail:iyunvcom@gmail.com

本站由青云提供云计算服务

运维网--中国最专业的运维工程师交流社区

豫ICP备20007574号-1 Copyright © 2012-2025

关于我们

发展历程

联系我们

关注我们

广告服务

广告案例

隐私条款

免责声明

用户登录

用户注册

版主守则

申请版主

手机版

金币获取

帮助中心

每日签到

Good Good

Study

Day Day

UP


客服E-mail:kefu@iyunv.com 客服QQ：1061981298


QQ群⑦：运维网交流群⑦ QQ群⑧：运维网交流群⑧ k8s群：运维网kubernetes交流群


提醒：禁止发布任何违反国家法律、法规的言论与图片等内容；本站内容均来自个人观点与网络等信息，非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来，其版权均归原作者及其网站所有，我们尊重他人的合法权益，如有内容侵犯您的合法权益，请及时与我们联系进行核实删除！



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表