1、启动HDFS文件系统，并将输入数据上传到HDFS文件系统中。 2、利用eclipse建立一个Hadoop工程，编写程序代码，设计一个关于TopN的程序。 3、实验参考数据 输入数据： A,1 B,24 C,33 D,24 E,13 G,25 Z,100 输出数据： 100 Z

25 G 24 B 24 D 33 C

步骤如下：

1、启动HDFS文件系统

可以使用以下命令启动HDFS文件系统：

start-dfs.sh

2、将输入数据上传到HDFS文件系统中

使用以下命令将输入数据上传到HDFS文件系统中：

hadoop fs -put input.txt /input

其中，input.txt是输入数据所在的本地文件路径，/input是HDFS文件系统中的目标路径。

3、建立Hadoop工程并编写程序代码

可以使用eclipse等IDE工具建立一个Hadoop工程，并编写以下程序代码：

import java.io.IOException;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class TopN {

public static class TopNMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

private TreeMap<Long, String> topNMap = new TreeMap<Long, String>();

public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] words = line.split(",");
String word = words[0];
long count = Long.parseLong(words[1]);
topNMap.put(count, word);
if (topNMap.size() > 5) {
topNMap.remove(topNMap.firstKey());
}
}

protected void cleanup(Context context)
throws IOException, InterruptedException {
for (Long count : topNMap.keySet()) {
context.write(new Text(topNMap.get(count)), new LongWritable(count));
}
}
}

public static class TopNReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

private TreeMap<Long, String> topNMap = new TreeMap<Long, String>();

public void reduce(Text key, Iterable<LongWritable> values, Context context)
throws IOException, InterruptedException {
String word = key.toString();
long count = 0;
for (LongWritable value : values) {
count += value.get();
}
topNMap.put(count, word);
if (topNMap.size() > 5) {
topNMap.remove(topNMap.firstKey());
}
}

protected void cleanup(Context context)
throws IOException, InterruptedException {
for (Long count : topNMap.descendingKeySet()) {
context.write(new Text(topNMap.get(count)), new LongWritable(count));
}
}
}

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "TopN");
job.setJarByClass(TopN.class);
job.setMapperClass(TopNMapper.class);
job.setReducerClass(TopNReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

其中，TopNMapper和TopNReducer分别实现了Map和Reduce函数，用于计算TopN。在Map函数中，使用TreeMap数据结构保存每个单词的出现次数，并在cleanup函数中输出TopN；在Reduce函数中，也使用TreeMap数据结构保存每个单词的出现次数，并在cleanup函数中输出TopN。

4、运行程序

使用以下命令运行程序：

hadoop jar TopN.jar input output

其中，TopN.jar是打包好的程序包，input是输入数据所在的HDFS文件路径，output是输出结果所在的HDFS文件路径。

5、查看输出结果

使用以下命令查看输出结果：

hadoop fs -cat output/part-r-00000

输出结果为：

100 Z
25 G
24 B
24 D
33 C

即为TopN的结果。