我写了一封信 MapReduce program 分析 dataset 属于此窗体的用户数
UserID::Gender::Age::MoviesRated::Zip Code
1::F::1::10::48067
2::M::56::16::70072
3::M::25::15::55117我想
根据用户的平均年龄，按平均年龄的降序排列，找出排名前10位的密码。前10名是指该zipcode用户平均年龄最小的10岁。
我有一个 MapClass ，一个 CombinerClass 和一个 ReducerClass .
我的代码如下
public class TopTenYoungestAverageAgeRaters extends Configured implements Tool {
private static TreeSet<AverageAge> top10 = new TreeSet<AverageAge>();
public static class MapClass extends Mapper<LongWritable, Text, Text, AverageAge>
{
public boolean isNumeric(String value) // Checks if record is valid
{
try
{
Integer.parseInt(value);
return true;
}
catch(NumberFormatException e)
{
return false;
}
}
public AverageAge toCustomWritable(String[] line)
{
AverageAge record = new AverageAge(new IntWritable(Integer.parseInt(line[0])), new IntWritable(Integer.parseInt(line[2])), new Text(line[1]), new IntWritable(Integer.parseInt(line[3])), new Text(line[4]));
return record;
}
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
{
String line = value.toString();
String[] values = line.split("::");
if(isNumeric(values[0]))
{
AverageAge customTuple = toCustomWritable(values);
context.write(new Text(values[4]), customTuple);
}
}
}
public static class CombinerClass extends Reducer<Text, AverageAge, Text, AverageAge>
{
public void reduce(Text key, Iterable<AverageAge> values, Context context) throws IOException, InterruptedException
{
AverageAge newRecord = new AverageAge();
long age = 0;
int count = 0;
for(AverageAge value:values)
{
age += value.getUserAge();
count += 1;
}
newRecord.setZipCode(key.toString());
newRecord.setAverageAge((double)(age/count));
context.write(key, newRecord);
}
}
public static class ReducerClass extends Reducer<Text, AverageAge, NullWritable, AverageAge>
{
public void reduce(Text key, Iterable<AverageAge> values, Context context) throws IOException, InterruptedException
{
for(AverageAge value:values)
{
top10.add(value);
if(top10.size() > 10)
top10.remove(top10.last());
}
}
protected void cleanup(Context context) throws IOException, InterruptedException
{
for(AverageAge avg: top10)
{
context.write(NullWritable.get(), avg);
}
}
}
public static void main(String[] args) throws Exception {
// TODO Auto-generated method stub
int res = ToolRunner.run(new Configuration(), new TopTenYoungestAverageAgeRaters(), args);
System.exit(res);
}
@Override
public int run(String[] arg0) throws Exception {
// TODO Auto-generated method stub
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setMapperClass(MapClass.class);
job.setCombinerClass(CombinerClass.class);
job.setReducerClass(ReducerClass.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(AverageAge.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(AverageAge.class);
FileInputFormat.addInputPath(job, new Path(arg0[0]));
FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
return job.waitForCompletion(true) ? 0 : 1;
}
}这个 MapClass 以zipcode作为 key 以及 AverageAge （自定义可写类）如 value 这个 CombinerClass 计算属于该zipcode的用户的平均年龄并写入 key 作为zipcode和value as AverageAge .
这个 ReducerClass 给出（应该给出）具有平均用户年龄的前10个zipcodes，但我只得到一个记录作为输出。
我也试过了 System.out.println() 在reducer类中查看要传递给哪些值 ReducerClass 但上面什么也没印 console （我正在eclipse环境中本地运行程序）
我是mapreduce的新手，无法找出此程序中的错误。
数据集源
1条答案
按热度按时间8zzbczxx1#
问题陈述似乎自相矛盾：平均年龄下降的前10名是年龄最大的10名，而不是年龄最小的10名。最好澄清一下。
不管怎样，这里有很多很多错误。
组合器不能保证被调用
如果有多个reducer任务，那么在不同的文件中，每个任务最多可以有10个输出
如前所述，您将得到的“前10个”将是10个最低的邮政编码（按字典排序）。
通常由
cleanup()当你不再写记录的时候。您想要的是使用shuffle将具有相同zipcode的记录放在一起，并使用聚合类（combiner和reducer）来计算平均值。“前10个”的要求不能确定，直到你有一个年龄为每个zipcode。不过，关键的一点是，为了以分布式方式计算平均值，在减少之前，永远不能丢失分母。你舰队中的合路器可能会收到相同密钥的记录。
mapper获取一个记录并生成一个三元组：
combiner获取具有相同键的三元组的集合，并对它们求平均值（并对分母求和）：
reducer获取具有相同键的三元组的集合，并对它们进行平均，抛出分母：
无论你是否提供合路器，你的算法都应该有效；组合器是一种优化，仅适用于某些map-reduce情况。
要限制到前10名，你现在需要按平均年龄对结果重新排序。
这意味着另一个Map器：
一个只输出前10个结果的减速机（练习留给读者）。另外，只能有一个reduce任务，否则您将得到前10x，其中x是reduce任务的数量。