从零开始学习Hadoop

1. Hadoop从头说
1.1 Google是一家做搜索的公司
    做搜索是技术难度很高的活。首先要存储很多的数据，要把全球的大部分网页都抓下来，可想而知存储量有多大。然后，要能快速检索网页，用户输入几个关键词找资料，越快越好，最好在一秒之内出结果。如果全球每秒有上亿个用户在检索，只有一两秒的检索时间，要在全球的网页里找到最合适的检索结果，难度很大。
    Google用三个最重要的核心技术解决上述问题，它们分别是GFS, MapReduce和BigTable。Google发表了它们的设计论文，但没有将它们开源，核心竞争力不可能开源的。论文在这里，有兴趣的同学可以去看看：GFS，http://labs.google.com/papers/gfs-sosp2003.pdf；MapReduce， http://labs.google.com/papers/mapreduce-osdi04.pdf
；Bigtable，http://labs.google.com/papers/bigtable-osdi06.pdf。
    Google的论文发表之后，Doug Cutting等人根据论文的思想，在开源项目Nutch的基础上实现了Hadoop。后来，Doug Cutting去了Yahoo，继续做Hadoop。后来，Hadoop的开发和应用开始爆发了。
    在对应关系上看，Hadoop MapReduce对应MapReduce，Hadoop Distributed File System (HDFS)对应GFS，HBase对应BigTable。一般我们所说的Hadoop其实是指Hadoop体系，它包括Hadoop MapReduce，HDFS，HBase，还有其他更多的技术。

















1.2 MapReduce和HDFS是如何工作的
    先用一种有助于理解的方式描述MapReduce和HDFS是如何工作的。假如有1000G的多个文本文件，内容是英文网页，需要统计词频，也就是哪些单词出现过，各出现过多少次，有1000台计算机可供使用，要求速度越快越好。最直接的想法是，把1000G的文件分成1000份，每台机器处理1G数据。处理完之后，其他999台机器将处理结果发送到一台固定的机器上，由这台机器进行合并然后输出结果。
    Hadoop将这个过程进行自动化的处理。首先看如何存储这1000G的文本文件。HDFS在这1000台机器上创建分布式文件系统，将1000G的文件切分成若干个固定大小的文件块，每个块一般是64M大小，分散存储在这1000台机器上。这么多机器，在运行的时候难免会出现有几台突然死机或者挂掉的情况，这导致上面存储的文件块丢失，会导致计算出错。为避免这种情况，HDFS对每个文件块都做复制，复制成3～5个相同的块，放到不同的机器上，这样死机的文件块在其他机器上仍然可以找得到，不影响计算。
    MapReduce其实是两部分，先是Map过程，然后是Reduce过程。从词频计算来说，假设某个文件块里的一行文字是”This is a small cat. That is a small dog.”，那么，Map过程会对这一行进行处理，将每个单词从句子解析出来，依次生成形如<“this”, 1>, <”is”, 1>, <”a”, 1>, <”small”, 1>, <”cat”, 1>, <”that”, 1>, <”is”, 1>, <”a”, 1>, <”small”, 1>, <”dog”, 1>的键值对，<”this”, 1>表示“this”这个单词出现了1次，在每个键值对里，单词出现的次数都是1次，允许有相同的键值对多次出现，比如<”is”,1>这个键值对出现了2次。Reduce过程就是合并同类项，将上述产生的相同的键值对合并起来，将这些单词出现的次数累加起来，计算结果就是<“this”, 1>, <”is”, 2>, <”a”, 2>, <”small”, 2>, <”cat”, 1>, <”that”, 1>, <”dog”, 1>。这种方式很简洁，并且可以进行多种形式的优化。比如说，在一个机器上，对本地存储的1G的文件块先Map，然后再Reduce，那么就得到了这1G的词频统计结果，然后再将这个结果传送到远程机器，跟其他999台机器的统计结果再次进行Reduce，就得到1000G文件的全部词频统计结果。如果文件没有那么大，只有三四个G，就不需要在本地进行Reduce了，每次Map之后直接将结果传送到远程机器做Reduce。
    具体地，如果用Hadoop来做词频统计，流程是这样的：
        1)先用HDFS的命令行工具，将1000G的文件复制到HDFS上；
        2)用Java写MapReduce代码，写完后调试编译，然后打包成Jar包；
        3)执行Hadoop命令，用这个Jar包在Hadoop集群上处理1000G的文件，然后将结果文件存放到指定的目录。
        4)用HDFS的命令行工具查看处理结果文件。
