云计算与海量数据管理实验

　　　　　export JAVA_HOME=/home/hadoop/app/jdk
　　　　　export JRE_HOME=$JAVA_HOME/jre
　　　　　export CLASSPATH=.:$CLASSPATH:$JAVA_HOME/lib:$JRE_HOME/lib
　　　　　export PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin

这段内容可以复制到文件末尾，注意修改红字部分

　　（2）编译配置文件

　　　　source ~/.bashrc

　　（3）测试jdk

　　　　java -vers ion

八、安装hadoop

　　1、安装hadoop

　　　tar -zxvf /mnt/hgfs/vmshare/hadoop-2.10.1.tar.gz -C ~/app //解压

　　　mv ~/app/hadoop-2.10.1 ~/app/hadoop //改名 (注意文件名匹配)

sudo chown -R hadoop ./hadoop //修改hadoop权限

　　2、配置hadoop

　　　　（1）配置环境变量

　　　　　　vim ~/.bashrc，添加或修改以下语句：

export HADOOP_HOME=/app/hadoop

export PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

　　　　　　source ~/.bashrc //编译生效

　　　　（2）配置hadoop-env.sh

　　　　　　vim ~/app/hadoop/etc/hadoop/hadoop-env.sh

　　　　　　找到export JAVA_HOME=${JAVA_HOME}这一行，将其修改为：
　　　　　　export JAVA_HOME=/home/hadoop/app/jdk

　　　　（3）配置core–site.xml (均位于~/app/hadoop/etc/hadoop/，下同)　　

　　　　　　<configuration>
　　　　　　　　<property>
　　　　　　　　　　<name>fs.defaultFS</name>
　　　　　　　　　　<value>hdfs://local host:9000</value>
　　　　　　　　</property>
　　　　　　　　<property>
　　　　　　　　　　<name>hadoop.tmp.dir</name>
　　　　　　　　　　<value>/home/hadoop/app/hadoop/tmp</value>

　　　　　　　　</property>
　　　　　　</configuration>

　　　　（4）配置hdfs–site.xml　　　　

　　　　　　　　　　<!–配置块的副本数 –>
　　　　　　　　　　<name>dfs.repli cation</name>
　　　　　　　　　　<value>1</value>
　　　　　　　　</property>
　　　　　　</configuration>

　　　　（5）配置map red–site.xml

　　　　　　说明：需将map red–site.xml.template复制成map red–site.xml

　　　　　　<configuration>
　　　　　　　　<property>
　　　　　　　　　　<name>map reduce.framework.name</name>
　　　　　　　　　　<val ue>yarn</val ue>
　　　　　　　　</property>
　　　　　　</configuration>

　　　　（6）配置yarn–site.xml

　　　　　　　　<property>
　　　　　　　　　　<name>yarn.node manager.aux-services</name>
　　　　　　　　　　<value>map reduce_shuffle</value>
　　　　　　　　</property>
　　　　　　</configuration>

九、运行hadoop

　　1、格式化namenode

　　　　cd ~/app/hadoop/bin

./hadoop namenode –format

　　　　说明：首次运行须格式化，以后运行就不需要这一步。格式化后会在core–site.xml中hadoop.tmp.dir所指定的目录中生成name文件夹。

　　2、运行hadoop (要在用户模式下)

　　　　cd ~/app/hadoop/sbin

　　　　./start-all.sh

　　3、查看进程：

　　　　jps　　　　　

　　　　　　14262 NameNode
　　　　　　15976 NodeManager
　　　　　　14385 DataNode
　　　　　　15884 ResourceManager
　　　　　　14601 Seconda ryNameNode
　　　　　　16120 Jps
　　　　如果出现以上进程（前面的数字不一定一致），说明hadoop环境搭建成功，分布式系统启动完成。　　

　　4、webUI查看（在虚拟机中的网络浏览器查看）

　　　　http://localhost:50070 //查看 hdfs

　　　　http://localhost:8088 //查看map reduce作业情况

十、测试 word count 程序

　　1、生成单词文件

　　　　mkdir ~/tmp
　　　　echo ‘hello world hello hadoop’ > ~/tmp/word1.txt
　　　　echo ‘hive run on hadoop’ > ~/tmp/word2.txt　　

　　2、上传至hdfs

