今天看Data-Intensive Text Processing with MapReduce 这本书的第三章的时候，里面有写到在map端优化wordcount。
对数据密集型数据进行分布式处理的时候，影响数据处理速度的非常重要的一个方面就是map的输出中间结果，在传送到reduce的过程中，很多的中间数据需要进行交换以及包括一些相应的处理，然后再交给相应的reduce。其中中间数据需要在网络中传输，另外中间数据在发送到网络上之前还要写到本地磁盘上，因为网络带宽和磁盘I/O是非常耗时的相比与其他的操作，所以减少中间数据的传输将会增加算法的执行效率，通过使用combiner函数或者其他的方式减少key-value对的个数。下面是一个改进的wordcount算法。
基本的思想是：
在map处理的时候定义一个关联数组，然后对文档进行处理，将<word，次数>加入到关联数组中，word存在，则将相应的次数加1，不存在则直接加入到关联数组中。所有的map任务结束后，然后再在run函数中输出处理结果。
伪代码：
class Mapper
method Map(docid a,doc d)
H =new AssociativeArray
for all term t 属于doc d do
H{t}=H{t}+1;
for all term t 属于 H do
EMIT(term t,count H{t})
class REDUCER
method REDUCE(term t,counts[c1,c2,...])
sum=0
for all count c 属于 counts[c1,c2,...] do
sum+=c
EMIT(term t,count sum)
代码如下：
import
java.io.IOException;
import
java.io.InputStream;
import
java.net.URI;
import
java.util.HashMap;
import
java.util.Iterator;
import
java.util.Map;
import
java.util.StringTokenizer;
import
java.util.Map.Entry;
import
org.apache.hadoop.conf.Configuration;
import
org.apache.hadoop.fs.FSDataInputStream;
import
org.apache.hadoop.fs.FileSystem;
import
org.apache.hadoop.io.IntWritable;
import
org.apache.hadoop.io.LongWritable;
import
org.apache.hadoop.io.Text;
import
org.apache.hadoop.mapreduce.Mapper.Context;
import
org.apache.hadoop.mapreduce.lib.input.FileSplit;
import
org.apache.hadoop.util.LineReader;
public
class
Mapper
extends
org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, IntWritable> {
int
c;
HashMap<String,IntWritable> map=
new
HashMap<String,IntWritable>();
@Override
protected
void
map(LongWritable key, Text value,
Context context)
throws
IOException, InterruptedException {
String str=value.toString();
StringTokenizer token=
new
StringTokenizer(str);
while
(token.hasMoreTokens()){
String value1=token.nextToken();
if
(map.containsKey(value1)){
//
System.out.println("ni");
int
p=map.get(value1).get()+1;
map.remove(value1);
map.put(value1,
new
IntWritable(p));
}
else
{
//
System.out.println("ni");
map.put(value1,
new
IntWritable(1));
}
}
//
TODO Auto-generated method stub
c++;
System.out.println(c);
}
@Override
protected
void
cleanup(org.apache.hadoop.mapreduce.Mapper.Context context)
throws
IOException, InterruptedException {
//
TODO Auto-generated method stub
System.out.println("cleanup");
super
.cleanup(context);
}
@Override
public
void
run(Context context)
throws
IOException, InterruptedException {
//
TODO Auto-generated method stub
super
.run(context);
System.out.println("run");
Iterator it=map.entrySet().iterator();
while
(it.hasNext()){
//
System.out.println("nihe");
Map.Entry<String, IntWritable> entry=(Map.Entry<String, IntWritable>) it.next();
//
System.out.println("nihe");
context.write(
new
Text(entry.getKey()), entry.getValue());
}
}
@Override
protected
void
setup(org.apache.hadoop.mapreduce.Mapper.Context context)
throws
IOException, InterruptedException {
//
TODO Auto-generated method stub
// System.out.println(context.getInputSplit().toString());
// System.out.println(context.getJobID());
// FileSplit input=(FileSplit)context.getInputSplit();
// String path=input.getPath().toString();
// Configuration conf=
new
Configuration();
// System.out.println(input.getPath().toString());
// FileSystem fs=FileSystem.get(URI.create(path), conf);
// FSDataInputStream filein=fs.open(input.getPath());
// LineReader in=
new
LineReader(filein,conf);
// Text line=
new
Text();
//
int
cd=in.readLine(line);
// System.out.println(line);
}
}
import
java.io.IOException;
import
org.apache.hadoop.io.IntWritable;
import
org.apache.hadoop.io.Text;
public
class
Reducer
extends
org.apache.hadoop.mapreduce.Reducer<Text, IntWritable, Text, IntWritable> {
@Override
protected
void
reduce(Text key, Iterable<IntWritable> values,
Context context)
throws
IOException, InterruptedException {
//
TODO Auto-generated method stub
int
sum=0;
for
(IntWritable it:values){
sum+=it.get();
}
context.write(key,
new
IntWritable(sum));
}
}
import
java.io.IOException;
import
java.net.URI;
import
org.apache.hadoop.conf.Configuration;
import
org.apache.hadoop.fs.FileSystem;
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
Word {
/**
*
@param
args
*
@throws
IOException
*
@throws
ClassNotFoundException
*
@throws
InterruptedException
*/
public
static
void
main(String[] args)
throws
IOException, InterruptedException, ClassNotFoundException {
//
TODO Auto-generated method stub
Job job=
new
Job();
Configuration conf=
new
Configuration();
Path in=
new
Path(args[0]);
Path out=
new
Path(args[1]);
FileSystem fs=FileSystem.get(URI.create(args[1]), conf);
fs.delete(out);
FileInputFormat.addInputPath(job, in);
FileOutputFormat.setOutputPath(job, out);
job.setMapperClass(Mapper.
class
);
job.setMapOutputKeyClass(Text.
class
);
job.setMapOutputValueClass(IntWritable.
class
);
job.waitForCompletion(
false
);
}
}