1.3 API参考
    开发过程需要的API全部在Java API和Hadoop API，在下面两个地方找：
    Hadoop 1.2.1的API文档：http://hadoop./docs/r1.2.1/api/index.html
    Java JDK1.7的API文档：http://docs.oracle.com/javase/7/docs/api/
2. 词频统计
    在这里，我们开始实现Word Count的MapReduce。这里的Word Count程序是从Hadoop的例子代码改编来的。
3. 标准形式的MapReduce程序
    所谓标准形式的MapReduce，就说需要写MapReduce的时候，脑海里立刻跳出的就是这个形式，一个Map的Java文件，一个Reduce的Java文件，一个负责调用的主程序Java文件。这个标准形式已经是最简了，没有多余的东东可以删除，没有肥肉，是干货。写MapReduce和主程序的时候，分别引用哪些包哪些类，每个包每个类是什么作用，这些要很清晰。如果记不住的话，将这些代码写几遍，编译调试运行，然后不看代码，自己从头写出来，编译调试运行，重复多次应该可以记住了。
3.1 目录和文件结构
    首先创建一个目录wordcount_01存放源代码、编译和打包结果，比如将这个目录放在/home/brian/wordcount_01。
    wordcount_01目录下，有两个子目录，分别是src目录和classes目录。src目录存放Java的源代码，classes目录存放编译结果。在src目录下，创建三个文件，分别是IntSumReducer.java，TokenizerMapper.java，WordCount.java。从MapReduce的角度看， TokenizerMapper.java文件是做Map的代码，IntSumReducer.java是做Reduce的代码， WordCount.java是主程序，负责执行整个流程。这三个Java文件内容在下面给出。
