使用 Hadoop Mapreduce 进行数据处理

1. 综述

　　使用HDP（下载: http://zh.hortonworks.com/products/releases/hdp-2-3/#install）搭建环境，进行分布式数据处理。

　　项目文件下载，解压文件后将看到项目文件夹。该程序将读取 cloudMR/internal_use/tmp/dataset/titles 目录下的四个文本文件， 文件中的每一行文本都是来自于 wikipedia 的一个标题， 读取每个标题，并使用 cloudMR/internal_use/tmp/dataset/misc/delimiters.txt 中指定的特殊符号分割标题成独立单词，然后将单词转换为全小写，然后将出现在
cloudMR/internal_use/tmp/dataset/misc/stopwords.txt 中的单词全部删除，最后统计剩余单词的出现次数，并输出。

　　程序的编译过程需要在“~/.bashrc”文件内定义自己的环境变量“$hadoop_CLASSPATH”，在“~/.bashrc”文件中添加一行：　　

export hadoop_CLASSPATH="/usr/hdp/2.3.2.0-2950/hadoop/conf:/usr/hdp/2.3.2.0-2950/hadoop/conf:/usr/hdp/2.3.2.0-2950/hadoop/conf:/usr/hdp/2.3.2.0-2950/hadoop/lib/*:/usr/hdp/2.3.2.0-2950/hadoop/.//*:/usr/hdp/2.3.2.0-2950/hadoop-hdfs/./:/usr/hdp/2.3.2.0-2950/hadoop-hdfs/lib/*:/usr/hdp/2.3.2.0-2950/hadoop-hdfs/.//*:/usr/hdp/2.3.2.0-2950/hadoop-yarn/lib/*:/usr/hdp/2.3.2.0-2950/hadoop-yarn/.//*:/usr/hdp/2.3.2.0-2950/hadoop-mapreduce/lib/*:/usr/hdp/2.3.2.0-2950/hadoop-mapreduce/.//*:::/usr/share/java/mysql-connector-java-5.1.17.jar:/usr/share/java/mysql-connector-java-5.1.31-bin.jar:/usr/share/java/mysql-connector-java.jar:/usr/hdp/2.3.2.0-2950/tez/*:/usr/hdp/2.3.2.0-2950/tez/lib/*:/usr/hdp/2.3.2.0-2950/tez/conf:/usr/hdp/current/hadoop-yarn-client/.//*:/usr/hdp/current/hadoop-yarn-client/lib/*"

  

 

2. 运行过程

Step (1): 将项目文件夹放入HDP虚拟机，进入cloudMR文件夹，运行下列命令启动:

./start.sh

要求输入账号，随意输入10位数字即可。再运行下列命令检查 hadoop 是否正常运行：

hadoop version

 

Step (2): 编写 TitleCount.java 文件，完成相应功能。完成后的 TitleCount.java 如下：

 

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.*;
import java.util.*;
/**
 * Classic "Word Count"
 */
public class TitleCount extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new TitleCount(), args);
        System.exit(res);
    }

    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(this.getConf(), "Title Count");
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setMapperClass(TitleCountMap.class);
        job.setReducerClass(TitleCountReduce.class);

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

        job.setJarByClass(TitleCount.class);
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static String readHDFSFile(String path, Configuration conf) throws IOException{
        Path pt=new Path(path);
        FileSystem fs = FileSystem.get(pt.toUri(), conf);
        FSDataInputStream file = fs.open(pt);
        BufferedReader buffIn=new BufferedReader(new InputStreamReader(file));

        StringBuilder everything = new StringBuilder();
        String line;
        while( (line = buffIn.readLine()) != null) {
            everything.append(line);
            everything.append("\n");
        }
        return everything.toString();
    }
    
    public static class TitleCountMap extends Mapper<Object, Text, Text, IntWritable> {
        Set<String> stopWords = new HashSet<String>();
        String delimiters;

        @Override
        protected void setup(Context context) throws IOException,InterruptedException {

            Configuration conf = context.getConfiguration();
            
            String delimitersPath = conf.get("delimiters");
            delimiters = readHDFSFile(delimitersPath, conf);
            
            String stopWordsPath = conf.get("stopwords");
            List<String> stopWordsList = Arrays.asList(readHDFSFile(stopWordsPath, conf).split("\n"));
            for(String e : stopWordsList){
                stopWords.add(e);
            }
        }

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer stk = new StringTokenizer(value.toString(),delimiters);
            while(stk.hasMoreTokens()){
                String e = stk.nextToken().trim().toLowerCase();
                if(stopWords.contains(e) == false){
                    context.write(new Text(e),new IntWritable(1));
                }
            }
        }
    }

    public static class TitleCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for(IntWritable e : values){
                sum += e.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }
}

 

