hadoop实践总结

参考：http://wind-bell.javaeye.com/blog/81490  《hadoop入门转》设法找原文作者url没有找到。

         http://hi.baidu.com/atlahla/blog/item/4e8e37039f64a48dd43f7cfe.html 《在Linux下设置最简单的ssh安全自动登录》

概论
    作为Hadoop程序员，他要做的事情就是：
    1、定义Mapper，处理输入的Key-Value对，输出中间结果。
    2、定义Reducer，可选，对中间结果进行规约，输出最终结果。
    3、定义InputFormat 和OutputFormat，可选，InputFormat将每行输入文件的内容转换为Java类供Mapper函数使用，不定义时默认为String。
    4、定义main函数，在里面定义一个Job并运行它。
   

    然后的事情就交给系统了。
    1.基本概念：Hadoop的HDFS实现了google的GFS文件系统，NameNode作为文件系统的负责调度运行在master，DataNode运行在每个机器上。同时Hadoop实现了Google的MapReduce，JobTracker作为MapReduce的总调度运行在master，TaskTracker则运行在每个机器上执行Task。

    2.main()函数，创建JobConf，定义Mapper，Reducer，Input/OutputFormat 和输入输出文件目录，最后把Job提交給JobTracker，等待Job结束。

    3.JobTracker，创建一个InputFormat的实例，调用它的getSplits()方法，把输入目录的文件拆分成FileSplist作为Mapper task 的输入，生成Mapper task加入Queue。

    4.TaskTracker 向 JobTracker索求下一个Map/Reduce。
      
     Mapper Task先从InputFormat创建RecordReader，循环读入FileSplits的内容生成Key与Value，传给Mapper函数，处理完后中间结果写成SequenceFile.
     Reducer Task 从运行Mapper的TaskTracker的Jetty上使用http协议获取所需的中间内容（33%），Sort/Merge后（66%），执行Reducer函数，最后按照OutputFormat写入结果目录。

      TaskTracker 每10秒向JobTracker报告一次运行情况，每完成一个Task10秒后，就会向JobTracker索求下一个Task。

      Nutch项目的全部数据处理都构建在Hadoop之上，详见Scalable Computing with Hadoop。

    二、程序员编写的代码
    我们做一个简单的分布式的Grep，简单对输入文件进行逐行的正则匹配，如果符合就将该行打印到输出文件。因为是简单的全部输出，所以我们只要写Mapper函数，不用写Reducer函数，也不用定义Input/Output Format。

package  demo.hadoop

public   class  HadoopGrep {

  public   static   class  RegMapper  extends  MapReduceBase  implements  Mapper {

   private  Pattern pattern;

   public   void  configure(JobConf job) {
   pattern  =  Pattern.compile(job.get( " mapred.mapper.regex " ));
  }

   public   void  map(WritableComparable key, Writable value, OutputCollector output, Reporter reporter)
     throws  IOException {
   String text  =  ((Text) value).toString();
   Matcher matcher  =  pattern.matcher(text);
    if  (matcher.find()) {
    output.collect(key, value);
   }
  }
 }

  private  HadoopGrep () {
 }  //  singleton

  public   static   void  main(String[] args)  throws  Exception {
  
  JobConf grepJob  =   new  JobConf(HadoopGrep. class );
  grepJob.setJobName( " grep-search " );
  grepJob.set( " mapred.mapper.regex " , args[ 2 ]);

  grepJob.setInputPath( new  Path(args[ 0 ]));
  grepJob.setOutputPath( new  Path(args[ 1 ]));
  grepJob.setMapperClass(RegMapper. class );
  grepJob.setReducerClass(IdentityReducer. class );
      
  JobClient.runJob(grepJob);
 }
}

        RegMapper类的configure()函数接受由main函数传入的查找字符串，map() 函数进行正则匹配，key是行数，value是文件行的内容，符合的文件行放入中间结果。
        main()函数定义由命令行参数传入的输入输出目录和匹配字符串，Mapper函数为RegMapper类，Reduce函数是什么都不做，直接把中间结果输出到最终结果的的IdentityReducer类，运行Job。

        整个代码非常简单，丝毫没有分布式编程的任何细节。

三.运行Hadoop程序
3.1实践前准备：
 3.1.1.下载hadoop程序  用户制作的程序
 3.1.2.两台机器： 
           a.主控机器server 192.168.0.52

           b.客户机client  192.168.0.53

3.2.非集群方式：
      把hadoop程序 用户制作的程序 上传到linux机器上 使hadoop/bin目录下的脚本具有可执行权限
      完全不进行任何分布式计算，不动用任何namenode,datanode的做法，适合一开始做调试代码。
       解压hadoop，其中conf目录是配置目录，hadoop的配置文件在hadoop-default.xml，如果要修改配置，不是直接修改该文件，而是修改hadoop-site.xml，将该属性在hadoop-site.xml里重新赋值。
       hadoop-default.xml的默认配置已经是local运行，不用任何修改，配置目录里唯一必须修改的是hadoop-env.sh 里JAVA_HOME的位置。
       将编译好的HadoopGrep与RegMapper.class 放入hadoop/build/classes/demo/hadoop/目录 找一个比较大的log文件放入一个目录，然后运行
       hadoop / bin / hadoop demo.hadoop.HadoopGrep log文件所在目录 任意的输出目录 grep的字符串

     查看输出目录的结果，查看hadoop/logs/里的运行日志。  
     在重新运行前，先删掉输出目录。