　　　　（1）在hdfs上新建输入数据的目录(在~/app/hadoop/bin目录下)
　　　　　　./hdfs dfs -mkdir /input
　　　　（2）上传文件

　　　　　　./hdfs dfs -put ~/tmp/word*.txt /input

　　3、运行hadoop自带的单词计数程序　　

　　　　./hadoop jar ~/app/hadoop/share/hadoop/map reduce/hadoop-map reduce-examples–2.7.3.jar wordcount /input output

注：这里的红字部分可能需要修改，这时你需要去对应的路径下查看文件名，按实际情况进行修改，比如我在做这一步骤的时候，就将2.7.3修改成了2.10.1

4、查看结果

./hdfs dfs –cat /user/hadoop/output/part-r-00000

　　　　可以看到每个单词出现的次数

十一、实现并测试矩阵相乘程序（必做）

由于伪分布式环境下，节点性能有限，尤其是跑在虚拟机环境下，试了一下跑500*500的矩阵，Map到66%左右就发现虚拟机硬盘爆了，所以，为了测试算法的正确性，下面使用50*50的矩阵来运行。只要算法是恰当的，数据规模可以随硬件支持程度而改变。

参考博客 https://blog.csdn.net/catharryy/article/details/9186757

若对具体实现思路有兴趣，可以看看上面这篇文章，这里只说要实际操作的部分。

1、在~/app/hadoop目录下新建一个名为 local_matrix 的文件夹

2、在 local_matrix 文件夹下新建一个名为 MartrixMultip li cation 的java文件，

参考 https://github.com/intergret/snippet/blob/master/MartrixMultiplication.java

文件内容如下（在链接所示的代码基础上略微改动）：

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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 MartrixMultiplication{

  public static class MartrixMapper extends Mapper<Object, Text, Text, Text>{
    
    private Text map_key = new Text();
    private Text map_value = new Text();
    
    int rNumber = 50;
    int cNumber = 50;
    String fileTarget;
    String i, j, k, ij, jk;
    
      
    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
      
        String eachterm[] = value.toString().split("#");
        fileTarget = eachterm[0];
        
        if(fileTarget.equals("M")){
          i = eachterm[1];
          j = eachterm[2];
          ij = eachterm[3];
            
          for(int c = 1; c<=cNumber; c++){
              map_key.set(i + "#" + String.valueOf(c));
              map_value.set("M" + "#" + j + "#" + ij);
              context.write(map_key, map_value);
          }
            
        }else if(fileTarget.equals("N")){
          j = eachterm[1];
          k = eachterm[2];
          jk = eachterm[3];

          for(int r = 1; r<=rNumber; r++){
              map_key.set(String.valueOf(r) + "#" +k);
              map_value.set("N" + "#" + j + "#" + jk);
              context.write(map_key, map_value);
          }
        }
    }

  } 
  
  
  public static class MartrixReducer extends Reducer<Text,Text,Text,Text> {
    
    private Text reduce_value = new Text();
    
    int jNumber = 50;

    int M_ij[] = new int[jNumber+1];
    int N_jk[] = new int[jNumber+1];

    int j, ij, jk;

    String fileTarget;
    int jsum = 0;
    
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
      
      jsum = 0; 

      for (Text val : values) {
        String eachterm[] = val.toString().split("#");
        
        fileTarget = eachterm[0];
        j = Integer.parseInt(eachterm[1]);
        
        if(fileTarget.equals("M")){
      	  ij = Integer.parseInt(eachterm[2]);
      	  M_ij[j] = ij;
        }else if(fileTarget.equals("N")){
      	  jk = Integer.parseInt(eachterm[2]);
      	  N_jk[j] = jk;
        }
        
      }
      
      for(int d = 1; d<=jNumber; d++){
    	 jsum +=  M_ij[d] * N_jk[d];
      }
      
      reduce_value.set(String.valueOf(jsum));
      context.write(key, reduce_value); 
    }

  }
  

  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: MartrixMultiplication <in> <out>");
	  System.exit(2);
      }
	    
      Job job = new Job(conf, "martrixmultiplication");
      job.setJarByClass(MartrixMultiplication.class);
      job.setMapperClass(MartrixMapper.class);
      job.setReducerClass(MartrixReducer.class);
	    
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);
	    
      FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
      FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
	    
      System.exit(job.waitForCompletion(true) ? 0 : 1);
	    		    
  }
  
}