Java通过Hadoop平台使用 MapReduce 实现数据全局排序

本次实验，在 Hadoop 平台上，使用 MapReduce 实现了数据的全局排序。本文将详细阐述实现所需环境及过程。

• 分布式系统与实践 - 使用 MapReduce 实现数据全局排序
  • 环境配置
    • Install JDK
    • Install Hadoop
    • 配置伪 Hadoop 集群
      • 检查无密码 ssh localhost
      • Hadoop xml 配置
      • 执行官方 Demo 测试
  • 具体实现
    • 需求
    • 数据
    • 读取数据并排序
    • 人工指定分区
    • 使用 TotalOrderPartitioner 完成分区
    • 打包并执行

使用阿里云服务器安装， OS: Ubuntu20.04 LTS . 本来尝试使用 WSL2，尝试无果。

首先安装 JDK.

sudo apt-get install openjdk-11-jdk

下面根据 Hadoop 的 官方文档 来安装。

首先检查有没有 ssh 和 pdsh, 没有的话依次安装。

sudo apt-get install ssh
sudo apt-get install pdsh

然后前往 Apache Download Mirrors 寻找合适的 Hadoop 版本，然后用 wget 下载.

# 下载链接根据自己的情况替换
wget https://dlcdn.apache.org/hadoop/common/current/hadoop-3.3.1.tar.gz

解压之。

tar -zxvf hadoop-3.3.1.tar.gz

配置环境变量.

# 编辑这个文件，并追加
vim etc/hadoop/hadoop-env.sh

# 路径根据自己的安装路径设置，一般 openjdk 都在 /usr/lib/jvm/
export JAVA_HOME=/usr/lib/jvm/java-1.11.0-openjdk-amd64
export HDFS_NAMENODE_USER=root
export HDFS_DATANODE_USER=root
export HDFS_SECONDARYNAMENODE_USER=root
export YARN_RESOURCEMANAGER_USER=root
export YARN_NODEMANAGER_USER=root
# 根据自己文件夹的位置配置
export HADOOP_HOME=~/hadoop-3.3.1
export PATH=$PATH:$HADOOP_HOME/bin
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

尝试运行。

bin/hadoop

如果显示了 Hadoop 的使用文档，就安装成功了。

Hadoop 支持用三种模式启动：单机模式、伪分布式模式、分布式集群模式。集群模式我们没有足够的机器，单机模式很难体现出分布式的优势，所以我们选用伪分布式。
Hadoop 可以在单节点上以所谓的伪分布式模式运行，此时每一个 Hadoop 守护进程都作为一个独立的 Java 进程运行。
下面开始配置。

检查自己能否无密码 ssh localhost. 如果不行的话，执行。

ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa
cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys
chmod 0600 ~/.ssh/authorized_keys
sudo service ssh restart

vim etc/hadoop/core-site.xml:


    
        fs.defaultFS
        hdfs://localhost:9000
    

vim etc/hadoop/hdfs-site.xml:


    
        dfs.replication
        1
    

注意不能使用 zsh，如果你跟我一样用的是 zsh，将默认 shell 改回 bash.

chsh -s /bin/bash

下面跑一下官方给的 WordCount Demo 测试安装是否成功。

1. 初始化分布式文件系统

   bin/hdfs namenode -format
   

2. 启动 NameNode 和 DataNode

   sbin/start-dfs.sh
   

   此时若遇到报错 connection refused, 如下图。

   如果确认 ssh localhost 可以成功，那么一般是 pdsh 造成的。在 stackoverflow 上找到这样一个解决办法。

   编辑文件 libexec/hadoop-functions.sh，将这一行：

   if [[ -e '/usr/bin/pdsh' ]]; then
   

   替换为：

   if [[ ! -e '/usr/bin/pdsh' ]]; then
   

   再重新尝试。答主给了两种解决方案，此种方案解决了我的问题。

3. 浏览 NameNode 的 Web 界面，默认地址：

   http://localhost:9870/
   

   效果见下图。

   此时 NameNode 启动成功。

4. 配置 HDFS 目录

   bin/hdfs dfs -mkdir /user
   bin/hdfs dfs -mkdir /user/user # 此处是你执行 NameNode 的 username
   

5. 将 Input 文件复制到 HDFS 中

   事先准备一个单词文件，例如 input/abc.txt. 将其复制到 HDFS 中。

   bin/hdfs dfs -mkdir wc_input
   bin/hdfs dfs -put input/abc.txt wc_input
   