3.2 TokenizerMapper .java文件的源代码
package com.brianchen.hadoop;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TokenizerMapper
      extends Mapper<Object, Text, Text, IntWritable>{
  IntWritable one = new IntWritable(1);
  Text word = new Text();
  
  public void map(Object key, Text value, Context context)
      throws IOException, InterruptedException {
    StringTokenizer itr = new StringTokenizer(value.toString());
    while (itr.hasMoreTokens()){
      word.set(itr.nextToken());
      context.write(word, one);
    }
  }
}
    
    下面逐行解释代码，所有的类更详细的资料其实都可以在1.3节的两个API地址里找到：
    1）“package com.brianchen.hadoop”
    Java提供包机制管理代码，关键词就是package，可以随意指定一个包的名字，诸如笔者的就是”com.brianchen.hadoop”，只要不跟其他的包重复就可以。为了保证包的唯一性，Sun公司推荐用公司的域名的逆序作为包名，于是大家就在代码里看到诸如”org.apache.hadoop”之类的包名。
    2)”import java.io.IOException”
    凡是以java开头的包，在JDK1.7的API里找类的资料。这一句从java的io包里导入IOException。IOException，输入输出异常类。所谓异常，就是 Exception，就是程序出错了，异常机制是Java的错误捕获机制。那么，IOException就是处理输入输出错误时候的异常，I是Input，O是Output。
    3) “import java.util.StringTokenizer”
     从java的util包引入StringTokenizer类。StringTokenizer将符合一定格式的字符串拆分开。比如说，”This is a cat”是一个字符串，这四个单词是用空格符隔开的，那么StringTokenizer可以将它们拆成四个单词”This”， “is”，”a”，“cat”。如果是用其他符号隔开，也能处理，比如”14;229;37”这个字符串，这三个数字是分号隔开的，StringTokenizer将它们拆成”14”，“229”，“37”。只要指定了分隔符，StringTokenizer就可以将字符串拆开。“拆开”的术语叫“解析”。
    4)”import org.apache.hadoop.io.IntWritable”
    凡是以org.apache.hadoop 开头的包，在Hadoop 1.2.1的API找类的详细信息。从hadoop的io包里引入IntWritable类。IntWritable类表示的是一个整数，是一个以类表示的整数，是一个以类表示的可序列化的整数。在Java里，要表示一个整数，假如是15，可以用int类型，int类型是Java的基本类型，占4个字节，也可以用Integer类，Integer类封装了一个int类型，让整数成为类。Integer类是可以序列化的。但Hadoop觉得Java的序列化不适合自己，于是实现了IntWritable类。至于什么是序列化，这个问题比较长，这个问题会在后面章节详细讲。
    5) “import org.apache.hadoop.io.Text”
    从hadoop的io包里引入Text类。Text类是存储字符串的可比较可序列化类。
    6) “import org.apache.hadoop.mapreduce.Mapper”
    Mapper类很重要，它将输入键值对映射到输出键值对，也就是MapReduce里的Map过程。
    7)”public class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>”
    定义一个自己的Map过程，类名是TokenizerMapper，它继承了Hadoop的Mapper类。“<Object, Text, Text, IntWritable>”，这里，第一个参数类型是Object，表示输入键key的参数类型，第二个参数参数类型是Text，表示输入值的类型，第三个参数类型也是Text，表示输出键类型，第四个参数类型是IntWritable，表示输出值类型。
    在这个例子里，第一个参数Object是Hadoop根据默认值生成的，一般是文件块里的一行文字的行偏移数，这些偏移数不重要，在处理时候一般用不上，第二个参数类型是要处理的字符串，形如”This is a cat.”。经过Map处理之后，输出的就是诸如<”This”, 1>的键值对，这个”This”就是第三个参数类型，是Text类型，而1就是第四个参数类型，是IntWritable。
    8)“IntWritable one = new IntWritable(1)”
    定义输出值，始终是1。
    9) “Text word = new Text()”
    定义输出键。
    10) “public void map(Object key, Text value, Context context) throws IOException, InterruptedException ”
    定义map函数，函数有三个参数，key是输入键，它是什么无所谓，实际上用不到它的，value是输入值。在map函数中，出错的时候会抛出异常，所以有“throws IOException, InterruptedException ”。至于Context类，这个类的定义是在TokenizerMapper的祖先类Mapper的内部，不需要引入，如果去查看Mapper类的源代码的话，能看到Context类是继承MapContext类的。
    11) “StringTokenizer itr = new StringTokenizer(value.toString())”
    定义StringTokenizer对象itr， StringTokenizer的构造函数只接受Java的String类，而value是Text类，所以要进行转化，将value转成String类，也就是“value.toString()”。
    12)Map过程
        while (itr.hasMoreTokens()){
          word.set(itr.nextToken());
          context.write(word, one);
        }
    在默认的情况下，StringTokenizer以空格符作为分隔符对字符串进行解析，每次解析会先调用hasMoreTokens看看是不是需要做解析，如果需要做，就用nextToken()函数获取解析结果，然后用这个结果给word赋值，然后，再将word和one作为一个键值对写到context里，context会存储键值留待Reduce过程处理。
3.3 IntSumReducer.java文件的源代码


package com.brianchen.hadoop;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class IntSumReducer extends
      Reducer<Text,IntWritable,Text,IntWritable> {
  IntWritable result = new IntWritable();

  public void reduce(Text key, Iterable<IntWritable> values, Context context)
      throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable val : values) {
      sum += val.get();
    }
    result.set(sum);
    context.write(key, result);
  }
}

    跟上节相同的地方就不解释了，只解释上节没有的东东。
    1)”import org.apache.hadoop.mapreduce.Reducer”
    引入hadoop的Reducer类，这个类负责MapReduce的Reduce过程。
    2) “public class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> “
    定义Reduce过程，也就是IntSumReducer类，这个类继承Hadoop的Reducer类。这里的”<Text,IntWritable,Text,IntWritable>”，含义跟上一节一样，依次分别是输入键类型，输入值类型，输出键类型，输出值类型。
    3)“IntWritable result = new IntWritable()”
    定义输出结果，这是一个整数。
    4) “public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException ”
    定义reduce函数。key是输入键类型，values是一个实现了Iterable接口的变量，可以把它理解成values里包含若干个IntWritable整数，可以通过迭代的方式遍历所有的值，至于Context类型，跟Mapper里的Context类似的方式，是在Redurer类内部实现的。
    举例来说，假如处理一个字符串”This is a That is a“，那么，经过Map过程之后，到达reduce函数的时候，依次传递给reduce函数的是：key=”This”，values=<1>；key = “is”，values=<1, 1>；key = “a”, values=<1, 1>；key=”That”, values=<1>。注意，在key = “is”和key=”a”的时候，values里有两个1。
    5) Reduce过程
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        result.set(sum);
        context.write(key, result);
    这个过程，就是用一个循环，不断从values里取值，然后累加计算和，循环结束后，将累加和赋值给result变量，然后，将键值和累加和作为一个键值对写入context。继续以上一步的例子来说，写入context的键值对依次就是<”This”，1>，<“is”，2>，<“a”, 2>，<”That”, 1>。
