利用MapReduce分析微博粉丝数

利用MapReduce分析微博粉丝数。

import java.io.IOException;

import java.util.HashMap;

import java.util.Map;

import java.util.Set;

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.Mapper;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.filecache.DistributedCache;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Weibo {

public static class weiboMapper extends Mapper{

@Override

protected void map(Object key, Text value, Mapper.Context context)

throws IOException, InterruptedException {

// TODO Auto-generated method stub

String[] strings = value.toString().split("\t");

if (strings.length == 5) {

String name = strings[0];

int fans = Integer.parseInt(strings[2]);

context.write(new Text(name),new IntWritable(fans));

}

}

}

public static class weiboReducer extends Reducer{

Map map = new HashMap<>();

@Override

protected void reduce(Text name, Iterable fans,

Reducer.Context context) throws IOException, InterruptedException {

// TODO Auto-generated method stub

for (IntWritable intWritable : fans) {

map.put(name.toString(), Integer.parseInt(intWritable.toString()));

}

}

@Override

protected void cleanup(Reducer.Context context)

throws IOException, InterruptedException {

// TODO Auto-generated method stub

Set set = map.keySet();

for(int i =0;i<5;i++){

int a = 0;

String nString = "";

for (String name : set) {

if (map.get(name)>a) {

nString = name;

a = map.get(name);

}

}

context.write(new Text(nString), new IntWritable(a));

map.remove(nString);

}

}

}

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

Job job = Job.getInstance(new Configuration());

job.setJarByClass(Weibo.class);

job.setMapperClass(weiboMapper.class);

job.setReducerClass(weiboReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job, new Path(args[1]));

job.waitForCompletion(true);

}

}