6. 执行 Hadoop WordCount 程序

   bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.1.jar wordcount wc_input wc_output2
   

7. 查看结果

   bin/hdfs dfs -cat wc_output2
   public class TotalSortMap extends Mapper {
       @Override
       protected void map(Text key, Text value,
                          Context context) throws IOException, InterruptedException {
           context.write(key, new IntWritable(Integer.parseInt(key.toString())));
       }
   }
   
   
   public class TotalSortReduce extends Reducer {
       @Override
       protected void reduce(Text key, Iterable values,
                             Context context) throws IOException, InterruptedException {
           for (IntWritable value : values)
               context.write(value, NullWritable.get());
       }
   }
   
   
   public class TotalSort extends Configured implements Tool {
   
       //实现一个Kye比较器，用于比较两个key的大小，将key由字符串转化为Integer，然后进行比较。
       public static class KeyComparator extends WritableComparator {
           protected KeyComparator() {
               super(Text.class, true);
           }
   
           @Override
           public int compare(WritableComparable writableComparable1, WritableComparable writableComparable2) {
               int num1 = Integer.parseInt(writableComparable1.toString());
               int num2 = Integer.parseInt(writableComparable2.toString());
   
               return num1 - num2;
           }
       }
       @Override
       public int run(String[] args) throws Exception {
           Configuration conf = new Configuration();
           conf.set("mapreduce.totalorderpartitioner.naturalorder", "false");
           Job job = Job.getInstance(conf, "Total Sort app");
           job.setJarByClass(TotalSort.class);
   
           //设置读取文件的路径，都是从 HDFS 中读取。读取文件路径从脚本文件中传进来
           FileInputFormat.addInputPath(job,new Path(args[0]));
           //设置mapreduce程序的输出路径，MapReduce的结果都是输入到文件中
           FileOutputFormat.setOutputPath(job,new Path(args[1]));
           job.setInputFormatClass(KeyValueTextInputFormat.class);
           //设置比较器，用于比较数据的大小，然后按顺序排序，该例子主要用于比较两个 key 的大小
           job.setSortComparatorClass(KeyComparator.class);
           job.setNumReduceTasks(3);//设置 reduce 数量
   
           job.setMapOutputKeyClass(Text.class);
           job.setMapOutputValueClass(IntWritable.class);
           job.setOutputKeyClass(IntWritable.class);
           job.setOutputValueClass(NullWritable.class);
   
           //设置保存partitions文件的路径
           TotalOrderPartitioner.setPartitionFile(job.getConfiguration(), new Path(args[2]));
           //key值采样，0.01是采样率，
           InputSampler.Sampler sampler = new InputSampler.RandomSampler<>(0.01, 1000, 100);
           //将采样数据写入到分区文件中
           InputSampler.writePartitionFile(job, sampler);
   
           job.setMapperClass(TotalSortMap.class);
           job.setReducerClass(TotalSortReduce.class);
           //设置分区类。
           job.setPartitionerClass(TotalOrderPartitioner.class);
           return job.waitForCompletion(true) ? 0 : 1;
       }
       public static void main(String[] args)throws Exception{
   
           int exitCode = ToolRunner.run(new TotalSort(), args);
           System.exit(exitCode);
       }
   }
   

   打包并执行

   通过 Maven 将我们写好的 MapReduce 函数打成 Jar 包。首先配置 main/java/META-INF/MANIFEST.MF.

   Manifest-Version: 1.0
   Main-Class: TotalSort
   

   通过 Maven 打包。

   mvn clean
   mvn package
   

   上传到服务器，并准备好测试数据，根据前面的步骤，将数据上传到 HDFS。

   • 输入数据路径 - hdfs://localhost:9000/user/root/sort-in
   • 输出路径 - hdfs://localhost:9000/user/root/sort-out
   • 分区路径 - hdfs://localhost:9000/user/root/total_sort_partitions

   执行我们写好的 MapReduce 函数。

   bin/hadoop jar globalsort.jar sort-in sort-out total_sort_partitions
   

   检查排序结果。

   bin/hdfs dfs -cat sort-out/*
   
   # 文件太长了，只显示最后十行
   
   bin/hdfs dfs -cat sort-out/* | tail -n 10
   

   部分结果如下图所示。可见数据排序完毕。