3.4 WordCount.java文件的源代码
package com.brianchen.hadoop;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length != 2) {
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);
    }

    Job job = new Job(conf, "wordcount");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.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);
  }
}
    1)”import org.apache.hadoop.conf.Configuration”
    Configuration类，顾名思义，读写和保存各种配置资源。
    2) “import org.apache.hadoop.fs.Path”
    引入Path类，Path类保存文件或者目录的路径字符串。
    3) “import org.apache.hadoop.mapreduce.Job”
    引入Job类。在hadoop里，每个需要执行的任务是一个Job，这个Job负责很多事情，包括参数配置，设置MapReduce细节，提交到Hadoop集群，执行控制，查询执行状态，等等。
    4)”import org.apache.hadoop.mapreduce.lib.input.FileInputFormat”
    引入FileInputFormat类。这个类的很重要的作用就是将文件进行切分split，因为只有切分才可以并行处理。这个会在后面章节有详细解释。
    5) “import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat”
    引入 FileOutputFormat类，处理结果写入输出文件。
    6) “import org.apache.hadoop.util.GenericOptionsParser”
    引入 GenericOptionsParser类，这个类负责解析hadoop的命令行参数。
    7)”public class WordCount ”
    这是word count主类，它负责读取命令行参数，配置Job，调用Mapper和Reducer，返回结果等等工作。
    8)“Configuration conf = new Configuration()”
    默认情况下，Configuration开始实例化的时候，会从Hadoop的配置文件里读取参数。
    9)”String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs()”
    读取参数分两步，上一步是从Hadoop的配置文件读取参数，这一步是从命令行参数读取参数，args是存放命令行参数的字符串数组。
    10) “if (otherArgs.length != 2) ”
    如果命令行参数不是2个，就出错了，退出。因为程序需要知道处理的是哪个输入文件，处理结果放到哪个目录，必须是两个参数。
    11)” Job job = new Job(conf, "wordcount")”
    每个运行的处理任务就是一个Job，”worodcount”是Job的名字。
    12) “ job.setJarByClass(WordCount.class)”
    Jar文件是Java语言的一个功能，可以将所有的类文件打包成一个Jar文件，setJarByClass的意思是，根据WordCount类的位置设置Jar文件。
    13) “job.setMapperClass(TokenizerMapper.class)”
    设置Mapper。
    14) “job.setReducerClass(IntSumReducer.class)”
    设置Reducer。
    15) “job.setOutputKeyClass(Text.class)”
    设置输出键的类型。
    16)“job.setOutputValueClass(IntWritable.class)”
    设置输出值的类型。
    17) “FileInputFormat.addInputPath(job, new Path(otherArgs[0]))”
    设置要处理的文件，也就是输入文件，它是otherArgs的第一个参数。
    18) “FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]))”
    设置输出文件，将处理结果写到这个文件里，它是otherArgs的第二个参数。
    19) “System.exit(job.waitForCompletion(true) ? 0 : 1)”
    最后一步，job开始执行，等待执行结束。
