实验环境：
操作系统：Linux（Centos7）；
Xsell 7
Hadoop版本：3.4.0(这里的版本根据自己的修改，可能小部分版本的Hadoop不适用于本文实验）
现在有多个输入文件，每个文件中的每行内容均为一个整数。要求读取所有文件中的整数，进行升序排序后，输出到一个新的文件中，输出的数据格式为每行两个整数，第一个数字为第二个整数的排序位次，第二个整数为原待排列的整数。下面是输入文件和输出文件的一个样例供参考。
输入文件1的样例如下：
33
37
12
40
输入文件2的样例如下：
4
16
39
5
输入文件3的样例如下：
1
45
25
根据输入文件1、2和3得到的输出文件如下：
1 1
2 4
3 5
4 12
5 16
6 25
7 33
8 37
9 39
10 40
11 45
具体步骤：
1、首先启动hadoop
start-all.sh2、新建文件夹以及相应的文件1、2和3
（若之前创建过，则可以跳过此步）
mkdir MapReduce进入该文件夹
cd MapReduce创建文件并存储
vi 1输入数据：
33
37
12
40
vi 2输入数据：
4
16
39
5
vi 3输入数据：
1
45
25
（切记要保存后退出！！！）
3、然后编码JAVA文件
vi MapReduce2.java编写内容如下：（记得修改连接）
import java.io.IOException;
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.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class MapReduce2{
public static class Map extends Mapper<Object, Text, IntWritable, IntWritable>{
private static IntWritable data = new IntWritable();
public void map(Object key, Text value, Context context) throws IOException,InterruptedException{
String text = value.toString();
data.set(Integer.parseInt(text));
context.write(data, new IntWritable(1));
}
}
public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
private static IntWritable line_num = new IntWritable(1);
public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException,InterruptedException{
for(IntWritable val : values){
context.write(line_num, key);
line_num = new IntWritable(line_num.get() + 1);
}
}
}
public static class Partition extends Partitioner<IntWritable, IntWritable>{
public int getPartition(IntWritable key, IntWritable value, int num_Partition){
int Maxnumber = 65223;
int bound = Maxnumber/num_Partition+1;
int keynumber = key.get();
for (int i = 0; i<num_Partition; i++){
if(keynumber<bound * (i+1) && keynumber>=bound * i){
return i;
}
}
return -1;
}
}
public static void main(String[] args) throws Exception{
Configuration conf = new Configuration();
conf.set("fs.default.name","hdfs://hadoop01:9000");//根据自己的对hadoop01:9000进行修改
String[] otherArgs = new String[]{"input","output"};
if (otherArgs.length != 2) {
System.err.println("Usage: wordcount <in><out>");
System.exit(2);
}
Job job = Job.getInstance(conf,"Merge and sort");
job.setJarByClass(MergeSort.class);
job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class);
job.setPartitionerClass(Partition.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}4、赋予权限
sudo chown -R root MapReduce这里的root为用户名，可以更改为你使用的用户名，后面的MapReduce则是存放文件的文件夹。
（前面做过可跳过）（为后面的创建目录做准备）
5、 配置环境
（前面修改过的话可以跳过）
//进入所修改的目标
vim ~/.bashrc
//所需要添加的内容
export HADOOP_HOME=/export/servers/hadoop //这里的路径改为你的hadoop安装目录即可
export CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath):$CLASSPATH
//使更改生效
source ~/.bashrc6、编译
（此时仍然在创建的MapReduce文件夹中）
javac MapReduce2.java7、打包为jar包
jar -cvf MapReduce2.jar *.class8、输入与输出
创建目录（需要与前面赋予权限的用户名相同，我前面是root，所以这里也是root）
这里我还在MapReduce文件夹中，所以用的这个，你也可以直接转到hadoop的bin目录下使用hdfs指令来完成。
/export/servers/hadoop/bin/hdfs dfs -mkdir -p /user/root先删除原来的input和output文件夹（如果原本没有可以跳过）
注意：一开始一定要没有，否则影响实验运行，导致最后生成jar包失败，即output文件夹内容错误或没有变化。
/export/servers/hadoop/bin/hdfs dfs -rm input/*/export/servers/hadoop/bin/hdfs dfs -rm -r output
创建input文件夹
/export/servers/hadoop/bin/hdfs dfs -mkdir input上传输入文件
/export/servers/hadoop/bin/hdfs dfs -put ./1 input
/export/servers/hadoop/bin/hdfs dfs -put ./2 input
/export/servers/hadoop/bin/hdfs dfs -put ./3 input使用生成的jar包
/export/servers/hadoop/bin/hadoop jar MapReduce2.jar MapReduce2查看输出
/export/servers/hadoop/bin/hdfs dfs -cat output/*成功输出：