Step (3): 编译java源文件。为了方便，在cloudMR文件夹中新建output文件夹，用来存放编译生成的.class文件。使用以下命令（在cloudMR文件夹下执行）：

mkdir output

javac -classpath $Hadoop_CLASSPATH -d output TitleCount.java

进入output文件夹会看到3个.class文件。

 

Step (4): 将编译生成的类文件打包。

首先在 cloudMR 文件夹下新建文本文件 manifest.mf，使用以下命令（在 cloudMR 文件夹下执行）：

touch manifest.mf

编辑内容为

Main-Class: TitleCount.class

manifest.mf 内是一些关于这个包的信息，这里定义了主类。

再使用下面命令打包（在cloudMR文件夹下执行）：

jar cvfM TitleCount.jar manifest.mf -C output/ .

这条命令的含义是;

jar　　　　　　　　打包命令

cvfM　　　　　　　

TitleCount.jar　　 打成的包的名字

manifest.mf　　   将这个文件打进包里

-C　　　　　　　　-C之后的文件夹内的所有文件打进包里

output/　　　　　 将output文件夹内的文件全部打进包里

.　　　　　　　　  打成的包TitleCount.jar放在当前文件夹

注意：打包过程很重要而且易错，请务必按照上文所述步骤进行。

 

Step (5): 将 TitleCount.jar 发布。

在发布（yarn）之前，还要完成准备工作。

将相关文件：cloudMR/internal_use/tmp/dataset/titles 目录下的四个文本文件、cloudMR/internal_use/tmp/dataset/misc/delimiters.txt 、
cloudMR/internal_use/tmp/dataset/misc/stopwords.txt 上传到hdfs.

在 hdfs 的 /user/root/ 文件夹内新建 data 文件夹，将 delimiters.txt、stopwords.txt 放入 data 文件夹，再在 data 文件夹中新建 titles 文件夹，将cloudMR/internal_use/tmp/dataset/titles 目录下的四个文本文件放入 titles 文件夹。

介绍相关的命令：

hadoop fs -ls　　　　　　                                   列出hdfs目录，由于没有参数，列出的是当前用户的主目录

hadoop fs -ls /　　　　                                    　列出hdfs根目录

hadoop fs -mkdir data                                       在默认目录下新建data目录

hadoop fs -mkdir data/titles                               在data目录中新建 titles目录

hadoop fs -copyFromLocal ./abc.txt  data           上传当前目录（本地）中的 abc.txt 到 hdfs 上的 data 目录

　　之后便可以发布了，使用命令：　　

yarn jar TitleCount.jar TitleCount -D delimiters="/user/root/data/delimiters.txt" -D stopwords="/user/root/data/stopwords.txt" data/titles output

这条命令的含义是;

yarn                              发布内容

jar                                要发布的内容为jar包

TitleCount.jar                发布的内容

TitleCount 　　              TitleCount.jar的入口

-D delimiters="/user/root/data/delimiters.txt" -D stopwords="/user/root/data/stopwords.txt" 　　　　　　-D后跟参数，这里定义了两个参数

data/titles                     输入文件夹，其内的文件作为Map 的输入

output                          输出文件存放的位置

　　yarn 命令执行完毕后，即可查看运行结果。