3.5 编译
    用javac编译项目。javac即Java programming language compiler，是Java JDK的命令行编译器。如前所说，wordcount_01目录存放源代码和编译结果，要在这个目录下进行编译。
    3.5.1 “cd ~/wordcount_01”
    先执行这个命令，切换目录到wordcount_01下。
    3.5.2 “javac -classpath /home/brian/usr/hadoop/hadoop-1.2.1/hadoop-core-1.2.1.jar:/home/brian/usr/hadoop/hadoop-1.2.1/lib/commons-cli-1.2.jar -d ./classes/ ./src/*.java”
    执行这条命令，编译源代码。-classpath，设置源代码里使用的各种类的库文件路径，路径之间用”:”隔开，-d参数，设置编译后的class文件存在路径。
3.6 打包
    3.6.1 “jar -cvf wordcount.jar -C ./classes/ .”
    将编译好的class文件打包成Jar包，jar命令是JDK的打包命令行工具，跟tar非常像。在命令里，-C是值在执行jar的时候将目录切换到当前目录下的classes目录，这个目录包含编译好的class文件。打包结果是wordcount.jar文件，放在当前目录下。
3.7 执行
    3.7.1 首先要确实一下Hadoop已经运行起来了。启动方式就是第1章的第7节。然后，执行
    3.7.2 “cd ~/usr/hadoop/hadoop-1.2.1”
    切换目录到Hadoop的安装目录下。
    3.7.3 “./bin/hadoop fs -put READER.txt readme.txt”
    仍然用README.txt做测试，将它复制到HDFS上，更名为readme.txt
    3.7.4 “./bin/hadoop fs -rmr output”
    处理结果要放在HDFS的output目录里，如果这个目录已经存在了，Hadoop是不会运行的，会报错，先删除它。
    3.7.5 “./bin/hadoop jar /home/brian/wordcount_01/wordcount.jar com.brianchen.hadoop.WordCount readme.txt output”
    运行程序，处理readme.txt文件，将结果写入output目录，其中”jar”参数是指定jar包的位置，而”com.brianchen.hadoop.WordCount”，这里”com.brianchen.hadoop”是包的名字， “WordCount”是主类，注意，如果不写包名字会报错的，必须有包名。
3.8 查看结果
    3.8.1 “./bin/hadoop fs -cat output/part-r-00000”
    处理结果output目录的 part-r-00000文件里，用cat命令可以输出到屏幕显示。
4. 最简形式的MapReduce
    最简单形式的Word Count的MapReduce代码是Hadoop自带的例子，略作改动放在这里。这个例子只有一个Java文件，Mapper和Reducer都写在WordCount类的内部。
4.1 目录和文件结构
    代码放在~/wordcount_02目录，它有两个子目录，分别是classes和src，classes目录存放编译结果，src目录存放源代码，src目录下只有一个java文件，即”WordCount.java”，所有的代码都在里面。
4.2 WordCount.java文件的源代码
package com.brianchen.hadoop;

import java.io.IOException;
import java.util.StringTokenizer;

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.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 WordCount {
  public static class TokenizerMapper
        extends Mapper<Object, Text, Text, IntWritable>{
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context)
        throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()){
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

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

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length != 2) {
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);
    }

    Job job = new Job(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.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);
  }
}

    这里的代码，跟前一节有点不太一样。
    1)”public static class TokenizerMapper”
    这表示TokenizerMapper类是WordCount类的内部静态类，这种方式可以将TokenizerMapper隐藏在WordCount类内部，且TokenizerMapper类不引用WordCount类的任何变量和函数。
    2) “private final static IntWritable one”
    跟上一节的定义相比，这里多了”private final static”，”private”表示这个变量是类的私有变量，“final”表示这变量只能在定义的时候被赋值一次，以后不可更改，”static”表示这是一个静态变量，独立于对象，被该类的所有实例共享，这种做法的好处是，one这个值是私有的不可更改的仅仅只有一个，代码更可靠，更节省内存空间。
4.3 编译
    4.3.1 “cd ~/wordcount_02”
    4.3.2 “javac -classpath /home/brian/usr/hadoop/hadoop-1.2.1/hadoop-core-1.2.1.jar:/home/brian/usr/hadoop/hadoop-1.2.1/lib/commons-cli-1.2.jar -d ./classes/ ./src/WordCount.java ”
4.4 打包
    4.4.1 “jar -cvf wordcount.jar -C ./classes/ . ”
4.5 运行
    4.5.1 “cd ~/usr/bin/hadoop/hadoop-1.2.1”
    4.5.2 “./bin/hadoop fs -rmr output”
    4.5.3 “./bin/hadoop jar /home/brian/wordcount_02/wordcount.jar com.brianchen.hadoop.WordCount readme.txt output”
4.6 查看结果
    4.6.1 “./bin/hadoop fs -cat output/part-r-00000”

举报