3.3.单机集群：
  第1步.    然后修改hadoop-site.xml ，加入如下内容：
< property > 
   < name > fs.default.name </ name > 
   < value > server:9000 </ value >   //指定名字节点服务
</ property > 
< property > 
   < name > mapred.job.tracker </ name > 
   < value > server:9001 </ value > //指定合并节点服务
</ property > 
< property > 
   < name > dfs.replication </ name > 
   < value > 1 </ value > 
</ property >

    从此就将运行从local文件系统转向了hadoop的hdfs系统，mapreduce的jobtracker也从local的进程内操作变成了分布式的任务系统，9000，9001两个端口号是随便选择的两个空余端口号。  
  另外，如果你的/tmp目录不够大，可能还要修改hadoop.tmp.dir属性。
第2步. 增加ssh不输入密码即可登陆。
A为本地主机(即用于控制其他主机的机器) ;
B为远程主机(即被控制的机器Server), 假如ip为192.168.60.110;
A和B的系统都是Linux

在A上运行命令:
# ssh-keygen -t rsa (连续三次回车,即在本地生成了公钥和私钥,不设置密码)
# ssh root@192.168.60.110 "mkdir .ssh" (需要输入密码)
# scp ~/.ssh/id_rsa.pub root@192.168.60.110:.ssh/id_rsa.pub (需要输入密码)  //用这种方法实现把公匙复制被控制机器上

在B上的命令:
# touch /root/.ssh/authorized_keys (如果已经存在这个文件, 跳过这条)
# cat /root/.ssh/id_rsa.pub >> /root/.ssh/authorized_keys (将id_rsa.pub的内容追加到authorized_keys 中)

回到A机器:
# ssh -i 私鈅路径 root@192.168.60.110 (不需要密码, 登录成功)     //-i 参数指定本地密钥,如果私匙在.ssh中，且名称为id_rsa,则不用i指定密匙位置
经典举例：,

首先 ssh-keygen -t rsa 命令生成了一个密钥和一个公钥, 而且密钥可以设置自己的密码

可以把密钥理解成一把钥匙, 公钥理解成这把钥匙对应的锁头,

把锁头(公钥)放到想要控制的server上, 锁住server, 只有拥有钥匙(密钥)的人, 才能打开锁头, 进入server并控制

而对于拥有这把钥匙的人, 必需得知道钥匙本身的密码,才能使用这把钥匙 (除非这把钥匙没设置密码), 这样就可以防止钥匙被了配了(私钥被人复制)   [比较经典]
    
上面是一个通用规则：

  本机运行只需要把id_rsa.pub追加到authorized_keys中即可
这样使用ssh root@server即可以不用输入密码登录

  3.格式化namenode，执行
  bin/hadoop namenode -format

  4.启动Hadoop
     执行hadoop/bin/start-all.sh, 在本机启动namenode,datanode,jobtracker,tasktracker
  
  5.现在将待查找的log文件放入hdfs,。   //很重要 否则下一步会报找不到文件路径错误
     执行hadoop/bin/hadoop dfs 可以看到它所支持的文件操作指令。
     执行hadoop/bin/hadoop dfs put log文件所在目录 in ，则log文件目录已放入hdfs的/user/user-name/in 目录中

  6.现在来执行Grep操作
      hadoop/bin/hadoop demo.hadoop.HadoopGrep in out
      查看hadoop/logs/里的运行日志，重新执行前。运行hadoop/bin/hadoop dfs rmr out 删除out目录。
  7.把hdfs中的目的文件取到本地：
    执行hadoop/bin/hadoop dfs -get out 本地路径
  8.运行hadoop/bin/stop-all.sh 结束

3.4.多机集群：
 
  进行3.3的所有配置后，在执行以下步骤
  第一步：  把hadoop 和应用程序 以及配置文件 完全拷贝到要加入成为节点服务器的client机器  第二步： 修改server机器上的slaves（中文奴隶的意思）文件，这个文件中一行表示一个节点 把client  ip地址加入
          例如：
               localhost 或server或192.168.0.52
               client 或192.168.0.53
  第三步： 按照3.3中所讲的方法实现从server不用输入密码登录到client机器
   方法是：在server机器上运行命令

            scp /root/.ssh/id_rsa.pu root@client:.ssh
            让输入登录到clint密码,输入即可
          然后登录到client机器运行命令：
           cp /root/.ssh/id_rsa.pub /root/.ssh/authorized_keys
     即可
 第四步：在hadoop-server执行hadoop/bin/start-all.sh
   将会在hadoop-server启动namenode,datanode,jobtracker,tasktracker
   在hadoop-server2启动datanode 和tasktracker
  
 第五步.现在来执行Grep操作
     hadoop/bin/hadoop demo.hadoop.HadoopGrep in out
    重新执行前,运行hadoop/bin/hadoop dfs rmr out 删除out目录

 第六步.运行hadoop/bin/stop-all.sh 结束
              
提示：新加入一个台节点服务器，只需要执行第1.2.3步即可